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

New AsyncSocket interface with bool failure

Browse Source
This commit is contained in:
Alex Hultman
2018-10-28 00:46:28 +02:00
parent 5fd1edc907
commit f558bce8ca
1 changed files with 92 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/AsyncSocket.h
+92 -120
View File
@@ -9,11 +9,6 @@
namespace uWS { 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<T> -> TimeoutSocket or Socket (make StaticDispath Socket)
template <bool SSL> template <bool SSL>
struct AsyncSocket : StaticDispatch<SSL> { struct AsyncSocket : StaticDispatch<SSL> {
template <bool> friend struct HttpContext; template <bool> friend struct HttpContext;
@@ -21,161 +16,134 @@ protected:
using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE; using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE;
using StaticDispatch<SSL>::static_dispatch; using StaticDispatch<SSL>::static_dispatch;
/* Get loop data for socket */
LoopData *getLoopData() { LoopData *getLoopData() {
if constexpr(SSL) { return (LoopData *) us_loop_ext(
return (LoopData *) us_loop_ext(us_ssl_socket_context_loop(us_ssl_socket_get_context((SOCKET_TYPE *) this))); static_dispatch(us_ssl_socket_context_loop, us_socket_context_loop)(
} else { static_dispatch(us_ssl_socket_get_context, us_socket_get_context)((SOCKET_TYPE *) this))
return (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((SOCKET_TYPE *) this))); );
}
} }
/* Get socket extension */
void *getExt() { void *getExt() {
return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this); return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
} }
/* Socket timeout */
void timeout(unsigned int seconds) { void timeout(unsigned int seconds) {
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds); static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds);
} }
/* Shutdown socket without any automatic drainage */
void shutdown() { void shutdown() {
static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this); static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this);
} }
/* Immediately close socket */
SOCKET_TYPE *close() { SOCKET_TYPE *close() {
return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); 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 */ /* Cork this socket. Only one socket may ever be corked per-loop at any given time */
void cork() { void cork() {
//std::cout << "Cork called" << std::endl; getLoopData()->corked = true;
LoopData *loopData = getLoopData();
loopData->corked = true;
} }
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. */ /* 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) * Returns pair of bytes written (anywhere) and wheter or not this call resulted in the polling for
int write(const char *src, int length, bool optionally = false, int nextLength = 0) { * writable (or we are in a state that implies polling for writable). */
std::pair<int, bool> write(const char *src, int length, bool optionally = false, int nextLength = 0) {
LoopData *loopData = getLoopData(); LoopData *loopData = getLoopData();
//std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl;
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt(); AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
/* We are limited if we have a per-socket buffer */ /* We are limited if we have a per-socket buffer */
if (asyncSocketData->buffer.length()) { 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) { if (optionally) {
std::cout << "Write returned: " << written << std::endl; /* Thankfully we can exit early here */
return written; 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) */ if (length) {
/* At least we can reserve room for next chunk if we know it up front */ if (loopData->corked) {
if (nextLength) { /* We are corked */
asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + length - written + nextLength); 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 */ /* Did we fail? */
asyncSocketData->buffer.append(src + written, length - written); 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; /* Default fall through return */
return length; return {length, false};
} }
/* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */ /* 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)! */ /* 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::pair<int, bool> uncork(const char *src = nullptr, int length = 0, bool optionally = false) {
//std::cout << "Uncork called with length: " << length << std::endl;
LoopData *loopData = getLoopData(); LoopData *loopData = getLoopData();
if (loopData->corked) { if (loopData->corked) {
@@ -183,17 +151,21 @@ protected:
if (loopData->corkOffset) { if (loopData->corkOffset) {
/* Corked data is already accounted for via its write call */ /* 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; 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 */ /* We should only return with new writes, not things written to cork already */
return write(src, length, optionally, 0); return write(src, length, optionally, 0);
} else { } else {
//std::cout << "Not even corked!" << std::endl; /* We are not even corked! */
return {0, false};
} }
return 0;
} }
}; };