diff --git a/src/TopicTree.h b/src/TopicTree.h index bc6b920..334f4b2 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -1,5 +1,5 @@ /* - * Authored by Alex Hultman, 2018-2020. + * Authored by Alex Hultman, 2018-2021. * Intellectual property of third-party. * Licensed under the Apache License, Version 2.0 (the "License"); @@ -28,6 +28,7 @@ #include #include +/* We use std::function here, not fu2::unique_function */ #include namespace uWS { @@ -76,19 +77,15 @@ struct Intersection { std::vector holes; void forSubscriber(Subscriber *s, std::vector &senderForMessages, std::function)> cb) { - /* How far we already emitted of the two dataChannels */ + /* How far we already emitted of the two dataChannels */ std::pair emitted = {}; - //std::cout << "Subscriber: " << s << std::endl; - /* Holes are global to the entire topic tree, so we are not guaranteed to find * holes in this intersection - they are sorted, though */ int examinedHoles = 0; /* This is a slow path of sorts, most subscribers will be observers, not active senders */ for (unsigned int id : senderForMessages) { - //std::cout << "We are sender for id: " << id << std::endl; - std::pair toEmit = {}; std::pair toIgnore = {}; @@ -135,10 +132,6 @@ struct Intersection { }; struct TopicTree { - - /* Sender holes */ - std::map> senderHoles; - private: std::function cb; @@ -147,6 +140,9 @@ private: /* Global messageId for deduplication of overlapping topics and ordering between topics */ unsigned int messageId = 0; + /* Sender holes */ + std::map> senderHoles; + /* The triggered topics */ Topic *triggeredTopics[64]; int numTriggeredTopics = 0; @@ -269,6 +265,18 @@ public: delete root; } + /* This is part of the fast path, so should be optimal */ + std::vector &getSenderFor(Subscriber *s) { + static thread_local std::vector emptyVector; + + auto it = senderHoles.find(s); + if (it != senderHoles.end()) { + return it->second; + } + + return emptyVector; + } + void subscribe(std::string_view topic, Subscriber *subscriber) { /* Start iterating from the root */ Topic *iterator = root; @@ -326,13 +334,13 @@ public: } void publish(std::string_view topic, std::pair message, Subscriber *sender = nullptr) { - /* Add a hole for the sender if one */ if (sender) { senderHoles[sender].push_back(messageId); } publish(root, 0, 0, topic, message); + /* MessageIDs are reset on drain - this should be fine since messages itself are cleared on drain */ messageId++; } @@ -421,6 +429,7 @@ public: if (!numTriggeredTopics) { senderHoles.clear(); + messageId = 0; return; } @@ -436,7 +445,7 @@ public: if (min != (Subscriber *)UINTPTR_MAX) { /* Up to 64 triggered Topics per batch */ - std::map*/ Intersection> intersectionCache; + std::map intersectionCache; /* Loop over these here */ std::set::iterator it[64]; @@ -492,20 +501,14 @@ public: } /* Create the linear cache, {inflated, deflated} */ - /*std::pair*/ Intersection res; - //std::string messageIds; // sorterade id:n för meddelanden - - //std::vector< - - + Intersection res; for (auto &p : complete) { - //printf("messageId = %d\n", p.first); - - res.dataChannels.first.append(p.second.first); res.dataChannels.second.append(p.second.second); - // appenda {id, längd, längd} + /* 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(); @@ -513,24 +516,14 @@ public: res.holes.push_back(h); } - //can we know the messageId here and lookup if "min" is the sender? - cb(min, intersectionCache[intersection] = std::move(res)); } else { - - // vi kan göra en cache som håller inflated, deflated, messageIds - - // sen, för varje subscriber, kollar vi upp en vektor av messageIds - senderHoles - - // sen måste vi loopa över - cb(min, intersectionCache[intersection]); } min = nextMin; } - } /* Clear messages of triggered Topics */ @@ -540,6 +533,7 @@ public: } numTriggeredTopics = 0; senderHoles.clear(); + messageId = 0; } }; diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index b3b3894..1148a19 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -19,7 +19,7 @@ void testCorrectness() { topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { - intersection.forSubscriber(s, topicTree->senderHoles[s], [s, &actualResult](std::pair dataChannels) { + intersection.forSubscriber(s, topicTree->getSenderFor(s), [s, &actualResult](std::pair dataChannels) { actualResult[s].first += dataChannels.first; actualResult[s].second += dataChannels.second; });