From f8800d3e091dda095e645b76a3995556472f90d2 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 29 Jan 2021 13:10:49 +0100 Subject: [PATCH] Proper topicTree integration --- src/WebSocketContextData.h | 130 ++++++++++++++++++++++--------------- 1 file changed, 78 insertions(+), 52 deletions(-) diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index e8d7dba..1488e5e 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -78,83 +78,109 @@ public: WebSocketContextData() : topicTree([this](Subscriber *s, Intersection &intersection) -> int { - - std::pair data = intersection.dataChannels; - - + /* We could potentially be called here even if we have nothing to send, since we can + * be the sender of every single message in this intersection. Also "fin" of a segment is not + * guaranteed to be set, in case remaining segments are all from us. + * Essentially, we cannot make strict assumptions here. Also, we can even come here corked, + * since publish can call drain! */ /* We rely on writing to regular asyncSockets */ auto *asyncSocket = (AsyncSocket *) s->user; - /* Check if we now have too much backpressure (todo: don't buffer up before check) */ - if (!maxBackpressure || (unsigned int) asyncSocket->getBufferedAmount() < maxBackpressure) { - /* Pick uncompressed data track */ - std::string_view selectedData = data.first; + /* If we are corked, do not uncork - otherwise if we cork in here, uncork before leaving */ + bool wasCorked = asyncSocket->isCorked(); - /* Are we using compression? Fine, pick the compressed data track */ - WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); - if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { + /* Do we even have room for potential data? */ + if (!maxBackpressure || asyncSocket->getBufferedAmount() < maxBackpressure) { - /* This is used for both shared and dedicated paths */ - selectedData = data.second; + /* Roll over all our segments */ + intersection.forSubscriber(topicTree.getSenderFor(s), [asyncSocket, this](std::pair data, bool fin) { - /* However, dedicated compression has its own path */ - if (compression != SHARED_COMPRESSOR) { + /* We have a segment that is not marked as last ("fin"). + * Cork if not already so (purely for performance reasons). Does not touch "wasCorked". */ + if (!fin && !asyncSocket->isCorked() && asyncSocket->canCork()) { + asyncSocket->cork(); + } - WebSocket *ws = (WebSocket *) asyncSocket; + /* Pick uncompressed data track */ + std::string_view selectedData = data.first; - /* We need to handle being corked, and corking here */ - bool needsUncorking = false; - if (!ws->isCorked() && ws->canCork()) { - asyncSocket->cork(); - needsUncorking = true; - } + /* Are we using compression? Fine, pick the compressed data track */ + WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); + if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { - while (selectedData.length()) { - /* Interpret the data like so, because this is how we shoved it in */ - MessageMetadata mm; - memcpy((char *) &mm, selectedData.data(), sizeof(MessageMetadata)); - std::string_view unframedMessage(selectedData.data() + sizeof(MessageMetadata), mm.length); + /* This is used for both shared and dedicated paths */ + selectedData = data.second; - /* Skip this message if our backpressure is too high */ - if (maxBackpressure && ws->getBufferedAmount() > maxBackpressure) { - break; + /* However, dedicated compression has its own path */ + if (compression != SHARED_COMPRESSOR) { + + WebSocket *ws = (WebSocket *) asyncSocket; + + /* For performance reasons we always cork when in dedicated mode. + * Is this really the best? We already kind of cork things in Zlib? + * Right, formatting needs a cork buffer, right. Never mind. */ + if (!ws->isCorked() && ws->canCork()) { + asyncSocket->cork(); } - /* Here we perform the actual compression and framing */ - ws->send(unframedMessage, mm.opCode, mm.compress); + while (selectedData.length()) { + /* Interpret the data like so, because this is how we shoved it in */ + MessageMetadata mm; + memcpy((char *) &mm, selectedData.data(), sizeof(MessageMetadata)); + std::string_view unframedMessage(selectedData.data() + sizeof(MessageMetadata), mm.length); - /* Advance until empty */ - selectedData.remove_prefix(sizeof(MessageMetadata) + mm.length); + /* Skip this message if our backpressure is too high */ + if (maxBackpressure && ws->getBufferedAmount() > maxBackpressure) { + break; + } + + /* Here we perform the actual compression and framing */ + ws->send(unframedMessage, mm.opCode, mm.compress); + + /* Advance until empty */ + selectedData.remove_prefix(sizeof(MessageMetadata) + mm.length); + } + + /* Continue to next segment without executing below path */ + return; } - - /* Here we need to uncork or keep it as was */ - if (needsUncorking) { - asyncSocket->uncork(); - } - - /* See below */ - return 0; } - } - /* Note: this assumes we are not corked, as corking will swallow things and fail later on */ - auto [written, failed] = asyncSocket->write(selectedData.data(), (int) selectedData.length()); + /* Common path for SHARED and DISABLED. It is an invalid assumption that we always are + * uncorked here, however the following (invalid) assumption is not critically wrong either way */ + + /* Note: this assumes we are not corked, as corking will swallow things and fail later on */ + auto [written, failed] = asyncSocket->write(selectedData.data(), (int) selectedData.length()); + /* If we want strict check for success, we can ignore this check if corked and repeat below + * when uncorking - however this is too strict as we really care about PROGRESS rather than + * ENTIRE SUCCESS - we need minor API changes to support correct checks */ + if (!failed) { + if (this->resetIdleTimeoutOnSend) { + asyncSocket->timeout(this->idleTimeout); + } + } + }); + } + + /* We are done sending, for whatever reasons we ended up corked while not starting with "wasCorked", + * we here need to uncork to restore the state we were called in */ + if (!wasCorked && asyncSocket->isCorked()) { + /* Regarding timeout for writes; */ + auto [written, failed] = asyncSocket->uncork(); + /* Again, this check should be more like DID WE PROGRESS rather than DID WE SUCCEED ENTIRELY */ if (!failed) { if (this->resetIdleTimeoutOnSend) { asyncSocket->timeout(this->idleTimeout); } } - - /* Failing here must not immediately close the socket, as that could result in stack overflow, - * iterator invalidation and other TopicTree::drain bugs. We may shutdown the reading side of the socket, - * causing next iteration to error-close the socket from that context instead, if we want to */ } - /* If we have too much backpressure, simply skip sending from here */ - - /* Also (defer) a close if we have too much backpressure if that is what we want */ + /* Defer a close if we now have (or already had) too much backpressure, or simply skip */ if (maxBackpressure && closeOnBackpressureLimit && asyncSocket->getBufferedAmount() > maxBackpressure) { + /* We must not immediately close the socket, as that could result in stack overflow, + * iterator invalidation and other TopicTree::drain bugs. We may shutdown the reading side of the socket, + * causing next iteration to error-close the socket from that context instead, if we want to */ us_socket_shutdown_read(SSL, (us_socket_t *) asyncSocket); }