diff --git a/fuzzing/TopicTree.cpp b/fuzzing/TopicTree.cpp index 8bc1897..24a91d1 100644 --- a/fuzzing/TopicTree.cpp +++ b/fuzzing/TopicTree.cpp @@ -9,7 +9,10 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { /* Create topic tree */ - uWS::TopicTree topicTree([](uWS::Subscriber *s, std::pair message) { + uWS::TopicTree topicTree([](uWS::Subscriber *s, uWS::Intersection &intersection) { + + /* For now we do not care about iterating over holes! TODO! */ + std::pair message = intersection.dataChannels; /* Subscriber must not be null, and at this point we have to have subscriptions. * This assumption seems to hold true. */ diff --git a/src/HttpResponse.h b/src/HttpResponse.h index d91fe64..dc31f4f 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -86,7 +86,7 @@ private: #ifndef UWS_HTTPRESPONSE_NO_WRITEMARK if (!Super::getLoopData()->noMark) { /* We only expose major version */ - writeHeader("uWebSockets", "18"); + writeHeader("uWebSockets", "19"); } #endif } diff --git a/src/TopicTree.h b/src/TopicTree.h index 43f3db5..44cb4fb 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -1,5 +1,5 @@ /* - * Authored by Alex Hultman, 2018-2020. + * Authored by Alex Hultman, 2018-2021. * Intellectual property of third-party. * Licensed under the Apache License, Version 2.0 (the "License"); @@ -28,6 +28,9 @@ #include #include +/* We use std::function here, not fu2::unique_function */ +#include + namespace uWS { /* A Subscriber is an extension of a socket */ @@ -64,15 +67,83 @@ struct Topic { std::set subs; }; +struct Hole { + std::pair lengths; + unsigned int messageId; +}; + +struct Intersection { + std::pair dataChannels; + std::vector holes; + + void forSubscriber(std::vector &senderForMessages, std::function, bool)> cb) { + /* How far we already emitted of the two dataChannels */ + std::pair emitted = {}; + + /* Holes are global to the entire topic tree, so we are not guaranteed to find + * holes in this intersection - they are sorted, though */ + unsigned int examinedHoles = 0; + + /* This is a slow path of sorts, most subscribers will be observers, not active senders */ + for (unsigned int id : senderForMessages) { + std::pair toEmit = {}; + std::pair toIgnore = {}; + + /* This linear search is most probably very small - it could be made log2 if every hole + * knows about its previous accumulated length, which is easy to set up. However this + * log2 search will most likely never be a warranted perf. gain */ + for (; examinedHoles < holes.size(); examinedHoles++) { + if (holes[examinedHoles].messageId == id) { + toIgnore.first += holes[examinedHoles].lengths.first; + toIgnore.second += holes[examinedHoles].lengths.second; + examinedHoles++; + break; + } + /* We are not the sender of this message so we should emit it in this segment */ + toEmit.first += holes[examinedHoles].lengths.first; + toEmit.second += holes[examinedHoles].lengths.second; + } + + /* Emit this segment */ + if (toEmit.first || toEmit.second) { + std::pair cutDataChannels = { + std::string_view(dataChannels.first.data() + emitted.first, toEmit.first), + std::string_view(dataChannels.second.data() + emitted.second, toEmit.second), + }; + + /* We only need to test the first data channel for "FIN" */ + cb(cutDataChannels, emitted.first + toEmit.first + toIgnore.first == dataChannels.first.length()); + } + + emitted.first += toEmit.first + toIgnore.first; + emitted.second += toEmit.second + toIgnore.second; + } + + if (emitted.first == dataChannels.first.length() && emitted.second == dataChannels.second.length()) { + return; + } + + std::pair cutDataChannels = { + std::string_view(dataChannels.first.data() + emitted.first, dataChannels.first.length() - emitted.first), + std::string_view(dataChannels.second.data() + emitted.second, dataChannels.second.length() - emitted.second), + }; + + cb(cutDataChannels, true); + } +}; + struct TopicTree { private: - std::function)> cb; + std::function cb; Topic *root = new Topic; /* Global messageId for deduplication of overlapping topics and ordering between topics */ unsigned int messageId = 0; + /* Sender holes */ + std::map> senderHoles; + /* The triggered topics */ Topic *triggeredTopics[64]; int numTriggeredTopics = 0; @@ -187,7 +258,7 @@ private: public: - TopicTree(std::function)> cb) { + TopicTree(std::function cb) { this->cb = cb; } @@ -195,6 +266,18 @@ public: delete root; } + /* This is part of the fast path, so should be optimal */ + std::vector &getSenderFor(Subscriber *s) { + static thread_local std::vector emptyVector; + + auto it = senderHoles.find(s); + if (it != senderHoles.end()) { + return it->second; + } + + return emptyVector; + } + void subscribe(std::string_view topic, Subscriber *subscriber) { /* Start iterating from the root */ Topic *iterator = root; @@ -251,8 +334,14 @@ public: } } - void publish(std::string_view topic, std::pair message) { + void publish(std::string_view topic, std::pair message, Subscriber *sender = nullptr) { + /* Add a hole for the sender if one */ + if (sender) { + senderHoles[sender].push_back(messageId); + } + publish(root, 0, 0, topic, message); + /* MessageIDs are reset on drain - this should be fine since messages itself are cleared on drain */ messageId++; } @@ -340,6 +429,8 @@ public: numTriggeredTopics = numFilteredTriggeredTopics; if (!numTriggeredTopics) { + senderHoles.clear(); + messageId = 0; return; } @@ -355,7 +446,7 @@ public: if (min != (Subscriber *)UINTPTR_MAX) { /* Up to 64 triggered Topics per batch */ - std::map> intersectionCache; + std::map intersectionCache; /* Loop over these here */ std::set::iterator it[64]; @@ -402,7 +493,7 @@ public: } /* Generate cache for intersection */ - if (intersectionCache[intersection].first.length() == 0) { + if (intersectionCache[intersection].dataChannels.first.length() == 0) { /* Build the union in order without duplicates */ std::map> complete; @@ -411,10 +502,19 @@ public: } /* Create the linear cache, {inflated, deflated} */ - std::pair res; + Intersection res; for (auto &p : complete) { - res.first.append(p.second.first); - res.second.append(p.second.second); + res.dataChannels.first.append(p.second.first); + res.dataChannels.second.append(p.second.second); + + /* Appends {id, length, length} + * We could possibly append byte offset also, + * if we want to use log2 search later. */ + Hole h; + h.lengths.first = p.second.first.length(); + h.lengths.second = p.second.second.length(); + h.messageId = p.first; + res.holes.push_back(h); } cb(min, intersectionCache[intersection] = std::move(res)); @@ -425,7 +525,6 @@ public: min = nextMin; } - } /* Clear messages of triggered Topics */ @@ -434,6 +533,8 @@ public: triggeredTopics[i]->triggered = false; } numTriggeredTopics = 0; + senderHoles.clear(); + messageId = 0; } }; diff --git a/src/WebSocket.h b/src/WebSocket.h index fcf40d1..6a69ea5 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -230,8 +230,15 @@ public: WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); - /* Is the same as publishing per websocket context */ - webSocketContextData->publish(topic, message, opCode, compress); + + /* Make us a subscriber if we aren't yet (important for allocating a sender address) */ + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); + if (!webSocketData->subscriber) { + webSocketData->subscriber = new Subscriber(this); + } + + /* Publish as sender, does not receive its own messages even if subscribed to relevant topics */ + webSocketContextData->publish(topic, message, opCode, compress, webSocketData->subscriber); } }; diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index db2a04f..99c2f07 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -76,79 +76,111 @@ public: Loop::get()->removePreHandler(this); } - WebSocketContextData() : topicTree([this](Subscriber *s, std::pair data) -> int { + WebSocketContextData() : topicTree([this](Subscriber *s, Intersection &intersection) -> int { + + /* 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); } @@ -168,7 +200,7 @@ public: } /* Helper for topictree publish, common path from app and ws */ - void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress) { + void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress, Subscriber *sender = nullptr) { /* We frame the message right here and only pass raw bytes to the pub/subber */ char *dst = (char *) malloc(protocol::messageFrameSize(message.size())); size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), opCode, message.length(), false); @@ -176,7 +208,7 @@ public: /* If compression is disabled */ if (compression == DISABLED) { /* Leave second field empty as nobody will ever read it */ - topicTree.publish(topic, {std::string_view(dst, dst_length), {}}); + topicTree.publish(topic, {std::string_view(dst, dst_length), {}}, sender); } else { /* DEDICATED_COMPRESSOR always takes the same path as must always have MessageMetadata as head */ if (compress || compression != SHARED_COMPRESSOR) { @@ -193,7 +225,7 @@ public: size_t dst_compressed_length = protocol::formatMessage(dst_compressed, compressedMessage.data(), compressedMessage.length(), opCode, compressedMessage.length(), true); /* Always publish the shortest one in any case */ - topicTree.publish(topic, {std::string_view(dst, dst_length), dst_compressed_length >= dst_length ? std::string_view(dst, dst_length) : std::string_view(dst_compressed, dst_compressed_length)}); + topicTree.publish(topic, {std::string_view(dst, dst_length), dst_compressed_length >= dst_length ? std::string_view(dst, dst_length) : std::string_view(dst_compressed, dst_compressed_length)}, sender); /* We don't care for allocation here */ ::free(dst_compressed); @@ -214,14 +246,14 @@ public: topicTree.publish(topic, { std::string_view(dst, dst_length), std::string_view(dst_compressed, message.length() + sizeof(MessageMetadata)) - }); + }, sender); ::free(dst_compressed); } } else { /* If not compressing, put same message on both tracks (only valid for SHARED_COMPRESSOR). * DEDICATED_COMPRESSOR_xKB must never end up here as we don't put a proper head here. */ - topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}); + topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}, sender); } } diff --git a/tests/Makefile b/tests/Makefile index c6ff50c..c846b03 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -1,9 +1,9 @@ default: - #$(CXX) -std=c++17 -fsanitize=address TopicTree.cpp -o TopicTree - #./TopicTree + $(CXX) -std=c++17 -fsanitize=address TopicTree.cpp -o TopicTree + ./TopicTree $(CXX) -std=c++17 -fsanitize=address HttpRouter.cpp -o HttpRouter ./HttpRouter $(CXX) -std=c++17 -fsanitize=address BloomFilter.cpp -o BloomFilter ./BloomFilter $(CXX) -std=c++17 -fsanitize=address ExtensionsNegotiator.cpp -o ExtensionsNegotiator - ./ExtensionsNegotiator \ No newline at end of file + ./ExtensionsNegotiator diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 1feb436..389d1aa 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -3,21 +3,40 @@ #include #include -void testUnsubscribeInside() { - std::cout << "TestUnsubscribeInside" << std::endl; +/* Modifying the topicTree inside callback is not allowed, we had + * tests for this before but we never need this to work anyways. + * Closing a socket when reaching too much backpressure is done + * deferred to next event loop iteration so we never need to modify + * topicTree inside callback - removed this test */ + +/* This tests pretty much all features for obvious incorrectness */ +void testCorrectness() { + std::cout << "TestCorrectness" << std::endl; uWS::TopicTree *topicTree; - std::map expectedResult; + std::map> expectedResult; + std::map> actualResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::string_view data) { - /* Check for unexpected subscribers */ - assert(expectedResult.find(s) != expectedResult.end()); + topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { - /* Check for unexpected data */ - assert(expectedResult[s] == data); + /* How many bytes we have in first data channel at time we get fin = true */ + unsigned int finAt = 0; - /* This one causes mess-up */ - topicTree->unsubscribeAll(s); + intersection.forSubscriber(topicTree->getSenderFor(s), [s, &finAt, &actualResult](std::pair dataChannels, bool fin) { + actualResult[s].first += dataChannels.first; + actualResult[s].second += dataChannels.second; + + /* Check that getting fin = true really is the last segment */ + if (!finAt && fin) { + finAt = actualResult[s].first.length(); + } + }); + + /* Assume finAt == actualResult[s].first.length() */ + if (actualResult[s].first.length() != finAt) { + std::cout << "ERROR! FinAt mismatching!" << std::endl; + exit(1); + } /* We actually don't use this one */ return 0; @@ -26,28 +45,60 @@ void testUnsubscribeInside() { uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); - /* Fill out expectedResult */ - expectedResult = { - {s1, "Ett!"}, - {s2, "Två!"} - }; - - /* Make sure s1 < s2 */ + /* Make sure s1 < s2 (for debugging) */ if (s2 < s1) { uWS::Subscriber *tmp = s1; s1 = s2; s2 = tmp; } - /* This order does not matter as it fills a tree */ - topicTree->subscribe("2", s2); - topicTree->subscribe("1", s1); + /* Publish to topic3 - nobody should see this */ + topicTree->publish("topic3", {std::string_view("Nobody"), std::string_view("should see")}, nullptr); - /* This order matters, as it fills triggeredTopics array in order */ - topicTree->publish("1", "Ett!"); - topicTree->publish("2", "Två!"); + /* Subscribe s1 to topic3 - s1 should not see above message */ + topicTree->subscribe("topic3", s1); + /* Publish to topic3 with s1 as sender - s1 should not get its own messages */ + topicTree->publish("topic3", {std::string_view("Nobody"), std::string_view("should see")}, s1); + + /* Subscribe s2 to topic3 - should not get any message */ + topicTree->subscribe("topic3", s2); + + /* Publish to topic3 without sender - both should see */ + topicTree->publish("topic3", {std::string_view("Both"), std::string_view("should see")}, nullptr); + + /* Publish to topic3 with s2 as sender - s1 should see */ + topicTree->publish("topic3", {std::string_view("s1"), std::string_view("should see, not s2")}, s2); + + /* Publish to topic3 with s1 as sender - s2 should see */ + topicTree->publish("topic3", {std::string_view("s2"), std::string_view("should see, not s1")}, s1); + + /* Publish to topic3 without sender - both should see */ + topicTree->publish("topic3", {std::string_view("Again, both"), std::string_view("should see this as well")}, nullptr); + + // todo: add more cases involving more topics and duplicates, etc + + /* Fill out expectedResult */ + expectedResult = { + {s1, {"Boths1Again, both", "should seeshould see, not s2should see this as well"}}, + {s2, {"Boths2Again, both", "should seeshould see, not s1should see this as well"}} + }; + + /* Compare result with expected result for every subscriber */ topicTree->drain(); + for (auto &p : expectedResult) { + std::cout << "Subscriber: " << p.first << std::endl; + + if (p.second.first != actualResult[p.first].first) { + std::cout << "ERROR: <" << actualResult[p.first].first << "> should be <" << p.second.first << ">" << std::endl; + exit(1); + } + + if (p.second.second != actualResult[p.first].second) { + std::cout << "ERROR: <" << actualResult[p.first].second << "> should be <" << p.second.second << ">" << std::endl; + exit(1); + } + } /* Release resources */ topicTree->unsubscribeAll(s1); @@ -60,5 +111,5 @@ void testUnsubscribeInside() { } int main() { - testUnsubscribeInside(); + testCorrectness(); } \ No newline at end of file