From 6c7844d65f9d74fcdc99a7970f29ebf88f5bb45a Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Mon, 7 Oct 2019 11:58:02 +0200 Subject: [PATCH] Fix up unsubscribeAll --- src/TopicTree.h | 195 ------------------------------------- src/TopicTreeDraft.h | 181 +++++++++++++++++++--------------- src/WebSocket.h | 11 ++- src/WebSocketContext.h | 2 +- src/WebSocketContextData.h | 2 +- src/WebSocketData.h | 7 ++ 6 files changed, 122 insertions(+), 276 deletions(-) delete mode 100644 src/TopicTree.h diff --git a/src/TopicTree.h b/src/TopicTree.h deleted file mode 100644 index 5608df3..0000000 --- a/src/TopicTree.h +++ /dev/null @@ -1,195 +0,0 @@ -/* - * Authored by Alex Hultman, 2018-2019. - * Intellectual property of third-party. - - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - - * http://www.apache.org/licenses/LICENSE-2.0 - - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/* Every WebSocketContext holds one TopicTree */ -#include "Loop.h" -#include "AsyncSocket.h" - -#ifndef UWS_TOPICTREE_H -#define UWS_TOPICTREE_H - -#include -#include -#include -#include - -// todo: obviously this module is WIP - -namespace uWS { - - // publishing to a node, then another node, then another node should prioritize draining that way - // sending and publishing will interleave undefined, they are separate streams - -struct TopicTree { -private: - struct Node : std::map { - /* Add and/or lookup node from topic */ - Node *get(std::string topic) { - std::pair::iterator, bool> p = insert({topic, nullptr}); - if (p.second) { - return p.first->second = new Node; - } else { - return p.first->second; - } - } - - /* Subscribers have int backpressureOffset */ - std::set subscribers; - - /* Current shared message */ - std::string sharedMessage; - - /* Backpressure is stored linearly up to a limit */ - std::string backpressure; - unsigned int backpressureOffset = 0; - - - } topicToNode; - - std::map> socketToNodeList; - - /* Nodes that hold something to send this iteration */ - std::set pubNodes; - - /* Settings */ - bool mergePublishedMessages = false; - - /* Where we store prepared messages to send */ - //std::string preparedMessage; - -public: - - ~TopicTree() { - /* We have a few leaks here, I think */ - } - - TopicTree() { - /* Dynamically hook us up with the Loop post handler */ - Loop::get()->addPostHandler([this](Loop *loop) { - - if (!pubNodes.size()) { - return; - } - - /* We say that all senders get their own message as well, for now being */ - - // if a node is very likely used together with another node? - - for (Node *topicNode : pubNodes) { - for (auto /*[*/ws/*, valid]*/ : topicNode->subscribers) { - AsyncSocket *asyncSocket = (AsyncSocket *) ws; // assumes non-SSL for now - - /* Writing optionally raw data */ - auto [written, failed] = asyncSocket->write(topicNode->sharedMessage.data(), topicNode->sharedMessage.length(), true, 0); - - /* We should probably reset timeout for a WebSocket getting something sent */ - - - /* Every subscriber to a topicNode will have int backpressure cursor to this room */ - - /* How far we wrote will be stored in the WebSocket's Pub/sub block and drained before any other sending (we need to fail sending if already sending pubsub) */ - - /* All messages not fully sent, will be stored in the topictree with an index so that websocket can refer to it by two index: what buffer, what offset */ - /* If total backpressure of the topictree is larger than a set limit we close all the slow receivers */ - - /* It is also possible to move topictree backpressure to the websockets themselves, if only one */ - } - - /* If not all sockets managed to send this message, move it to backpressure */ - topicNode->sharedMessage.clear(); - } - pubNodes.clear(); - - - }); - } - - /* WebSocket.subscribe will lookup the Loop and subscribe in its tree */ - void subscribe(std::string topic, void *connection, bool *valid) { - Node *curr = &topicToNode; - for (int i = 0; i < topic.length(); i++) { - int start = i; - while (topic[i] != '/' && i < topic.length()) { - i++; - } - curr = curr->get(topic.substr(start, i - start)); - } - curr->subscribers.insert(connection); - /* Only do this if we did not aleady exist */ - socketToNodeList[connection].push_back(curr); - } - - /* Unsubscribe from all subscriptions */ - void unsubscribeAll(void *connection) { - - for (Node *node : socketToNodeList[connection]) { - - /* Also make sure to update any backpressure here */ - - node->subscribers.erase(connection); - } - - socketToNodeList.erase(connection); - } - - /* WebSocket.publish looks up its tree and publishes to it */ - void publish(std::string topic, char *data, size_t length) { - Node *curr = &topicToNode; - for (int i = 0; i < topic.length(); i++) { - int start = i; - while (topic[i] != '/' && i < topic.length()) { - i++; - } - std::string path(topic.data() + start, i - start); - - // end wildcard consumes traversal - auto it = curr->find("#"); - if (it != curr->end()) { - curr = it->second; - //matches.push_back(curr); - curr->sharedMessage.append(data, length); - if (curr->subscribers.size()) { - pubNodes.insert(curr); - } - break; - } else { - it = curr->find(path); - if (it == curr->end()) { - it = curr->find("+"); - if (it != curr->end()) { - goto skip; - } - break; - } else { - skip: - curr = it->second; - if (i == topic.length()) { - //matches.push_back(curr); - curr->sharedMessage.append(data, length); - if (curr->subscribers.size()) { - pubNodes.insert(curr); - } - break; - } - } - } - } - } -}; -} - -#endif // UWS_TOPICTREE_H diff --git a/src/TopicTreeDraft.h b/src/TopicTreeDraft.h index 6ee839a..f950dec 100644 --- a/src/TopicTreeDraft.h +++ b/src/TopicTreeDraft.h @@ -1,3 +1,23 @@ +/* + * Authored by Alex Hultman, 2018-2019. + * Intellectual property of third-party. + + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + + * http://www.apache.org/licenses/LICENSE-2.0 + + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef UWS_TOPICTREE_H +#define UWS_TOPICTREE_H + #include #include #include @@ -5,13 +25,16 @@ #include #include #include +#include namespace uWS { /* A Subscriber is an extension of a socket */ struct Subscriber { - /* List of all our subscriptions (subscribersNextSubscription) */ - struct Subscription *subscriptions; + std::list subscriptions; + void *user; + + Subscriber(void *user) : user(user) {} }; struct Topic { @@ -40,24 +63,6 @@ struct Topic { std::set subs; }; -/* A Subscription is a link between Topic and Subscriber */ -struct Subscription { - /* The Topic we are subscribed to */ - Topic *topic; - - /* What subscriber we are linked to */ - Subscriber *subscriber; - - /* Backpressure relating to the Topic */ - int backpressure; - - /* Next subscription from subscriber point of view */ - Subscription *subscribersNextSubscription; - - /* Next subscription from topic point of view */ - Subscription *topicNextSubscription; -}; - struct TopicTree { private: std::function cb; @@ -194,6 +199,9 @@ public: /* Add socket to Topic's Set */ iterator->subs.insert(subscriber); + + /* Add Topic to list of subscriptions */ + subscriber->subscriptions.push_back(iterator); } void publish(std::string_view topic, std::string_view message) { @@ -207,89 +215,108 @@ public: } + /* Can be called with nullptr, ignore it then */ void unsubscribeAll(Subscriber *subscriber) { - for (Subscription *iterator = subscriber->subscriptions; iterator; iterator = iterator->subscribersNextSubscription) { - iterator->topic->subs.erase(subscriber); - trimTree(iterator->topic); + if (subscriber) { + for (Topic *topic : subscriber->subscriptions) { + topic->subs.erase(subscriber); + trimTree(topic); + } } } /* Drain the tree by emitting what to send with every Subscriber */ - void drain(/*std::function cb*/) { + void drain() { + /* Do nothing if nothing to send */ if (!numTriggeredTopics) { return; } - /* Up to 64 triggered Topics per batch */ - std::map intersectionCache; + /* Fast path for one topic (can also be used with heuristics) */ + if (numTriggeredTopics == -555555) { + /* Disabled */ + /*std::string res; + for (auto &p : triggeredTopics[0]->messages) { + res.append(p.second); + } - /* 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; ) { + for (Subscriber *s : triggeredTopics[0]->subs) { + cb(s, res); + }*/ + } else { - Subscriber *nextMin = (Subscriber *)UINTPTR_MAX; - - /* The message sets relevant for this intersection */ - std::map *perSubscriberIntersectingTopicMessages[64]; - int numPerSubscriberIntersectingTopicMessages = 0; - - uint64_t intersection = 0; + /* 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++) { - if ((it[i] != end[i]) && (*it[i] == min)) { + it[i] = triggeredTopics[i]->subs.begin(); + end[i] = triggeredTopics[i]->subs.end(); + } + + /* Empty all sets from unique subscribers */ + for (int nonEmpty = numTriggeredTopics; nonEmpty; ) { - /* Mark this intersection */ - intersection |= (1 << i); - perSubscriberIntersectingTopicMessages[numPerSubscriberIntersectingTopicMessages++] = &triggeredTopics[i]->messages; + Subscriber *nextMin = (Subscriber *)UINTPTR_MAX; - it[i]++; - if (it[i] == end[i]) { - nonEmpty--; + /* 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 |= (1 << i); + perSubscriberIntersectingTopicMessages[numPerSubscriberIntersectingTopicMessages++] = &triggeredTopics[i]->messages; + + it[i]++; + if (it[i] == end[i]) { + nonEmpty--; + } + else { + if (nextMin > *it[i]) { + nextMin = *it[i]; + } + } } else { - if (nextMin > *it[i]) { + /* 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]; } } } - 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].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 */ + std::string res; + for (auto &p : complete) { + res.append(p.second); + } + + cb(min, intersectionCache[intersection] = std::move(res)); } - } - - /* Generate cache for intersection */ - if (intersectionCache[intersection].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()); + else { + cb(min, intersectionCache[intersection]); } - /* Create the linear cache */ - std::string res; - for (auto &p : complete) { - res.append(p.second); - } - - cb(min, intersectionCache[intersection] = std::move(res)); - } - else { - cb(min, intersectionCache[intersection]); + min = nextMin; } - min = nextMin; } /* Clear messages of triggered Topics */ @@ -316,4 +343,6 @@ public: } }; -} \ No newline at end of file +} + +#endif \ No newline at end of file diff --git a/src/WebSocket.h b/src/WebSocket.h index 1cfb369..6fbc92d 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -128,7 +128,7 @@ public: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll((Subscriber *) this); + webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber); } /* Subscribe to a topic according to MQTT rules and syntax */ @@ -137,8 +137,13 @@ public: (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); - /* Fix this up */ - webSocketContextData->topicTree.subscribe(topic, (Subscriber *) this); + /* 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); + } + + webSocketContextData->topicTree.subscribe(topic, webSocketData->subscriber); } /* Publish a message to a topic according to MQTT rules and syntax */ diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 8dac930..57f5a51 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -237,7 +237,7 @@ private: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll((Subscriber *) s); + webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber); } /* Destruct in-placed data struct */ diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index bae3451..01ee514 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -48,7 +48,7 @@ struct WebSocketContextData { //std::cout << "Skickar data: " << data << " på sub: " << s << std::endl; - auto *asyncSocket = (AsyncSocket *) s; + auto *asyncSocket = (AsyncSocket *) s->user; asyncSocket->write(data.data(), data.length()); diff --git a/src/WebSocketData.h b/src/WebSocketData.h index 2ed3003..0f2e59d 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -41,6 +41,9 @@ private: /* We might have a dedicated compressor */ DeflationStream *deflationStream = nullptr; + + /* We could be a subscriber */ + Subscriber *subscriber = nullptr; public: WebSocketData(bool perMessageDeflate, bool slidingCompression, std::string &&backpressure) : AsyncSocketData(std::move(backpressure)), WebSocketState() { compressionStatus = perMessageDeflate ? ENABLED : DISABLED; @@ -55,6 +58,10 @@ public: if (deflationStream) { delete deflationStream; } + + if (subscriber) { + delete subscriber; + } } };