From 60939de843e2812106d3a5b7e11be71ba2534d25 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sat, 25 Sep 2021 21:24:59 +0000 Subject: [PATCH] Move TopicTree to the App --- src/App.h | 127 ++++++++++++++++++++++++++++--------- src/AsyncSocket.h | 2 + src/WebSocket.h | 28 +++----- src/WebSocketContext.h | 6 +- src/WebSocketContextData.h | 62 ++---------------- 5 files changed, 116 insertions(+), 109 deletions(-) diff --git a/src/App.h b/src/App.h index 37d6aa6..aa5fd29 100644 --- a/src/App.h +++ b/src/App.h @@ -18,6 +18,17 @@ #ifndef UWS_APP_H #define UWS_APP_H +#include + +namespace uWS { + /* Type queued up when publishing */ + struct TopicTreeMessage { + std::string message; + /*OpCode*/ int opCode; + bool compress; + }; +} + /* An app is a convenience wrapper of some of the most used fuctionalities and allows a * builder-pattern kind of init. Apps operate on the implicit thread local Loop */ @@ -53,10 +64,13 @@ struct TemplatedApp { private: /* The app always owns at least one http context, but creates websocket contexts on demand */ HttpContext *httpContext; - std::vector *> webSocketContexts; + /* WebSocketContexts are of differing type, but we as owners and creators must delete them correctly */ + std::vector> webSocketContextDeleters; public: + TopicTree *topicTree = nullptr; + /* Server name */ TemplatedApp &&addServerName(std::string hostname_pattern, SocketContextOptions options = {}) { @@ -100,27 +114,19 @@ public: * TopicTree of this app (technically there are many TopicTrees, however the concept is that one * 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()->topicTree.publish(nullptr, topic, {std::string(message), opCode, compress}); - } + topicTree->publish(nullptr, topic, {std::string(message), opCode, compress}); } /* Returns number of subscribers for this topic, or 0 for failure. * This function should probably be optimized a lot in future releases, * it could be O(1) with a hash map of fullnames and their counts. */ unsigned int numSubscribers(std::string_view topic) { - unsigned int subscribers = 0; - - for (auto *webSocketContext : webSocketContexts) { - auto *webSocketContextData = webSocketContext->getExt(); - - Topic *t = webSocketContextData->topicTree.lookupTopic(topic); - if (t) { - subscribers += t->size(); - } + Topic *t = topicTree->lookupTopic(topic); + if (t) { + return t->size(); } - return subscribers; + return 0; } ~TemplatedApp() { @@ -128,10 +134,21 @@ public: if (httpContext) { httpContext->free(); - for (auto *webSocketContext : webSocketContexts) { - webSocketContext->free(); + /* Free all our webSocketContexts in a type less way */ + for (auto &webSocketContextDeleter : webSocketContextDeleters) { + webSocketContextDeleter(); } } + + /* Delete TopicTree */ + if (topicTree) { + delete topicTree; + + /* And unregister loop callbacks */ + /* We must unregister any loop post handler here */ + Loop::get()->removePostHandler(topicTree); + Loop::get()->removePreHandler(topicTree); + } } /* Disallow copying, only move */ @@ -142,8 +159,12 @@ public: httpContext = other.httpContext; other.httpContext = nullptr; - /* Move webSocketContexts */ - webSocketContexts = std::move(other.webSocketContexts); + /* Move webSocketContextDeleters */ + webSocketContextDeleters = std::move(other.webSocketContextDeleters); + + /* Move TopicTree */ + other.topicTree = topicTree; + topicTree = nullptr; } TemplatedApp(SocketContextOptions options = {}) { @@ -200,21 +221,67 @@ public: std::cerr << "Warning: idleTimeout should be a multiple of 4!" << std::endl; } + /* If we don't have a TopicTree yet, create one now */ + if (!topicTree) { + + topicTree = new TopicTree([](Subscriber *s, TopicTreeMessage &message, auto flags) { + /* Subscriber's user is the socket */ + /* Unfortunately we need to cast is to PerSocketData = int + * since many different WebSocketContexts use the same + * TopicTree now */ + auto *ws = (WebSocket *) s->user; + + /* 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; + } + } + + /* If we ever overstep maxBackpresure, exit immediately */ + if (WebSocket::SendStatus::DROPPED == ws->send(message.message, (OpCode)message.opCode, message.compress)) { + + if (needsUncork) { + ((AsyncSocket *)ws)->uncork(); + } + /* Stop draining */ + return true; + } + + /* 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; + }); + + /* And hook it up with the loop */ + /* We empty for both pre and post just to make sure */ + Loop::get()->addPostHandler(topicTree, [topicTree = topicTree](Loop */*loop*/) { + /* Commit pub/sub batches every loop iteration */ + topicTree->drain(); + }); + + Loop::get()->addPreHandler(topicTree, [topicTree = topicTree](Loop */*loop*/) { + /* Commit pub/sub batches every loop iteration */ + topicTree->drain(); + }); + } + /* Every route has its own websocket context with its own behavior and user data type */ - auto *webSocketContext = WebSocketContext::create(Loop::get(), (us_socket_context_t *) httpContext); - - /* Add all other WebSocketContextData to this new WebSocketContextData */ - for (WebSocketContext *adjacentWebSocketContext : webSocketContexts) { - webSocketContext->getExt()->adjacentWebSocketContextDatas.push_back(adjacentWebSocketContext->getExt()); - } - - /* Add this WebSocketContextData to all other WebSocketContextData */ - for (WebSocketContext *adjacentWebSocketContext : webSocketContexts) { - adjacentWebSocketContext->getExt()->adjacentWebSocketContextDatas.push_back((WebSocketContextData *) webSocketContext->getExt()); - } + auto *webSocketContext = WebSocketContext::create(Loop::get(), (us_socket_context_t *) httpContext, topicTree); /* We need to clear this later on */ - webSocketContexts.push_back((WebSocketContext *) webSocketContext); + webSocketContextDeleters.push_back([webSocketContext]() { + webSocketContext->free(); + }); /* Quick fix to disable any compression if set */ #ifdef UWS_NO_ZLIB diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index f44509a..9cf7aa5 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -44,8 +44,10 @@ namespace uWS { template struct AsyncSocket { + /* This guy is promiscuous */ template friend struct HttpContext; template friend struct WebSocketContext; + template friend struct TemplatedApp; template friend struct WebSocketContextData; template friend struct TopicTree; diff --git a/src/WebSocket.h b/src/WebSocket.h index 5d4aac7..f5c10a3 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -93,7 +93,7 @@ public: WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData(); if (webSocketData->subscriber) { /* This will call back into us, send. */ - webSocketContextData->topicTree.drain(webSocketData->subscriber); + webSocketContextData->topicTree->drain(webSocketData->subscriber); } /* Transform the message to compressed domain if requested */ @@ -187,7 +187,7 @@ public: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.freeSubscriber(webSocketData->subscriber); + webSocketContextData->topicTree->freeSubscriber(webSocketData->subscriber); webSocketData->subscriber = nullptr; } @@ -215,12 +215,12 @@ 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 = webSocketContextData->topicTree.createSubscriber(); + webSocketData->subscriber = webSocketContextData->topicTree->createSubscriber(); webSocketData->subscriber->user = this; } /* Cannot return numSubscribers as this is only for this particular websocket context */ - webSocketContextData->topicTree.subscribe(webSocketData->subscriber, topic); + webSocketContextData->topicTree->subscribe(webSocketData->subscriber, topic); /* Subscribe always succeeds */ return true; @@ -235,11 +235,11 @@ public: WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); /* Cannot return numSubscribers as this is only for this particular websocket context */ - auto [ok, last] = webSocketContextData->topicTree.unsubscribe(webSocketData->subscriber, topic); + 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); + webSocketContextData->topicTree->freeSubscriber(webSocketData->subscriber); webSocketData->subscriber = nullptr; } @@ -257,7 +257,7 @@ public: return false; } - Topic *topicPtr = webSocketContextData->topicTree.lookupTopic(topic); + Topic *topicPtr = webSocketContextData->topicTree->lookupTopic(topic); if (!topicPtr) { return false; } @@ -277,14 +277,14 @@ public: WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); if (webSocketData->subscriber) { /* Lock this subscriber for unsubscription / subscription */ - webSocketContextData->topicTree.iteratingSubscriber = webSocketData->subscriber; + webSocketContextData->topicTree->iteratingSubscriber = webSocketData->subscriber; for (Topic *topicPtr : webSocketData->subscriber->topics) { cb({topicPtr->name.data(), topicPtr->name.length()}); } /* Unlock subscriber */ - webSocketContextData->topicTree.iteratingSubscriber = nullptr; + webSocketContextData->topicTree->iteratingSubscriber = nullptr; } } @@ -304,15 +304,7 @@ public: } /* Publish as sender, does not receive its own messages even if subscribed to relevant topics */ - 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->topicTree.publish(nullptr, topic, {std::string(message), opCode, compress}); - } - } + bool success = webSocketContextData->topicTree->publish(webSocketData->subscriber, topic, {std::string(message), opCode, compress}); return success; } diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index b55182e..6fd7ba2 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -249,7 +249,7 @@ private: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.freeSubscriber(webSocketData->subscriber); + webSocketContextData->topicTree->freeSubscriber(webSocketData->subscriber); webSocketData->subscriber = nullptr; } @@ -390,14 +390,14 @@ private: public: /* WebSocket contexts are always child contexts to a HTTP context so no SSL options are needed as they are inherited */ - static WebSocketContext *create(Loop */*loop*/, us_socket_context_t *parentSocketContext) { + static WebSocketContext *create(Loop */*loop*/, us_socket_context_t *parentSocketContext, TopicTree *topicTree) { WebSocketContext *webSocketContext = (WebSocketContext *) us_create_child_socket_context(SSL, parentSocketContext, sizeof(WebSocketContextData)); if (!webSocketContext) { return nullptr; } /* Init socket context data */ - new ((WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *)webSocketContext)) WebSocketContextData; + new ((WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *)webSocketContext)) WebSocketContextData(topicTree); return webSocketContext->init(); } }; diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 5585a1e..06138e7 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -40,16 +40,9 @@ struct WebSocketContextData { private: 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; + /* This one points to the App's shared topicTree */ + TopicTree *topicTree; /* The callbacks for this context */ MoveOnlyFunction *)> openHandler = nullptr; @@ -75,9 +68,6 @@ public: /* These are calculated on creation */ std::pair idleTimeoutComponents; - /* Each websocket context has a topic tree for pub/sub */ - TopicTree topicTree; - /* This is run once on start-up */ void calculateIdleTimeoutCompnents(unsigned short idleTimeout) { unsigned short margin = 4; @@ -92,55 +82,11 @@ public: } ~WebSocketContextData() { - /* We must unregister any loop post handler here */ - Loop::get()->removePostHandler(this); - Loop::get()->removePreHandler(this); + } - WebSocketContextData() : topicTree([](Subscriber *s, TopicTreeMessage &message, auto flags) { - /* Subscriber's user is the socket */ - auto *ws = (WebSocket *) s->user; + WebSocketContextData(TopicTree *topicTree) : topicTree(topicTree) { - /* 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; - } - } - - /* 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; - } - - /* 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*/) { - /* Commit pub/sub batches every loop iteration */ - topicTree.drain(); - }); - - Loop::get()->addPreHandler(this, [this](Loop */*loop*/) { - /* Commit pub/sub batches every loop iteration */ - topicTree.drain(); - }); } };