From f58a9e1e0dd8928ef706cc51f7f7041b83effe69 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Tue, 8 Oct 2019 03:55:27 +0200 Subject: [PATCH] Pass autobahn with pub/sub --- examples/BroadcastingEchoServer.cpp | 14 ++++++- src/TopicTreeDraft.h | 62 ++++++++++++++++++++++------- src/WebSocket.h | 9 +++-- src/WebSocketContext.h | 3 +- src/WebSocketContextData.h | 12 +++--- 5 files changed, 74 insertions(+), 26 deletions(-) diff --git a/examples/BroadcastingEchoServer.cpp b/examples/BroadcastingEchoServer.cpp index e071c66..f8d2225 100644 --- a/examples/BroadcastingEchoServer.cpp +++ b/examples/BroadcastingEchoServer.cpp @@ -1,5 +1,7 @@ #include "App.h" +struct us_listen_socket_t *listen_socket; + int main() { /* ws->getUserData returns one of these */ struct PerSocketData { @@ -10,7 +12,7 @@ int main() { uWS::App().ws("/*", { /* Settings */ .compression = uWS::SHARED_COMPRESSOR, - .maxPayloadLength = 16 * 1024, + .maxPayloadLength = 16 * 1024 * 1024, .idleTimeout = 10, /* Handlers */ .open = [](auto *ws, auto *req) { @@ -18,8 +20,15 @@ int main() { ws->subscribe("broadcast"); }, .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) { + /* Exit gracefully if we get a closedown message (ASAN debug) */ + if (message == "closedown") { + /* Bye bye */ + us_listen_socket_close(0, listen_socket); + ws->close(); + } + /* Simply broadcast every single message we get */ - ws->publish("broadcast", message/*, opCode*/); + ws->publish("broadcast", message, opCode); }, .drain = [](auto *ws) { /* Check getBufferedAmount here */ @@ -34,6 +43,7 @@ int main() { /* We automatically unsubscribe from any topic here */ } }).listen(9001, [](auto *token) { + listen_socket = token; if (token) { std::cout << "Listening on port " << 9001 << std::endl; } diff --git a/src/TopicTreeDraft.h b/src/TopicTreeDraft.h index f950dec..85e53c8 100644 --- a/src/TopicTreeDraft.h +++ b/src/TopicTreeDraft.h @@ -58,7 +58,7 @@ struct Topic { Topic *terminatingWildcardChild = nullptr; /* What we published */ - std::map messages; + std::map messages; std::set subs; }; @@ -70,7 +70,7 @@ private: Topic *root = new Topic; /* Global messageId for deduplication of overlapping topics and ordering between topics */ - int messageId = 0; + unsigned int messageId = 0; /* The triggered topics */ Topic *triggeredTopics[64]; @@ -89,7 +89,29 @@ private: parent->wildcardChild = nullptr; } } + /* Erase us from our parents set (wildcards also live here) */ parent->children.erase(std::string_view(topic->name, topic->length)); + + /* 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]; + } + } + + for (int i = 0; i < length; i++) { + triggeredTopics[i] = tmp[i]; + } + numTriggeredTopics = length; + } + + /* Free various memory for the node */ + delete [] topic->name; + delete topic; + if (parent != root) { trimTree(parent); } @@ -155,6 +177,10 @@ public: this->cb = cb; } + ~TopicTree() { + delete root; + } + void subscribe(std::string_view topic, Subscriber *subscriber) { /* Start iterating from the root */ Topic *iterator = root; @@ -222,10 +248,12 @@ public: topic->subs.erase(subscriber); trimTree(topic); } + subscriber->subscriptions.clear(); } } /* 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 */ void drain() { /* Do nothing if nothing to send */ @@ -233,18 +261,16 @@ public: return; } - /* 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); + /* 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(); } + } - for (Subscriber *s : triggeredTopics[0]->subs) { - cb(s, res); - }*/ - } else { + /* Check if we really have any sockets still */ + if (min != (Subscriber *)UINTPTR_MAX) { /* Up to 64 triggered Topics per batch */ std::map intersectionCache; @@ -263,7 +289,7 @@ public: Subscriber *nextMin = (Subscriber *)UINTPTR_MAX; /* The message sets relevant for this intersection */ - std::map *perSubscriberIntersectingTopicMessages[64]; + std::map *perSubscriberIntersectingTopicMessages[64]; int numPerSubscriberIntersectingTopicMessages = 0; uint64_t intersection = 0; @@ -297,7 +323,7 @@ public: if (intersectionCache[intersection].length() == 0) { /* Build the union in order without duplicates */ - std::map complete; + std::map complete; for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) { complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end()); } @@ -337,7 +363,13 @@ public: for (int i = 0; i < indentation; i++) { std::cout << " "; } - std::cout << std::string_view(p.second->name, p.second->length) << " = " << p.second->messages.size() << " publishes, " << p.second->subs.size() << " subscribers" << std::endl; + std::cout << std::string_view(p.second->name, p.second->length) << " = " << p.second->messages.size() << " publishes, " << p.second->subs.size() << " subscribers {"; + + for (auto &p : p.second->subs) { + std::cout << p << " referring to socket: " << p->user << ", "; + } + std::cout << "}" << std::endl; + print(p.second, indentation + 1); } } diff --git a/src/WebSocket.h b/src/WebSocket.h index 6fbc92d..97ef4cd 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -129,6 +129,8 @@ public: /* Make sure to unsubscribe from any pub/sub node at exit */ webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber); + delete webSocketData->subscriber; + webSocketData->subscriber = nullptr; } /* Subscribe to a topic according to MQTT rules and syntax */ @@ -147,16 +149,17 @@ public: } /* Publish a message to a topic according to MQTT rules and syntax */ - void publish(std::string_view topic, std::string_view message) { + void publish(std::string_view topic, std::string_view message, OpCode opCode = OpCode::TEXT, bool compress = false) { WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); /* We frame the message right here and only pass raw bytes to the pub/subber */ - char dst[1024]; - size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), OpCode::TEXT, message.length(), false); + char *dst = (char *) malloc(protocol::messageFrameSize(message.size())); + size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), opCode, message.length(), false); webSocketContextData->topicTree.publish(topic, std::string_view(dst, dst_length)); + free(dst); } }; diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 57f5a51..0e17ec9 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -225,7 +225,6 @@ private: /* Handle socket disconnections */ us_socket_context_on_close(SSL, getSocketContext(), [](auto *s) { - /* For whatever reason, if we already have emitted close event, do not emit it again */ WebSocketData *webSocketData = (WebSocketData *) (us_socket_ext(SSL, s)); if (!webSocketData->isShuttingDown) { @@ -238,6 +237,8 @@ private: /* Make sure to unsubscribe from any pub/sub node at exit */ webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber); + delete webSocketData->subscriber; + webSocketData->subscriber = nullptr; } /* Destruct in-placed data struct */ diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 01ee514..1fc117c 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -41,22 +41,24 @@ struct WebSocketContextData { size_t maxPayloadLength = 0; int idleTimeout = 0; + /* There needs to be a maxBackpressure which will force close everything over that limit */ + size_t maxBackpressure = 16 * 1024; + /* Each websocket context has a topic tree for pub/sub */ TopicTree topicTree; WebSocketContextData() : topicTree([](Subscriber *s, std::string_view data) -> int { - //std::cout << "Skickar data: " << data << " på sub: " << s << std::endl; - - + /* We rely on writing to regular asyncSockets */ auto *asyncSocket = (AsyncSocket *) s->user; asyncSocket->write(data.data(), data.length()); + /* Reserved, unused */ return 0; }) { - + /* bug: This should probably happen in both post and pre, esp for libuv */ Loop::get()->addPostHandler([this](Loop *loop) { - + /* Commit pub/sub batches every loop iteration */ topicTree.drain(); }); }