From 7f3753e9db5798709e1b61b4ab0bcf86c714d6de Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sun, 23 Sep 2018 18:00:18 +0200 Subject: [PATCH] Various comments and fixes --- main.cpp | 12 +++-- src/AsyncSocket.h | 15 ++++-- src/HttpContext.h | 115 +++++++++++++++++++++++++++++---------------- src/HttpResponse.h | 26 ++++++---- 4 files changed, 111 insertions(+), 57 deletions(-) diff --git a/main.cpp b/main.cpp index f0bd86e..313b17a 100644 --- a/main.cpp +++ b/main.cpp @@ -47,8 +47,6 @@ int main(int argc, char **argv) { for (auto *x : delayedResponses) { std::cout << "Resuming a response now!" << std::endl; - - // resume should take a string_view! x->resume(); } @@ -76,10 +74,16 @@ int main(int argc, char **argv) { /* This route streams back chunks of data in delayed fashion */ res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) { + /* Handle aborted stream out */ + if (offset == -1) { + std::cout << "Removing closed stream!" << std::endl; + delayedResponses.erase(res); + return uWS::HTTP_STREAM_FIN; + } + + /* Delay stream out */ std::cout << "Delaying stream now" << std::endl; delayedResponses.insert(res); - - //asyncFetchData(res) return uWS::HTTP_STREAM_PAUSE; }, 100); diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 2e0d141..5cbd3d6 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -148,7 +148,7 @@ public: } /* Drain any socket-buffer while also optionally sending a chunk */ - int mergeDrain(std::string_view optionalChunk) { + int mergeDrain(std::string_view optionalChunk = {}) { // strategy: if we have two parts and both will fit in cork buffer then cork them and recursively send them off @@ -167,8 +167,17 @@ public: return write(optionalChunk.data(), optionalChunk.length(), true, 0); } - void close() { - static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); + /* These should not be public to the user! */ + void timeout(unsigned int seconds) { + static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds); + } + + void shutdown() { + static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this); + } + + SOCKET_TYPE *close() { + return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); } }; diff --git a/src/HttpContext.h b/src/HttpContext.h index 45d195c..52cb70f 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -44,16 +44,12 @@ private: /* Init the HttpContext by registering libusockets event handlers */ HttpContext *init() { - //new (data = (Data *) static_dispatch(us_ssl_socket_context_ext, us_socket_context_ext)(httpServerContext)) Data(); - /* Handle socket connections */ static_dispatch(us_ssl_socket_context_on_open, us_socket_context_on_open)(getSocketContext(), [](auto *s, int is_client) { - HttpContextData *httpContextData = getSocketContextData(s); - - std::cout << "Opened http connection" << std::endl; - + /* Any connected socket should timeout until it has a request */ static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); + /* Init socket ext */ new (static_dispatch(us_ssl_socket_ext, us_socket_ext)(s)) HttpResponseData; return s; @@ -61,9 +57,21 @@ private: /* Handle socket disconnections */ static_dispatch(us_ssl_socket_context_on_close, us_socket_context_on_close)(getSocketContext(), [](auto *s) { - HttpContextData *httpContextData = getSocketContextData(s); + /* Get socket ext */ + HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s); - ((HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s))->~HttpResponseData(); + /* Signal broken out stream */ + if (httpResponseData->outStream) { + httpResponseData->outStream(-1); + } + + /* Signal broken in stream */ + if (httpResponseData->inStream) { + httpResponseData->inStream(std::string_view(nullptr, 0)); + } + + /* Destruct socket ext */ + httpResponseData->~HttpResponseData(); return s; }); @@ -72,7 +80,14 @@ private: static_dispatch(us_ssl_socket_context_on_data, us_socket_context_on_data)(getSocketContext(), [](auto *s, char *data, int length) { HttpContextData *httpContextData = getSocketContextData(s); - // cork this socket (move this to loop?) + /* Do not accept any data while in shutdown state */ + if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) s)) { + return s; + } + + // basically, when getting a new request we need to disable timeouts and enter paused mode! + + /* Cork this socket */ ((AsyncSocket *) s)->cork(); // pass this pointer to pointer along with the routing and change it if upgraded @@ -81,9 +96,13 @@ private: HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s); httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) { + // whenever we get (the first) request we should disable timeout? + // warning: if we are in shutdown state, resetting the timer is a security issue! + // todo: do not reset timer, disable it to allow hang requests! static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, HTTP_IDLE_TIMEOUT_S); + /* Reset httpResponse */ HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s); httpResponseData->offset = 0; httpResponseData->state = 0; @@ -94,6 +113,8 @@ private: }; httpContextData->router.route("get", 3, httpRequest->getUrl().data(), httpRequest->getUrl().length(), &userData); + // here we can be closed and in shutdown? + }, [httpResponseData](void *user, std::string_view data) { if (httpResponseData->inStream) { httpResponseData->inStream(data); @@ -114,25 +135,41 @@ private: /* Handle HTTP write out */ static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) { - // I think it's fair to never mind this one -> if we keep writing data after shutting down then that's an issue for us - static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); - + /* Silence any spurious writable events due to SSL_read failing to write */ AsyncSocket *asyncSocket = (AsyncSocket *) s; - - // get next chunk to send HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getExt(); - - if (httpResponseData->outStream) { - auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); - - // send, including any buffered up - httpResponseData->offset += asyncSocket->mergeDrain(chunk); - } else { - std::cout << "We did not have any outStream!" << std::endl; - - asyncSocket->mergeDrain(std::string_view(nullptr, 0)); + if (httpResponseData->state & HttpResponseData::HTTP_PAUSED_STREAM_OUT) { + return s; } + /* Writing data should reset the timeout */ + static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); + + /* Are we already ended and just waiting for a drain / shutdown? */ + if (httpResponseData->state & HttpResponseData::HTTP_ENDED_STREAM_OUT) { + + /* Try and send everything buffered up */ + asyncSocket->mergeDrain(); + + /* If we succeed with drainage we can finally shut down */ + if (!asyncSocket->hasBuffer()) { + asyncSocket->shutdown(); + } + + /* Nothing here for us */ + return s; + } + + if (httpResponseData->outStream) { + /* Regular path, request more data */ + auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); + httpResponseData->offset += asyncSocket->mergeDrain(chunk); + + // todo: we should loop until we cannot send anymore just like we do in HttpResponse::write(stream)! + } else { + /* We can come here if we only have socket buffers to drain yet no attached stream */ + asyncSocket->mergeDrain(); + } return s; }); @@ -140,26 +177,18 @@ private: /* Handle FIN, HTTP does not support half-closed sockets, so simply close */ static_dispatch(us_ssl_socket_context_on_end, us_socket_context_on_end)(getSocketContext(), [](auto *s) { - // static_dispatch(us_ssl_socket_close, us_socket_close)(s); + /* We do not care for half closed sockets */ AsyncSocket *asyncSocket = (AsyncSocket *) s; - asyncSocket->close(); + return asyncSocket->close(); - return s; }); - /* Handle socket timeouts */ + /* Handle socket timeouts, simply close them so to not confuse client with FIN */ static_dispatch(us_ssl_socket_context_on_timeout, us_socket_context_on_timeout)(getSocketContext(), [](auto *s) { - if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)(s)) { - std::cout << "Forcefully closing socket since shutdown was not answered in time" << std::endl; - static_dispatch(us_ssl_socket_close, us_socket_close)(s); - } else { - std::cout << "Shutting down socket now" << std::endl; - static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); - static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)(s); - } - - return s; + /* Force close rather than gracefully shutdown and risk confusing the client with a complete download */ + AsyncSocket *asyncSocket = (AsyncSocket *) s; + return asyncSocket->close(); }); @@ -177,16 +206,22 @@ public: httpContext = (HttpContext *) us_create_socket_context((us_loop *) loop, sizeof(HttpContextData)); } + if (!httpContext) { + return nullptr; + } + /* Init socket context data */ new ((HttpContextData *) static_dispatch(us_ssl_socket_context_ext, us_socket_context_ext)((SOCKET_CONTEXT_TYPE *) httpContext)) HttpContextData(); return httpContext->init(); } /* Destruct the HttpContext, it does not follow RAII */ void free() { + /* Destruct socket context data */ + HttpContextData *httpContextData = getSocketContextData(); + httpContextData->~HttpContextData(); - // call destructor! - + /* Free the socket context in whole */ static_dispatch(us_ssl_socket_context_free, us_socket_context_free)(getSocketContext()); } diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 052ce7a..602fca6 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -96,19 +96,25 @@ public: return; } + std::cout << "Resume called and we really are paused" << std::endl; + /* Remove paused status */ httpResponseData->state &= ~HttpResponseData::HTTP_PAUSED_STREAM_OUT; - int written = AsyncSocket::write(chunk.data(), chunk.length(), true); + if (chunk.length()) { + int written = AsyncSocket::write(chunk.data(), chunk.length(), true); - if (written == chunk.length()) { - // pull a new chunk from the callback (basically call onWritable) + if (written == chunk.length()) { + // pull a new chunk from the callback (basically call onWritable) + } } + + // no, basically just write this off and if all written, call streamOut callback } - /* Attach a write handler for sending data. Chunks might be read more than once */ + /* Attach an output stream function. Chunks may be read more than once. Negative offset mean broken stream */ void write(std::function(int)> cb, int length = 0) { HttpResponseData *httpResponseData = getHttpResponseData(); @@ -140,21 +146,21 @@ public: /* Handle PAUSE and FIN */ if (chunk.length() == 0) { - if (chunk == HTTP_STREAM_FIN.second) { + /* FIN */ + if (chunk.data()) { /* Try flush and shut down */ AsyncSocket::uncork(); if (!AsyncSocket::hasBuffer()) { /* Uncork finished with no buffered data */ - us_socket_shutdown((us_socket *) this); + AsyncSocket::shutdown(); } else { /* Let it shut down when drained */ httpResponseData->state |= HttpResponseData::HTTP_ENDED_STREAM_OUT; } } else { - std::cout << "Paused stream is not implemented!" << std::endl; - - // skip sending optional, yet keep refusing to call onWritable while in paused mode (SSL may poll for writable!) - // we thus need a status: paused to check for before requesting more data (also check for this in resume call!) + /* Disable timeout and mark this stream as paused (important to silence spurious onWritable events) */ + AsyncSocket::timeout(0); + httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; } return; }