diff --git a/examples/BroadcastingEchoServer.cpp b/examples/BroadcastingEchoServer.cpp index 00abece..e072109 100644 --- a/examples/BroadcastingEchoServer.cpp +++ b/examples/BroadcastingEchoServer.cpp @@ -36,7 +36,7 @@ int main() { PerSocketData *perSocketData = (PerSocketData *) ws->getUserData(); - for (int i = 0; i < 100; i++) { + for (int i = 0; i < 32; i++) { std::string topic = std::to_string((uintptr_t)ws) + "-" + std::to_string(i); perSocketData->topics.push_back(topic); ws->subscribe(topic); @@ -45,7 +45,8 @@ int main() { .message = [&app](auto *ws, std::string_view message, uWS::OpCode opCode) { PerSocketData *perSocketData = (PerSocketData *) ws->getUserData(); - app->publish(perSocketData->topics[++perSocketData->nr % 100], message, opCode); + app->publish(perSocketData->topics[(size_t)(++perSocketData->nr % 32)], message, opCode); + ws->publish(perSocketData->topics[(size_t)(++perSocketData->nr % 32)], message, opCode); }, .drain = [](auto */*ws*/) { /* Check ws->getBufferedAmount() here */ diff --git a/fuzzing/EpollHelloWorld.cpp b/fuzzing/EpollHelloWorld.cpp index b6b15e6..831029e 100644 --- a/fuzzing/EpollHelloWorld.cpp +++ b/fuzzing/EpollHelloWorld.cpp @@ -55,7 +55,7 @@ void test() { } /* Some invalid queries */ req->getParameter(30000); - req->getParameter(-34234); + req->getParameter((unsigned short) -34234); req->getHeader("yhello"); req->getQuery(); req->getQuery("assd"); diff --git a/fuzzing/TopicTree.cpp b/fuzzing/TopicTree.cpp index 2917e3e..2fc770c 100644 --- a/fuzzing/TopicTree.cpp +++ b/fuzzing/TopicTree.cpp @@ -9,28 +9,25 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { /* Create topic tree */ - 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. */ - if (!s->subscriptions.size()) { - free((void *) -1); - } + uWS::TopicTree topicTree([](uWS::Subscriber *s, std::string &message, auto flags) { /* Depending on what publishing we do below (with or without empty strings), * this assumption can hold true or not. For now it should hold true */ - if (!message.first.length()) { + if (!message.length()) { free((void *) -1); } - return 0; + /* Break if we have no subscriptions (not really an error, just to bring more randomness) */ + if (s->topics.size() == 0) { + return true; + } + + /* Success */ + return false; }); /* Holder for all manually allocated subscribers */ - std::map> subscribers; + std::map subscribers; /* Iterate the padded fuzz as chunks */ makeChunked(makePadded(data, size), size, [&topicTree, &subscribers](const uint8_t *data, size_t size) { @@ -62,35 +59,48 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { return; } - uWS::Subscriber *subscriber = new uWS::Subscriber(nullptr); - subscribers[id] = std::unique_ptr(subscriber); - topicTree.subscribe(lastString, subscriber); + uWS::Subscriber *subscriber = topicTree.createSubscriber(); + subscribers[id] = subscriber; + topicTree.subscribe(subscriber, lastString); } else { /* Limit per subscriber subscriptions (OOM) */ - uWS::Subscriber *subscriber = subscribers[id].get(); - if (subscriber->subscriptions.size() < 50) { - topicTree.subscribe(lastString, subscriber); + uWS::Subscriber *subscriber = subscribers[id]; + if (subscriber->topics.size() < 50) { + topicTree.subscribe(subscriber, lastString); } } } else if (data[4] == 'U') { /* Unsubscribe */ auto it = subscribers.find(id); if (it != subscribers.end()) { - topicTree.unsubscribe(lastString, it->second.get()); + topicTree.unsubscribe(it->second, lastString); } } else if (data[4] == 'A') { /* Unsubscribe from all */ auto it = subscribers.find(id); if (it != subscribers.end()) { - topicTree.unsubscribeAll(it->second.get()); + std::vector topics; + for (auto *topic : it->second->topics) { + topics.push_back(topic->name); + } + + for (std::string &topic : topics) { + topicTree.unsubscribe(it->second, topic); + } + } + } else if (data[4] == 'O') { + /* Drain one socket */ + auto it = subscribers.find(id); + if (it != subscribers.end()) { + topicTree.drain(it->second); } } else if (data[4] == 'P') { /* Publish only if we actually have data */ if (lastString.length()) { - topicTree.publish(lastString, {lastString, lastString}); + topicTree.publish(nullptr, lastString, std::string(lastString)); } else { /* We could use having more strings */ - topicTree.publish("", {"anything", "something else"}); + topicTree.publish(nullptr, "", "anything"); } } else { /* Drain for everything else (OOM) */ @@ -101,7 +111,7 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { /* Remove any subscriber from the tree */ for (auto &p : subscribers) { - topicTree.unsubscribeAll(p.second.get()); + topicTree.freeSubscriber(p.second); } return 0; diff --git a/src/App.h b/src/App.h index aa1f0a5..37d6aa6 100644 --- a/src/App.h +++ b/src/App.h @@ -101,7 +101,7 @@ public: * app has one conceptual Topic tree) */ void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress = false) { for (auto *webSocketContext : webSocketContexts) { - webSocketContext->getExt()->publish(topic, message, opCode, compress); + webSocketContext->getExt()->topicTree.publish(nullptr, topic, {std::string(message), opCode, compress}); } } @@ -116,7 +116,7 @@ public: Topic *t = webSocketContextData->topicTree.lookupTopic(topic); if (t) { - subscribers += t->subs.size(); + subscribers += t->size(); } } diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 188df64..f44509a 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -47,7 +47,7 @@ struct AsyncSocket { template friend struct HttpContext; template friend struct WebSocketContext; template friend struct WebSocketContextData; - friend struct TopicTree; + template friend struct TopicTree; protected: /* Returns SSL pointer or FD as pointer */ diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 2ce13ee..148db07 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -87,7 +87,7 @@ private: #ifndef UWS_HTTPRESPONSE_NO_WRITEMARK if (!Super::getLoopData()->noMark) { /* We only expose major version */ - writeHeader("uWebSockets", "19"); + writeHeader("uWebSockets", "20"); } #endif } diff --git a/src/TopicTree.h b/src/TopicTree.h index 6567008..ba6a263 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -18,599 +18,296 @@ #ifndef UWS_TOPICTREE_H #define UWS_TOPICTREE_H -#include -#include #include +#include +#include +#include +#include +#include +#include +#include #include #include #include -#include -#include -#include - -/* We use std::function here, not MoveOnlyFunction */ -#include namespace uWS { -/* A Subscriber is an extension of a socket */ +struct Subscriber; + +struct Topic : std::unordered_set { + + Topic(std::string_view topic) : name(topic) { + + } + + std::string name; +}; + struct Subscriber { - std::list subscriptions; - void *user; - Subscriber(void *user) : user(user) {} -}; - -struct Topic { - /* Memory for our name */ - char *name; - size_t length; - - /* Our parent or nullptr */ - Topic *parent = nullptr; - - /* Next triggered Topic */ - bool triggered = false; - - /* Exact string matches */ - std::map children; - - /* Wildcard child */ - Topic *wildcardChild = nullptr; - - /* Terminating wildcard child */ - Topic *terminatingWildcardChild = nullptr; - - /* What we published, {inflated, deflated} */ - std::map> messages; - - std::set subs; - - /* Locked or not, used only when iterating over a Subscriber's topics */ - bool locked = false; - - /* Full name is used when iterating topcis */ - std::string fullName; -}; - -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 = {}; - - /* This is a slow path of sorts, most subscribers will be observers, not active senders */ - if (!senderForMessages.empty()) { - - std::pair toEmit = {}; - unsigned int startAt = 0; - - /* Iterate each message looking for any to skip */ - for (auto &message : holes) { - - /* If this message was sent by this subscriber skip it */ - bool skipMessage = false; - for (unsigned int i = startAt; i < senderForMessages.size(); i++) { - if (senderForMessages[i] > message.messageId) { - startAt = i; - break; - } - if (message.messageId == senderForMessages[i]) { - skipMessage = true; - startAt = ++i; - break; - } - } - - /* Collect messages until a skip, then emit messages */ - if (!skipMessage) { - toEmit.first += message.lengths.first; - toEmit.second += message.lengths.second; - } else { - 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), - }; - /* Only need to test the first data channel for "FIN" */ - cb(cutDataChannels, emitted.first + toEmit.first + message.lengths.first == dataChannels.first.length()); - emitted.first += toEmit.first; - emitted.second += toEmit.second; - toEmit = {}; - } - /* This message is now accounted for, mark as emitted */ - emitted.first += message.lengths.first; - emitted.second += message.lengths.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 { - /* Returns Topic, or nullptr. Topic can be root if empty string given. */ - Topic *lookupTopic(std::string_view topic) { - /* Lookup exact Topic ptr from string */ - Topic *iterator = root; - for (size_t start = 0, stop = 0; stop != std::string::npos; start = stop + 1) { - stop = topic.find('/', start); - std::string_view segment = topic.substr(start, stop - start); - - std::map::iterator it = iterator->children.find(segment); - if (it == iterator->children.end()) { - /* This topic does not even exist */ - return nullptr; - } - - iterator = it->second; - } - - return iterator; - } + template friend struct TopicTree; private: - std::function cb; + /* We use a factory */ + Subscriber() = default; - Topic *root = new Topic; + /* State of prev, next does not matter unless we are needsDrainage() since we are not in the list */ + Subscriber *prev, *next; - /* Global messageId for deduplication of overlapping topics and ordering between topics */ - unsigned int messageId = 0; + /* Any one subscriber can be part of at most 32 publishes before it needs a drain, + * or whatever encoding of runs or whatever we might do in the future */ + uint16_t messageIndices[32]; - /* Sender holes */ - std::map> senderHoles; + /* This one matters the most, if it is 0 we are not in the list of drainableSubscribers */ + unsigned char numMessageIndices = 0; - /* The triggered topics */ - Topic *triggeredTopics[64]; - int numTriggeredTopics = 0; - Subscriber *min = (Subscriber *) UINTPTR_MAX; +public: - /* Cull or trim unused Topic nodes from leaf to root */ - void trimTree(Topic *topic) { - while (!topic->subs.size() && !topic->children.size() && !topic->terminatingWildcardChild && !topic->wildcardChild) { - Topic *parent = topic->parent; + /* We have a list of topics we subscribe to (read by WebSocket::iterateTopics) */ + std::set topics; - if (topic->length == 1) { - if (topic->name[0] == '#') { - parent->terminatingWildcardChild = nullptr; - } else if (topic->name[0] == '+') { - parent->wildcardChild = nullptr; - } - } - /* Erase us from our parents set (wildcards also live here) */ - parent->children.erase(std::string_view(topic->name, topic->length)); + /* User data */ + void *user; - /* If this node is triggered, make sure to remove it from the triggered list */ - if (topic->triggered) { - Topic *tmp[64]; - int length = 0; - for (int i = 0; i < numTriggeredTopics; i++) { - if (triggeredTopics[i] != topic) { - tmp[length++] = triggeredTopics[i]; - } - } + bool needsDrainage() { + return numMessageIndices; + } +}; - for (int i = 0; i < length; i++) { - triggeredTopics[i] = tmp[i]; - } - numTriggeredTopics = length; - } +template +struct TopicTree { - /* Free various memory for the node */ - delete [] topic->name; - delete topic; + enum IteratorFlags { + LAST = 1, + FIRST = 2 + }; - if (parent == root) { - break; - } +private: + /* Whomever is iterating this topic is locked to not modify its own list */ + Subscriber *iteratingSubscriber = nullptr; - topic = parent; + /* The drain callback must not publish, unsubscribe or subscribe. + * It must only cork, uncork, send, write */ + std::function cb; + + /* The topics */ + std::unordered_map> topics; + + /* List of subscribers that needs drainage */ + Subscriber *drainableSubscribers = nullptr; + + /* Palette of outgoing messages, up to 64k */ + std::vector outgoingMessages; + + void checkIteratingSubscriber(Subscriber *s) { + /* Notify user that they are doing something wrong here */ + if (iteratingSubscriber == s) { + std::cerr << "Error: WebSocket must not subscribe or unsubscribe to topics while iterating its topics!" << std::endl; + std::terminate(); } } - /* Publishes to all matching topics and wildcards. Returns whether at least one topic was a match. */ - bool publish(Topic *iterator, size_t start, size_t stop, std::string_view topic, std::pair message) { + /* Warning: does NOT unlink from drainableSubscribers or modify next, prev. */ + void drainImpl(Subscriber *s) { + /* Before we call cb we need to make sure this subscriber will not report needsDrainage() + * since WebSocket::send will call drain from within the cb in that case.*/ + int numMessageIndices = s->numMessageIndices; + s->numMessageIndices = 0; - /* Whether we matched with at least one topic */ - bool didMatch = false; + /* Then we emit cb */ + for (int i = 0; i < numMessageIndices; i++) { + T &outgoingMessage = outgoingMessages[s->messageIndices[i]]; - /* Iterate over all segments in given topic */ - for (; stop != std::string::npos; start = stop + 1) { - stop = topic.find('/', start); - std::string_view segment = topic.substr(start, stop - start); + int flags = (i == numMessageIndices - 1) ? LAST : 0; - /* It is very important to disallow wildcards when publishing. - * We will not catch EVERY misuse this lazy way, but enough to hinder - * explosive recursion. - * Terminating wildcards MAY still get triggered along the way, if for - * instace the error is found late while iterating the topic segments. */ - if (segment.length() == 1) { - if (segment[0] == '+' || segment[0] == '#') { - /* "Fail" here, but not necessarily for the entire publish */ - return didMatch; - } + /* Returning true will stop drainage short (such as when backpressure is too high) */ + if (cb(s, outgoingMessage, (IteratorFlags)(flags | (i == 0 ? FIRST : 0)))) { + break; } - - /* Do we have a terminating wildcard child? */ - if (iterator->terminatingWildcardChild) { - - /* Add this topic to triggered */ - if (!iterator->terminatingWildcardChild->triggered) { - /* If we already have 64 triggered topics make sure to drain it here */ - if (numTriggeredTopics == 64) { - drain(); - } - - triggeredTopics[numTriggeredTopics++] = iterator->terminatingWildcardChild; - iterator->terminatingWildcardChild->triggered = true; - } - - /* Above drain can reset messageId so we have to add new messages after */ - iterator->terminatingWildcardChild->messages[messageId] = message; - - didMatch = true; - } - - /* Do we have a wildcard child? */ - if (iterator->wildcardChild) { - didMatch |= publish(iterator->wildcardChild, stop + 1, stop, topic, message); - } - - std::map::iterator it = iterator->children.find(segment); - if (it == iterator->children.end()) { - /* Stop trying to match by exact string */ - return didMatch; - } - - iterator = it->second; } - - /* If we went all the way we matched exactly */ - - /* Add this topic to triggered */ - if (!iterator->triggered) { - /* If we already have 64 triggered topics make sure to drain it here */ - if (numTriggeredTopics == 64) { - drain(); - } - - triggeredTopics[numTriggeredTopics++] = iterator; - iterator->triggered = true; - } - - /* Above drain can change messageId to 0, so we put the message after */ - iterator->messages[messageId] = message; - - /* We obviously matches exactly here */ - return true; } public: - TopicTree(std::function cb) { - this->cb = cb; + TopicTree(std::function cb) : cb(cb) { + } - ~TopicTree() { - delete root; + /* Returns nullptr if not found */ + Topic *lookupTopic(std::string_view topic) { + auto it = topics.find(topic); + if (it == topics.end()) { + return nullptr; + } + return it->second.get(); } - /* This is part of the fast path, so should be optimal */ - std::vector &getSenderFor(Subscriber *s) { - static thread_local std::vector emptyVector; + /* Subscribe fails if we already are subscribed */ + bool subscribe(Subscriber *s, std::string_view topic) { + /* Notify user that they are doing something wrong here */ + checkIteratingSubscriber(s); - auto it = senderHoles.find(s); - if (it != senderHoles.end()) { - return it->second; + /* Lookup or create new topic */ + Topic *topicPtr = lookupTopic(topic); + if (!topicPtr) { + Topic *newTopic = new Topic(topic); + topics.insert({std::string_view(newTopic->name.data(), newTopic->name.length()), std::unique_ptr(newTopic)}); + topicPtr = newTopic; } - return emptyVector; + /* Insert us in topic, insert topic in us */ + auto [it, inserted] = s->topics.insert(topicPtr); + if (!inserted) { + return false; + } + topicPtr->insert(s); + + /* Success */ + return true; } - /* Returns number of subscribers after the call and whether or not we were successful in subscribing */ - std::pair subscribe(std::string_view topic, Subscriber *subscriber, bool nonStrict = false) { - /* Start iterating from the root */ - Topic *iterator = root; + /* Returns ok, last */ + std::pair unsubscribe(Subscriber *s, std::string_view topic) { + /* Notify user that they are doing something wrong here */ + checkIteratingSubscriber(s); - /* Traverse the topic, inserting a node for every new segment separated by / */ - for (size_t start = 0, stop = 0; stop != std::string::npos; start = stop + 1) { - stop = topic.find('/', start); - std::string_view segment = topic.substr(start, stop - start); + /* Lookup topic */ + Topic *topicPtr = lookupTopic(topic); + if (!topicPtr) { + /* If the topic doesn't exist we are assumed to still be subscribers of something */ + return {false, false}; + } - auto lb = iterator->children.lower_bound(segment); + /* Erase from our list first */ + if (s->topics.erase(topicPtr) == 0) { + return {false, false}; + } - if (lb != iterator->children.end() && !(iterator->children.key_comp()(segment, lb->first))) { - iterator = lb->second; + /* Remove us from topic */ + topicPtr->erase(s); + + /* If there is no subscriber to this topic, remove it */ + if (!topicPtr->size()) { + /* Unique_ptr deletes the topic */ + topics.erase(topic); + } + + /* If we don't hold any topics we are to be freed altogether */ + return {true, topics.size() == 0}; + } + + /* Factory function for creating a Subscriber */ + Subscriber *createSubscriber() { + return new Subscriber(); + } + + /* This is used to end a Subscriber, before freeing it */ + void freeSubscriber(Subscriber *s) { + + /* I guess we call this one even if we are not subscribers */ + if (!s) { + return; + } + + /* For all topics, unsubscribe */ + for (Topic *topicPtr : s->topics) { + /* If we are the last subscriber, simply remove the whole topic */ + if (topicPtr->size() == 1) { + topics.erase(topicPtr->name); } else { - /* Allocate and insert new node */ - Topic *newTopic = new Topic; - newTopic->parent = iterator; - newTopic->name = new char[segment.length()]; - newTopic->length = segment.length(); - newTopic->terminatingWildcardChild = nullptr; - newTopic->wildcardChild = nullptr; - memcpy(newTopic->name, segment.data(), segment.length()); - - /* Set fullname as parent's name plus our name */ - newTopic->fullName.reserve(newTopic->parent->fullName.length() + 1 + segment.length()); - - /* Only append parent's name if parent is not root */ - if (newTopic->parent != root) { - newTopic->fullName.append(newTopic->parent->fullName); - newTopic->fullName.append("/"); - } - - newTopic->fullName.append(segment); - - /* For simplicity we do insert wildcards with text */ - iterator->children.insert(lb, {std::string_view(newTopic->name, segment.length()), newTopic}); - - /* Store fast lookup to wildcards */ - if (segment.length() == 1) { - /* If this segment is '+' it is a wildcard */ - if (segment[0] == '+') { - iterator->wildcardChild = newTopic; - } - /* If this segment is '#' it is a terminating wildcard */ - if (segment[0] == '#') { - iterator->terminatingWildcardChild = newTopic; - } - } - - iterator = newTopic; + /* Otherwise just remove us */ + topicPtr->erase(s); } } - /* If this topic is triggered, drain the tree before we join */ - if (iterator->triggered) { - if (!nonStrict) { - drain(); - } - } - - /* Add socket to Topic's Set */ - auto [it, inserted] = iterator->subs.insert(subscriber); - - /* Add Topic to list of subscriptions only if we weren't already subscribed */ - if (inserted) { - subscriber->subscriptions.push_back(iterator); - return {(unsigned int) iterator->subs.size(), true}; - } - return {(unsigned int) iterator->subs.size(), false}; + delete s; } - bool 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); - } - - auto ret = publish(root, 0, 0, topic, message); - /* MessageIDs are reset on drain - this should be fine since messages itself are cleared on drain */ - messageId++; - return ret; - } - - /* Returns a pair of numSubscribers after operation, and whether we were subscribed prior */ - std::pair unsubscribe(std::string_view topic, Subscriber *subscriber, bool nonStrict = false) { - /* Subscribers are likely to have very few subscriptions (20 or fewer) */ - if (subscriber) { - /* Lookup exact Topic ptr from string */ - Topic *iterator = root; - for (size_t start = 0, stop = 0; stop != std::string::npos; start = stop + 1) { - stop = topic.find('/', start); - std::string_view segment = topic.substr(start, stop - start); - - std::map::iterator it = iterator->children.find(segment); - if (it == iterator->children.end()) { - /* This topic does not even exist */ - return {0, false}; - } - - iterator = it->second; + /* Mainly used by WebSocket::send to drain one socket before sending */ + void drain(Subscriber *s) { + /* The list is undefined and cannot be touched unless needsDrainage(). */ + if (s->needsDrainage()) { + /* This function differs from drainImpl by properly unlinking + * the subscriber from drainableSubscribers. drainImpl does not. */ + if (s->prev) { + s->prev->next = s->next; } - - /* Is this topic locked? If so, we cannot unsubscribe from it */ - if (iterator->locked) { - return {iterator->subs.size(), false}; + if (s->next) { + s->next->prev = s->prev; } - - /* Try and remove this topic from our list */ - for (auto it = subscriber->subscriptions.begin(); it != subscriber->subscriptions.end(); it++) { - if (*it == iterator) { - /* If this topic is triggered, drain the tree before we leave */ - if (iterator->triggered) { - if (!nonStrict) { - drain(); - } - } - - /* Remove topic ptr from our list */ - subscriber->subscriptions.erase(it); - - /* Remove us from Topic's subs */ - iterator->subs.erase(subscriber); - unsigned int numSubscribers = (unsigned int) iterator->subs.size(); - trimTree(iterator); - return {numSubscribers, true}; - } + /* If we are the head, then we also need to reset the head */ + if (drainableSubscribers == s) { + drainableSubscribers = nullptr; } - } - return {0, false}; - } - - /* Can be called with nullptr, ignore it then */ - void unsubscribeAll(Subscriber *subscriber, bool mayFlush = true) { - if (subscriber) { - for (Topic *topic : subscriber->subscriptions) { - - /* We do not want to flush when closing a socket, it makes no sense to do so */ - - /* If this topic is triggered, drain the tree before we leave */ - if (mayFlush && topic->triggered) { - /* Never mind nonStrict here (yet?) */ - drain(); - } - - /* Remove us from the topic's set */ - topic->subs.erase(subscriber); - trimTree(topic); - } - /* If we are sender with outstanding holes, remove them now that we are dead */ - senderHoles.erase(subscriber); - - subscriber->subscriptions.clear(); + + /* This one always resets needsDrainage before it calls any cb's. + * Otherwise we would stackoverflow when sending after publish but before drain. */ + drainImpl(s); } } - /* Drain the tree by emitting what to send with every Subscriber */ - /* Better name would be commit() and making it public so that one can commit and shutdown, etc */ + /* Called everytime we call send, to drain published messages so to sync outgoing messages */ void drain() { - - /* Do nothing if nothing to send */ - if (!numTriggeredTopics) { - return; - } - - /* bug fix: Filter triggered topics without subscribers */ - int numFilteredTriggeredTopics = 0; - for (int i = 0; i < numTriggeredTopics; i++) { - if (triggeredTopics[i]->subs.size()) { - triggeredTopics[numFilteredTriggeredTopics++] = triggeredTopics[i]; - } else { - /* If we no longer have any subscribers, yet still keep this Topic alive (parent), - * make sure to clear its potential messages. */ - triggeredTopics[i]->messages.clear(); - triggeredTopics[i]->triggered = false; + if (drainableSubscribers) { + /* Drain one socket a time */ + for (Subscriber *s = drainableSubscribers; s; s = s->next) { + /* Instead of unlinking every single subscriber, we just leave the list undefined + * and reset drainableSubscribers ptr below. */ + drainImpl(s); } + /* Drain always clears drainableSubscribers and outgoingMessages */ + drainableSubscribers = nullptr; + outgoingMessages.clear(); } - numTriggeredTopics = numFilteredTriggeredTopics; - - if (!numTriggeredTopics) { - senderHoles.clear(); - messageId = 0; - return; - } - - /* bug fix: update min, as the one tracked via subscribe gets invalid as you unsubscribe */ - min = (Subscriber *)UINTPTR_MAX; - for (int i = 0; i < numTriggeredTopics; i++) { - if ((triggeredTopics[i]->subs.size()) && (min > *triggeredTopics[i]->subs.begin())) { - min = *triggeredTopics[i]->subs.begin(); - } - } - - /* Check if we really have any sockets still */ - if (min != (Subscriber *)UINTPTR_MAX) { - - /* Up to 64 triggered Topics per batch */ - std::map intersectionCache; - - /* Loop over these here */ - std::set::iterator it[64]; - std::set::iterator end[64]; - for (int i = 0; i < numTriggeredTopics; i++) { - it[i] = triggeredTopics[i]->subs.begin(); - end[i] = triggeredTopics[i]->subs.end(); - } - - /* Empty all sets from unique subscribers */ - for (int nonEmpty = numTriggeredTopics; nonEmpty; ) { - - Subscriber *nextMin = (Subscriber *)UINTPTR_MAX; - - /* The message sets relevant for this intersection */ - std::map> *perSubscriberIntersectingTopicMessages[64]; - int numPerSubscriberIntersectingTopicMessages = 0; - - uint64_t intersection = 0; - - for (int i = 0; i < numTriggeredTopics; i++) { - if ((it[i] != end[i]) && (*it[i] == min)) { - - /* Mark this intersection */ - intersection |= ((uint64_t)1 << i); - perSubscriberIntersectingTopicMessages[numPerSubscriberIntersectingTopicMessages++] = &triggeredTopics[i]->messages; - - it[i]++; - if (it[i] == end[i]) { - nonEmpty--; - } - else { - if (nextMin > *it[i]) { - nextMin = *it[i]; - } - } - } - else { - /* We need to lower nextMin to us, in the case of min being the last in a set */ - if ((it[i] != end[i]) && (nextMin > *it[i])) { - nextMin = *it[i]; - } - } - } - - /* Generate cache for intersection */ - if (intersectionCache[intersection].dataChannels.first.length() == 0) { - - /* Build the union in order without duplicates */ - std::map> complete; - for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) { - complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end()); - } - - /* Create the linear cache, {inflated, deflated} */ - Intersection res; - for (auto &p : complete) { - 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)); - } - else { - cb(min, intersectionCache[intersection]); - } - - min = nextMin; - } - } - - /* Clear messages of triggered Topics */ - for (int i = 0; i < numTriggeredTopics; i++) { - triggeredTopics[i]->messages.clear(); - triggeredTopics[i]->triggered = false; - } - numTriggeredTopics = 0; - senderHoles.clear(); - messageId = 0; } + + /* Linear in number of affected subscribers */ + bool publish(Subscriber *sender, std::string_view topic, T &&message) { + /* Do we even have this topic? */ + auto it = topics.find(topic); + if (it == topics.end()) { + return false; + } + + /* If we have more than 65k messages we need to drain every socket. */ + if (outgoingMessages.size() == UINT16_MAX) { + /* If there is a socket that is currently corked, this will be ugly as all sockets will drain + * to their own backpressure */ + drain(); + } + + /* For all subscribers in topic */ + for (Subscriber *s : *it->second) { + + /* If we are sender then ignore us */ + if (sender != s) { + + /* If we already have too many outgoing messages on this subscriber, drain it now */ + if (s->numMessageIndices == 32) { + /* This one does not need to check needsDrainage here but still does. */ + drain(s); + } + + /* Finally we can continue */ + s->messageIndices[s->numMessageIndices++] = (uint16_t)outgoingMessages.size(); + /* First message adds subscriber to list of drainable subscribers */ + if (s->numMessageIndices == 1) { + /* Insert us in the head of drainable subscribers */ + s->next = drainableSubscribers; + s->prev = nullptr; + if (s->next) { + s->next->prev = s; + } + drainableSubscribers = s; + } + } + } + + /* Push this message and return with success */ + outgoingMessages.emplace_back(message); + return true; + } + }; } diff --git a/src/WebSocket.h b/src/WebSocket.h index 6e878e0..5d4aac7 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -89,6 +89,13 @@ public: return DROPPED; } + /* If we are subscribers and have messages to drain we need to drain them here to stay synced */ + WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData(); + if (webSocketData->subscriber) { + /* This will call back into us, send. */ + webSocketContextData->topicTree.drain(webSocketData->subscriber); + } + /* Transform the message to compressed domain if requested */ if (compress) { WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData(); @@ -180,8 +187,7 @@ public: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber, false); - delete webSocketData->subscriber; + webSocketContextData->topicTree.freeSubscriber(webSocketData->subscriber); webSocketData->subscriber = nullptr; } @@ -201,7 +207,7 @@ public: } /* Subscribe to a topic according to MQTT rules and syntax. Returns success */ - /*std::pair*/ bool subscribe(std::string_view topic, bool nonStrict = false) { + bool subscribe(std::string_view topic, bool = false) { WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); @@ -209,15 +215,19 @@ public: /* Make us a subscriber if we aren't yet */ WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); if (!webSocketData->subscriber) { - webSocketData->subscriber = new Subscriber(this); + webSocketData->subscriber = webSocketContextData->topicTree.createSubscriber(); + webSocketData->subscriber->user = this; } /* Cannot return numSubscribers as this is only for this particular websocket context */ - return webSocketContextData->topicTree.subscribe(topic, webSocketData->subscriber, nonStrict).second; + webSocketContextData->topicTree.subscribe(webSocketData->subscriber, topic); + + /* Subscribe always succeeds */ + return true; } /* Unsubscribe from a topic, returns true if we were subscribed. */ - /*std::pair*/ bool unsubscribe(std::string_view topic, bool nonStrict = false) { + bool unsubscribe(std::string_view topic, bool = false) { WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); @@ -225,7 +235,15 @@ public: WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); /* Cannot return numSubscribers as this is only for this particular websocket context */ - return webSocketContextData->topicTree.unsubscribe(topic, webSocketData->subscriber, nonStrict).second; + auto [ok, last] = webSocketContextData->topicTree.unsubscribe(webSocketData->subscriber, topic); + + /* Free us as subscribers if we unsubscribed from our last topic */ + if (ok && last) { + webSocketContextData->topicTree.freeSubscriber(webSocketData->subscriber); + webSocketData->subscriber = nullptr; + } + + return ok; } /* Returns whether this socket is subscribed to the specified topic */ @@ -239,30 +257,34 @@ public: return false; } - Topic *t = webSocketContextData->topicTree.lookupTopic(topic); - if (t) { - return t->subs.find(webSocketData->subscriber) != t->subs.end(); + Topic *topicPtr = webSocketContextData->topicTree.lookupTopic(topic); + if (!topicPtr) { + return false; } - return false; + return topicPtr->count(webSocketData->subscriber); } /* Iterates all topics of this WebSocket. Every topic is represented by its full name. * Can be called in close handler. It is possible to modify the subscription list while * inside the callback ONLY IF not modifying the topic passed to the callback. * Topic names are valid only for the duration of the callback. */ - void iterateTopics(MoveOnlyFunction cb) { + void iterateTopics(MoveOnlyFunction cb) { + WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, + (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) + ); + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); - if (webSocketData->subscriber) { - for (Topic *t : webSocketData->subscriber->subscriptions) { - /* Lock this topic so that nobody may unsubscribe from it during this callback */ - t->locked = true; + /* Lock this subscriber for unsubscription / subscription */ + webSocketContextData->topicTree.iteratingSubscriber = webSocketData->subscriber; - cb(t->fullName/*, (unsigned int) t->subs.size()*/); - - t->locked = false; + for (Topic *topicPtr : webSocketData->subscriber->topics) { + cb({topicPtr->name.data(), topicPtr->name.length()}); } + + /* Unlock subscriber */ + webSocketContextData->topicTree.iteratingSubscriber = nullptr; } } @@ -282,13 +304,13 @@ public: } /* Publish as sender, does not receive its own messages even if subscribed to relevant topics */ - bool success = webSocketContextData->publish(topic, message, opCode, compress, webSocketData->subscriber); + bool success = webSocketContextData->topicTree.publish(webSocketData->subscriber, topic, {std::string(message), opCode, compress}); /* Loop over all websocket contexts for this App */ if (success) { /* Success is really only determined by the first publish. We must be subscribed to the topic. */ for (auto *adjacentWebSocketContextData : webSocketContextData->adjacentWebSocketContextDatas) { - adjacentWebSocketContextData->publish(topic, message, opCode, compress); + adjacentWebSocketContextData->topicTree.publish(nullptr, topic, {std::string(message), opCode, compress}); } } diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 6a14a07..b55182e 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -249,8 +249,7 @@ private: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber, false); - delete webSocketData->subscriber; + webSocketContextData->topicTree.freeSubscriber(webSocketData->subscriber); webSocketData->subscriber = nullptr; } diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index a14ba0c..5585a1e 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -38,18 +38,15 @@ template struct WebSocket; template struct WebSocketContextData { private: - /* Used for prepending unframed messages when using dedicated compressors */ - struct MessageMetadata { - unsigned int length; - OpCode opCode; - bool compress; - /* Undefined init of all members */ - MessageMetadata() {} - MessageMetadata(unsigned int length, OpCode opCode, bool compress) - : length(length), opCode(opCode), compress(compress) {} - }; public: + /* Type queued up when publishing */ + struct TopicTreeMessage { + std::string message; + OpCode opCode; + bool compress; + }; + /* All WebSocketContextData holds a list to all other WebSocketContextData in this app. * We cannot type it USERDATA since different WebSocketContextData can have different USERDATA. */ std::vector *> adjacentWebSocketContextDatas; @@ -79,7 +76,7 @@ public: std::pair idleTimeoutComponents; /* Each websocket context has a topic tree for pub/sub */ - TopicTree topicTree; + TopicTree topicTree; /* This is run once on start-up */ void calculateIdleTimeoutCompnents(unsigned short idleTimeout) { @@ -100,120 +97,39 @@ public: Loop::get()->removePreHandler(this); } - WebSocketContextData() : topicTree([this](Subscriber *s, Intersection &intersection) -> int { + WebSocketContextData() : topicTree([](Subscriber *s, TopicTreeMessage &message, auto flags) { + /* Subscriber's user is the socket */ + auto *ws = (WebSocket *) s->user; - /* 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; - - /* If we are corked, do not uncork - otherwise if we cork in here, uncork before leaving */ - bool wasCorked = asyncSocket->isCorked(); - - /* Do we even have room for potential data? */ - if (!maxBackpressure || asyncSocket->getBufferedAmount() < maxBackpressure) { - - /* Roll over all our segments */ - intersection.forSubscriber(topicTree.getSenderFor(s), [asyncSocket, this](std::pair data, bool fin) { - - /* 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(); - } - - /* Pick uncompressed data track */ - std::string_view selectedData = data.first; - - /* Are we using compression? Fine, pick the compressed data track */ - WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); - if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { - - /* This is used for both shared and dedicated paths */ - selectedData = data.second; - - /* 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(); - } - - 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); - - /* 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; - } - } - - /* 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) { - auto *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) asyncSocket); - webSocketData->hasTimedOut = false; - asyncSocket->timeout(this->idleTimeoutComponents.first); - } - } - }); - } - - /* 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) { - auto *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) asyncSocket); - webSocketData->hasTimedOut = false; - asyncSocket->timeout(this->idleTimeoutComponents.first); - } + /* If this is the first message we try and cork */ + bool needsUncork = false; + if (flags & TopicTree::IteratorFlags::FIRST) { + if (ws->canCork() && !ws->isCorked()) { + ((AsyncSocket *)ws)->cork(); + needsUncork = true; } } - /* 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); + /* If we ever overstep maxBackpresure, exit immediately */ + if (WebSocket::SendStatus::DROPPED == ws->send(message.message, message.opCode, message.compress)) { + + if (needsUncork) { + ((AsyncSocket *)ws)->uncork(); + } + /* Stop draining */ + return true; } - /* Reserved, unused */ - return 0; + /* If this is the last message we uncork if we are corked */ + if (flags & TopicTree::IteratorFlags::LAST) { + /* We should not uncork in all cases? */ + if (needsUncork) { + ((AsyncSocket *)ws)->uncork(); + } + } + + /* Success */ + return false; }) { /* We empty for both pre and post just to make sure */ Loop::get()->addPostHandler(this, [this](Loop */*loop*/) { @@ -226,71 +142,6 @@ public: topicTree.drain(); }); } - - /* Helper for topictree publish, common path from app and ws */ - bool publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress, Subscriber *sender = nullptr) { - bool didMatch = false; - - /* 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); - - /* If compression is disabled */ - if (compression == DISABLED) { - /* Leave second field empty as nobody will ever read it */ - didMatch |= 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) { - /* Shared compression mode publishes compressed, framed data */ - if (compression == SHARED_COMPRESSOR) { - /* Loop data holds shared compressor */ - LoopData *loopData = (LoopData *) us_loop_ext((us_loop_t *) Loop::get()); - - /* Compress it */ - std::string_view compressedMessage = loopData->deflationStream->deflate(loopData->zlibContext, message, true); - - /* Frame it */ - char *dst_compressed = (char *) malloc(protocol::messageFrameSize(compressedMessage.size())); - 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 */ - didMatch |= 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); - } else { - /* Dedicated compression mode publishes metadata + unframed uncompressed data */ - char *dst_compressed = (char *) malloc(message.length() + sizeof(MessageMetadata)); - - MessageMetadata mm( - (unsigned int) message.length(), - opCode, - compress - ); - - memcpy(dst_compressed, (char *) &mm, sizeof(MessageMetadata)); - memcpy(dst_compressed + sizeof(MessageMetadata), message.data(), message.length()); - - /* Interpretation of compressed data depends on what compressor we use */ - didMatch |= 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. */ - didMatch |= topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}, sender); - } - } - - ::free(dst); - - return didMatch; - } }; } diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 35245db..da5e3ca 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -13,37 +13,20 @@ void testCorrectness() { std::cout << "TestCorrectness" << std::endl; - uWS::TopicTree *topicTree; - std::map> expectedResult; - std::map> actualResult; + uWS::TopicTree *topicTree; + std::map expectedResult; + std::map actualResult; - topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { + topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, std::string &message, auto flags) { - /* How many bytes we have in first data channel at time we get fin = true */ - unsigned int finAt = 0; + actualResult[s] += message; - 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; + /* Success */ + return false; }); - uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); - uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); + uWS::Subscriber *s1 = topicTree->createSubscriber(); + uWS::Subscriber *s2 = topicTree->createSubscriber(); /* Make sure s1 < s2 (for debugging) */ if (s2 < s1) { @@ -53,35 +36,35 @@ void testCorrectness() { } /* Publish to topic3 - nobody should see this */ - topicTree->publish("topic3", {std::string_view("Nobody"), std::string_view("should see")}, nullptr); + topicTree->publish(nullptr, "topic3", "Nobody should see"); /* Subscribe s1 to topic3 - s1 should not see above message */ - topicTree->subscribe("topic3", s1); + topicTree->subscribe(s1, "topic3"); /* 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); + topicTree->publish(s1, "topic3", "Nobody should see"); /* Subscribe s2 to topic3 - should not get any message */ - topicTree->subscribe("topic3", s2); + topicTree->subscribe(s2, "topic3"); /* Publish to topic3 without sender - both should see */ - topicTree->publish("topic3", {std::string_view("Both"), std::string_view("should see")}, nullptr); + topicTree->publish(nullptr, "topic3", "Both should see"); /* 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); + topicTree->publish(s2, "topic3", "s1 should see, not 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); + topicTree->publish(s1, "topic3", "s2 should see, not 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); + topicTree->publish(nullptr, "topic3", "Again, both should see this as well"); // 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"}} + {s1, "Both should sees1 should see, not s2Again, both should see this as well"}, + {s2, "Both should sees2 should see, not s1Again, both should see this as well"} }; /* Compare result with expected result for every subscriber */ @@ -89,23 +72,15 @@ void testCorrectness() { 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; + if (p.second != actualResult[p.first]) { + std::cout << "ERROR: <" << actualResult[p.first] << "> should be <" << p.second << ">" << std::endl; exit(1); } } /* Release resources */ - topicTree->unsubscribeAll(s1); - topicTree->unsubscribeAll(s2); - - delete s1; - delete s2; + topicTree->freeSubscriber(s1); + topicTree->freeSubscriber(s2); delete topicTree; } @@ -113,37 +88,20 @@ void testCorrectness() { void testBugReport() { std::cout << "TestBugReport" << std::endl; - uWS::TopicTree *topicTree; - std::map> expectedResult; - std::map> actualResult; + uWS::TopicTree *topicTree; + std::map expectedResult; + std::map actualResult; - topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { + topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, std::string &message, auto flags) { - /* How many bytes we have in first data channel at time we get fin = true */ - unsigned int finAt = 0; + actualResult[s] += message; - 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; + /* Success */ + return false; }); - uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); - uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); + uWS::Subscriber *s1 = topicTree->createSubscriber(); + uWS::Subscriber *s2 = topicTree->createSubscriber(); /* Make sure s1 < s2 (for debugging) */ if (s2 < s1) { @@ -153,21 +111,21 @@ void testBugReport() { } /* Each subscriber to its own topic */ - topicTree->subscribe("b1", s1); - topicTree->subscribe("b2", s2); + topicTree->subscribe(s1, "b1"); + topicTree->subscribe(s2, "b2"); /* This one should send b2 to s2 */ - topicTree->publish("b1", {std::string_view("b1"), std::string_view("b1")}, s1); - topicTree->publish("b2", {std::string_view("b2"), std::string_view("b2")}, s1); + topicTree->publish(s1, "b1", "b1"); + topicTree->publish(s1, "b2", "b2"); /* This one should send b1 to s1 */ - topicTree->publish("b1", {std::string_view("b1"), std::string_view("b1")}, s2); - topicTree->publish("b2", {std::string_view("b2"), std::string_view("b2")}, s2); + topicTree->publish(s2, "b1", "b1"); + topicTree->publish(s2, "b2", "b2"); /* Fill out expectedResult */ expectedResult = { - {s1, {"b1", "b1"}}, - {s2, {"b2", "b2"}} + {s1, "b1"}, + {s2, "b2"} }; /* Compare result with expected result for every subscriber */ @@ -175,23 +133,15 @@ void testBugReport() { 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; + if (p.second != actualResult[p.first]) { + std::cout << "ERROR: <" << actualResult[p.first] << "> should be <" << p.second << ">" << std::endl; exit(1); } } /* Release resources */ - topicTree->unsubscribeAll(s1); - topicTree->unsubscribeAll(s2); - - delete s1; - delete s2; + topicTree->freeSubscriber(s1); + topicTree->freeSubscriber(s2); delete topicTree; } @@ -199,48 +149,30 @@ void testBugReport() { void testReorderingv19() { std::cout << "TestReorderingv19" << std::endl; - uWS::TopicTree *topicTree; - std::map> expectedResult; - std::map> actualResult; + uWS::TopicTree *topicTree; + std::map expectedResult; + std::map actualResult; - topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { + topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, std::string &message, auto flags) { - /* How many bytes we have in first data channel at time we get fin = true */ - unsigned int finAt = 0; + actualResult[s] += message; - 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; + /* Success */ + return false; }); - uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); + uWS::Subscriber *s1 = topicTree->createSubscriber(); /* Subscribe to 100 topics */ for (int i = 0; i < 100; i++) { - topicTree->subscribe(std::to_string(i), s1); + topicTree->subscribe(s1, std::to_string(i)); } /* Publish to 100 topics in order with messages in order */ for (int i = 0; i < 100; i++) { - topicTree->publish(std::to_string(i), {std::to_string(i) + ",", std::to_string(i) + ","}, nullptr); + topicTree->publish(nullptr, std::to_string(i), std::to_string(i) + ","); - expectedResult[s1].first.append(std::to_string(i) + ","); - expectedResult[s1].second.append(std::to_string(i) + ","); + expectedResult[s1].append(std::to_string(i) + ","); } /* Compare result with expected result for every subscriber */ @@ -248,21 +180,14 @@ void testReorderingv19() { 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; + if (p.second != actualResult[p.first]) { + std::cout << "ERROR: <" << actualResult[p.first] << "> should be <" << p.second << ">" << std::endl; exit(1); } } /* Release resources */ - topicTree->unsubscribeAll(s1); - - delete s1; + topicTree->freeSubscriber(s1); delete topicTree; }