talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Begin implementing AsyncSocket & host Rocket homepage again

Browse Source
This commit is contained in:
Alex Hultman
2018-09-12 01:56:25 +02:00
parent ee2756c880
commit 80e64b59fb
5 changed files with 114 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.cpp
+13 -15
View File
@@ -25,7 +25,7 @@ std::string_view getFile(std::string_view file) {
if (it == cache.end()) {
std::cout << "Did not have file: " << file << std::endl;
std::ifstream fin("rocket_files/" + std::string(file), std::ios::binary);
std::ifstream fin(std::string(file), std::ios::binary);
std::ostringstream oss;
oss << fin.rdbuf();
@@ -59,21 +59,19 @@ int main(int argc, char **argv) {
uWS::HttpContext<false> *httpContext = uWS::HttpContext<false>::create(loop.loop);
// req, res?
httpContext->onGet("/", [](auto *res, auto *req) { // maybe use the terminology of HttpRequest for both?
/*std::cout << "URL: <" << req->getUrl() << ">" << std::endl;
std::cout << "Query: <" << req->getQuery() << ">" << std::endl;
std::cout << "User-Agent: <" << req->getHeader("user-agent") << ">" << std::endl;
*/
// read some data being passed
/*res->read([](std::string_view chunk) {
std::cout << "Reading some streamed in data:" << chunk << std::endl;
});*/
res->writeStatus(uWS::HTTP_200_OK)->write([](int offset) {
return std::string_view("Hello world!").substr(offset);
}, 12);
httpContext->onGet("/:folder/:file", [](auto *res, auto *req) {
// what file are we serving?
std::string_view fileName = req->getUrl();
if (fileName == "/") {
// this is our index
fileName = "/rocket_files/rocket.html";
}
// load the file from cache and stream it as response
std::string_view file = getFile(fileName.substr(1));
res->writeStatus(uWS::HTTP_200_OK)->write([file](int offset) {
return file.substr(offset);
}, file.length());
});
httpContext->onGet("/yolo", [](auto *res, auto *req) {
src/new_design/AsyncSocket.h
+81 -62
View File
@@ -41,52 +41,110 @@ protected:
}
public:
void *getExt() {
return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
}
// cork should bascially mark corked
// if already corked, crash and burn!
/* Cork this socket. Only one socket may ever be corked per-loop at any given time */
void cork() {
LoopData *loopData = getLoopData();
loopData->corked = true;
}
// if not in corked mode, write & buffer. if in corked mode: either write & buffer or buffer
// that is, corking is optional
void write(const char *src, int length) {
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer */
int write(const char *src, int length, bool optionally = false, int nextLength = 0) {
LoopData *loopData = getLoopData();
// append it
memcpy(loopData->corkBuffer + loopData->corkOffset, src, length);
loopData->corkOffset += length;
if (length == 0) {
return 0;
}
if (loopData->corked) {
// prio 1: write to cork
if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) {
memcpy(loopData->corkBuffer + loopData->corkOffset, src, length);
loopData->corkOffset += length;
} else {
/* Strategy differences between SSL and non-SSL */
if constexpr(SSL) {
/* For SSL we do not want to emit small chunks, so fill the cork (assuming the cork is about 16k) */
int remainingCork = LoopData::CORK_BUFFER_SIZE - loopData->corkOffset;
memcpy(loopData->corkBuffer + loopData->corkOffset, src, remainingCork);
loopData->corkOffset += remainingCork;
uncork(src + remainingCork, length - remainingCork);
} else {
/* For non-SSL we take the penalty of two syscalls */
uncork(src, length);
}
}
} else {
// not corked
// we should not write here if if have buffer!
int written = 0;
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
if (!asyncSocketData->buffer.length()) {
written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, src, length, nextLength != 0);
}
if (written < length) {
// bail out if we can
if (optionally) {
return written;
}
// shit, we are fucked
std::cout << "Buffering up per-socket" << std::endl;
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
// at least reserve enough for next failure
if (nextLength) {
asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + length + nextLength);
}
asyncSocketData->buffer.append(src, length);
}
}
return length;
}
// should follow same rules as for write
/* Write an unsigned 32-bit integer */
void writeUnsigned(unsigned int value) {
LoopData *loopData = getLoopData();
loopData->corkOffset += u32toa(value, loopData->corkBuffer + loopData->corkOffset);
char buf[10];
int length = u32toa(value, buf);
// for now we do this copy
write(buf, length);
}
// uncork should always send and clean the loop's cork buffer. anything that is not able to send, buffer it up in the socket
void uncork() {
/* Uncork this socket. It is essential to remember doing this. */
void uncork(const char *src = nullptr, int length = 0) {
LoopData *loopData = getLoopData();
loopData->corked = false;
if (loopData->corked) {
loopData->corked = false;
// send it off now!
int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, loopData->corkBuffer, loopData->corkOffset, false);
loopData->corkOffset = 0;
// buffer the rest up in the outbuffer of this asynsocket!
if (loopData->corkOffset) {
write(loopData->corkBuffer, loopData->corkOffset, false, length);
write(src, length, false, 0);
loopData->corkOffset = 0;
}
}
}
// when we are writable AND have buffer length, that's a really bad situation that should not happen!
void drain(std::string_view optionalChunk) {
/* Drain any socket-buffer while also optionally sending a chunk */
void mergeDrain(std::string_view optionalChunk) {
std::cout << "mergeDrain" << std::endl;
// the question here is: should we recursively copy things to the cork buffer and try and send them off in one go?
// it would work in a recursively manner and since optionalChunk is optional, it would never increase the buffer size
@@ -96,45 +154,6 @@ public:
// the cunk is optional meaning we do not care to buffer it up
}
// this one will write up to length and simply leave things be
int writeOptionally(const char *src, int length) {
// not optional for now
AsyncSocket<SSL>::write(src, length);
return length;
/*
// kopiera upp till (SSL eller icke-ssl) max copy distance
// om mer än detta, fortsätt skicka
// this entire strategy should be made entirely in AsyncSocket!
// this strategy can be simplified to one, we can even have MAX_COPY_DISTANCE_SSL and MAX_COPY_DISTANCE
if (length < LoopData::MAX_COPY_DISTANCE) {
AsyncSocket<SSL>::write("Content-Length: ", 16);
AsyncSocket<SSL>::writeUnsigned(chunk.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
AsyncSocket<SSL>::write(chunk.data(), chunk.length());
} else {
// copying some data with the headers is a good idea for SSL but probably not for non-SSL
//writeToCorkBufferAndReset(chunk.data(), LoopData::MAX_COPY_DISTANCE, length, true);
// just assume this went fine
HttpResponseData<SSL> *httpData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// write that off! (should never happen here!)
//static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, chunk.data() + LoopData::MAX_COPY_DISTANCE, chunk.length() - LoopData::MAX_COPY_DISTANCE, 0);
// if offset is at the end, we are done
if (httpData->offset < length) {
httpData->outStream = cb;
}
}*/
}
void close() {
static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
}
src/new_design/HttpContext.h
+17 -7
View File
@@ -75,6 +75,9 @@ private:
// cork this socket (move this to loop?)
((AsyncSocket<SSL> *) s)->cork();
// pass this pointer to pointer along with the routing and change it if upgraded
SOCKET_TYPE *returnedSocket = s;
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) {
@@ -92,12 +95,11 @@ private:
httpResponseData->inStream(data);
}
}, [](void *user) {
std::cout << "INVALID HTTP!" << std::endl;
// close it down
// close any socket on HTTP errors
//static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) user);
});
// uncork
// uncork only if not closed
((AsyncSocket<SSL> *) s)->uncork();
// how do we return a new socket here, from the http route?
@@ -115,10 +117,18 @@ private:
// get next chunk to send
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
std::string_view chunk = httpResponseData->outStream(httpResponseData->offset);
// send, including any buffered up
asyncSocket->drain(chunk);
if (httpResponseData->outStream) {
std::string_view chunk = httpResponseData->outStream(httpResponseData->offset);
// send, including any buffered up
asyncSocket->mergeDrain(chunk);
} else {
std::cout << "We did not have any outStream!" << std::endl;
asyncSocket->mergeDrain(std::string_view(nullptr, 0));
}
return s;
});
src/new_design/HttpResponse.h
+2 -2
View File
@@ -19,7 +19,6 @@ private:
}
public:
/* Write the HTTP status */
HttpResponse *writeStatus(std::string_view status) {
AsyncSocket<SSL>::write("HTTP/1.1 ", 9);
@@ -43,7 +42,8 @@ public:
AsyncSocket<SSL>::write("Content-Length: ", 16);
AsyncSocket<SSL>::writeUnsigned(chunk.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
if (AsyncSocket<SSL>::writeOptionally(chunk.data(), chunk.length()) < length) {
if (AsyncSocket<SSL>::write(chunk.data(), chunk.length(), true) < length) {
std::cout << "HttpResponse::write failed to write everything" << std::endl;
getHttpResponseData()->outStream = cb;
}
}
src/new_design/LoopData.h
+1
View File
@@ -7,6 +7,7 @@ private:
public:
/* Good 16k for SSL perf. */
static const int CORK_BUFFER_SIZE = 16 * 1024;
char *corkBuffer = new char[CORK_BUFFER_SIZE];