From f558bce8ca151f4a10c76c736f49a285f3fac9cf Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sun, 28 Oct 2018 00:46:28 +0200 Subject: [PATCH] New AsyncSocket interface with bool failure --- src/AsyncSocket.h | 212 ++++++++++++++++++++-------------------------- 1 file changed, 92 insertions(+), 120 deletions(-) diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 979950c..88c7746 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -9,11 +9,6 @@ namespace uWS { -// we need a variant of this where any failed write triggers a timeout! -// AsyncSocket needs to derive from a uSockets base we can select like so: - -// HttpResponse -> AsyncSocket -> TimeoutSocket or Socket (make StaticDispath Socket) - template struct AsyncSocket : StaticDispatch { template friend struct HttpContext; @@ -21,161 +16,134 @@ protected: using SOCKET_TYPE = typename StaticDispatch::SOCKET_TYPE; using StaticDispatch::static_dispatch; + /* Get loop data for socket */ LoopData *getLoopData() { - if constexpr(SSL) { - return (LoopData *) us_loop_ext(us_ssl_socket_context_loop(us_ssl_socket_get_context((SOCKET_TYPE *) this))); - } else { - return (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((SOCKET_TYPE *) this))); - } + return (LoopData *) us_loop_ext( + static_dispatch(us_ssl_socket_context_loop, us_socket_context_loop)( + static_dispatch(us_ssl_socket_get_context, us_socket_get_context)((SOCKET_TYPE *) this)) + ); } + /* Get socket extension */ void *getExt() { return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this); } + /* Socket timeout */ void timeout(unsigned int seconds) { static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds); } + /* Shutdown socket without any automatic drainage */ void shutdown() { static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this); } + /* Immediately close socket */ SOCKET_TYPE *close() { return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); } - bool isFullyOpen() { - // todo:: not shutdown or closed - return true; - } - /* Cork this socket. Only one socket may ever be corked per-loop at any given time */ void cork() { - //std::cout << "Cork called" << std::endl; - - LoopData *loopData = getLoopData(); - loopData->corked = true; + getLoopData()->corked = true; } - /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. */ - // todo: consider supporting nextLength = UNKNOWN as -1 (more but unknown size) - int write(const char *src, int length, bool optionally = false, int nextLength = 0) { + /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. + * Returns pair of bytes written (anywhere) and wheter or not this call resulted in the polling for + * writable (or we are in a state that implies polling for writable). */ + std::pair write(const char *src, int length, bool optionally = false, int nextLength = 0) { LoopData *loopData = getLoopData(); - - //std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl; - AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); /* We are limited if we have a per-socket buffer */ if (asyncSocketData->buffer.length()) { + /* Write off as much as we can */ + int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), /*nextLength != 0 | */length); - std::cout << "WHAT THE FUCK WE HAVE BUFFER!" << std::endl; + /* On failure return, otherwise continue down the function */ + if (written < asyncSocketData->buffer.length()) { - // probably want to swap the hierarchy here to: if buffer, if length, if optionally + /* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */ + asyncSocketData->buffer = asyncSocketData->buffer.substr(written); - /* We only try and drain if we are in optional mode */ - if (optionally) { - - - /* Is this a merge drain or not? */ - if (length) { - // merge drain may happen if the developer continues to write data despite having buffered up content - - - // for now we do not support this yet - return 0; - - } else { - /* Write off as much as we can */ - int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), nextLength != 0); - - /* Update buffering (should probably have different allocation strategies here) */ - asyncSocketData->buffer = asyncSocketData->buffer.substr(written); - - /* We consumed no new data */ - return 0; - } - } else { - /* Currently there is no drainage for non-optional writes */ - std::cout << "Buffering at top of write (really bad)!" << std::endl; - - /* At least we can reserve room for next chunk if we know it up front */ - if (nextLength) { - asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + length + nextLength); - } - - /* Buffer this chunk */ - asyncSocketData->buffer.append(src, length); - //std::cout << "Write returned: " << length << std::endl; - return length; - } - } - - /* Do nothing for a null sized chunk */ - if (!length) { - //std::cout << "Trying to write 0 length!" << std::endl; - return 0; - } - - if (loopData->corked) { - /* We are corked */ - if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) { - /* If the entire chunk fits in cork buffer */ - memcpy(loopData->corkBuffer + loopData->corkOffset, src, length); - loopData->corkOffset += length; - } else { - /* Strategy differences between SSL and non-SSL */ - if constexpr(SSL) { - /* Cork up as much as we can, optionally does not matter here as we know it will fit in cork */ - int written = write(src, std::min(LoopData::CORK_BUFFER_SIZE - loopData->corkOffset, length), false, 0); - - /* Optionally matters here though */ - written += uncork(src + written, length - written, optionally); - //std::cout << "Write returned: " << written << std::endl; - return written; - } else { - /* For non-SSL we take the penalty of two syscalls */ - int written = uncork(src, length, optionally); - //std::cout << "Write returned: " << written << std::endl; - return written; - } - } - } else { - /* We are not corked */ - int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, src, length, nextLength != 0); - - /* Did we fail? */ - if (written < length) { - /* If the write was optional then just bail out */ if (optionally) { - std::cout << "Write returned: " << written << std::endl; - return written; + /* Thankfully we can exit early here */ + return {0, true}; + } else { + /* This path is horrible and points towards erroneous usage */ + asyncSocketData->buffer.append(src, length); + + // todo: remove this when we are no longer morons + std::cout << "Warning: fudge this is horrible!!" << std::endl; + + return {length, true}; } + } - std::cout << "Buffering at bottom of write (okay)!" << std::endl; + /* At this point we simply have no buffer and can continue as normal */ + asyncSocketData->buffer.clear(); + } - /* Fall back to worst possible case (should be very rare for HTTP) */ - /* At least we can reserve room for next chunk if we know it up front */ - if (nextLength) { - asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + length - written + nextLength); + if (length) { + if (loopData->corked) { + /* We are corked */ + if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) { + /* If the entire chunk fits in cork buffer */ + memcpy(loopData->corkBuffer + loopData->corkOffset, src, length); + loopData->corkOffset += length; + /* Fall through to default return */ + } else { + /* Strategy differences between SSL and non-SSL regarding syscall minimizing */ + if constexpr (SSL) { + /* Cork up as much as we can */ + int stripped = LoopData::CORK_BUFFER_SIZE - loopData->corkOffset; + memcpy(loopData->corkBuffer + loopData->corkOffset, src, stripped); + loopData->corkOffset = LoopData::CORK_BUFFER_SIZE; + + auto [written, failed] = uncork(src + stripped, length - stripped, optionally); + return {written + stripped, failed}; + } + + /* For non-SSL we take the penalty of two syscalls */ + return uncork(src, length, optionally); } + } else { + /* We are not corked */ + int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, src, length, nextLength != 0); - /* Buffer this chunk */ - asyncSocketData->buffer.append(src + written, length - written); + /* Did we fail? */ + if (written < length) { + /* If the write was optional then just bail out */ + if (optionally) { + return {written, true}; + } + + std::cout << "Buffering at bottom of write (okay)!" << std::endl; + + /* Fall back to worst possible case (should be very rare for HTTP) */ + /* At least we can reserve room for next chunk if we know it up front */ + if (nextLength) { + asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + length - written + nextLength); + } + + /* Buffer this chunk */ + asyncSocketData->buffer.append(src + written, length - written); + + /* Return the failure */ + return {length, true}; + } + /* Fall through to default return */ } } - //std::cout << "Write returned: " << length << std::endl; - return length; + /* Default fall through return */ + return {length, false}; } /* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */ /* It does NOT count bytes written from cork buffer (they are already accounted for in the write call responsible for its corking)! */ - int uncork(const char *src = nullptr, int length = 0, bool optionally = false) { - - //std::cout << "Uncork called with length: " << length << std::endl; - + std::pair uncork(const char *src = nullptr, int length = 0, bool optionally = false) { LoopData *loopData = getLoopData(); if (loopData->corked) { @@ -183,17 +151,21 @@ protected: if (loopData->corkOffset) { /* Corked data is already accounted for via its write call */ - write(loopData->corkBuffer, loopData->corkOffset, false, length); + auto [written, failed] = write(loopData->corkBuffer, loopData->corkOffset, false, length); loopData->corkOffset = 0; + + if (failed) { + /* We do not need to care for buffering here, write does that */ + return {0, true}; + } } /* We should only return with new writes, not things written to cork already */ return write(src, length, optionally, 0); } else { - //std::cout << "Not even corked!" << std::endl; + /* We are not even corked! */ + return {0, false}; } - - return 0; } };