From 8690dd31ca05492de4ce960ece6dae4179a51446 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 28 Jan 2021 00:28:10 +0100 Subject: [PATCH 1/9] Add working TopicTree unit test --- tests/Makefile | 6 ++-- tests/TopicTree.cpp | 74 ++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 69 insertions(+), 11 deletions(-) diff --git a/tests/Makefile b/tests/Makefile index c6ff50c..c846b03 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -1,9 +1,9 @@ default: - #$(CXX) -std=c++17 -fsanitize=address TopicTree.cpp -o TopicTree - #./TopicTree + $(CXX) -std=c++17 -fsanitize=address TopicTree.cpp -o TopicTree + ./TopicTree $(CXX) -std=c++17 -fsanitize=address HttpRouter.cpp -o HttpRouter ./HttpRouter $(CXX) -std=c++17 -fsanitize=address BloomFilter.cpp -o BloomFilter ./BloomFilter $(CXX) -std=c++17 -fsanitize=address ExtensionsNegotiator.cpp -o ExtensionsNegotiator - ./ExtensionsNegotiator \ No newline at end of file + ./ExtensionsNegotiator diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 1feb436..db974d4 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -7,14 +7,16 @@ void testUnsubscribeInside() { std::cout << "TestUnsubscribeInside" << std::endl; uWS::TopicTree *topicTree; - std::map expectedResult; + std::map> expectedResult; + + topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair dataChannels) { + std::string_view data = dataChannels.second; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::string_view data) { /* Check for unexpected subscribers */ assert(expectedResult.find(s) != expectedResult.end()); /* Check for unexpected data */ - assert(expectedResult[s] == data); + assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second); /* This one causes mess-up */ topicTree->unsubscribeAll(s); @@ -28,8 +30,8 @@ void testUnsubscribeInside() { /* Fill out expectedResult */ expectedResult = { - {s1, "Ett!"}, - {s2, "Två!"} + {s1, {"Ett!", "Ett!"}}, + {s2, {"Två!", "Två!"}} }; /* Make sure s1 < s2 */ @@ -44,8 +46,62 @@ void testUnsubscribeInside() { topicTree->subscribe("1", s1); /* This order matters, as it fills triggeredTopics array in order */ - topicTree->publish("1", "Ett!"); - topicTree->publish("2", "Två!"); + topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")}); + topicTree->publish("2", {std::string_view("Två!"), std::string_view("Ett!")}); + + topicTree->drain(); + + /* Release resources */ + topicTree->unsubscribeAll(s1); + topicTree->unsubscribeAll(s2); + + delete s1; + delete s2; + + delete topicTree; +} + +void testPublisherHoles() { + std::cout << "TestPublisherHoles" << std::endl; + + uWS::TopicTree *topicTree; + std::map> expectedResult; + + topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair dataChannels) { + + /* Check for unexpected subscribers */ + assert(expectedResult.find(s) != expectedResult.end()); + + /* Check for unexpected data */ + assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second); + + /* We actually don't use this one */ + return 0; + }); + + uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); + uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); + + /* Fill out expectedResult */ + expectedResult = { + {s1, {"Två!", "Två!"}}, // todo: this one should not receive what he sent himself + {s2, {"Två!", "Två!"}} + }; + + /* Make sure s1 < s2 */ + if (s2 < s1) { + uWS::Subscriber *tmp = s1; + s1 = s2; + s2 = tmp; + } + + /* This order does not matter as it fills a tree */ + topicTree->subscribe("1", s2); + topicTree->subscribe("1", s1); + + /* This order matters, as it fills triggeredTopics array in order */ + //topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")}); + topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}); topicTree->drain(); @@ -60,5 +116,7 @@ void testUnsubscribeInside() { } int main() { - testUnsubscribeInside(); + //testUnsubscribeInside(); + + testPublisherHoles(); } \ No newline at end of file From 0d74439933a55f1ef69c3d3b4a9c2fda690a36d5 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 28 Jan 2021 01:52:22 +0100 Subject: [PATCH 2/9] Initial sender hole experiment --- src/TopicTree.h | 62 +++++++++++++++++++++++++++++++++++++++------ tests/TopicTree.cpp | 44 ++++++++++++++++++++++++++------ 2 files changed, 90 insertions(+), 16 deletions(-) diff --git a/src/TopicTree.h b/src/TopicTree.h index 43f3db5..20250e8 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -28,6 +28,16 @@ #include #include +struct Hole { + std::pair lengths; + unsigned int messageId; +}; + +struct Intersection { + std::pair dataChannels; + std::vector holes; +}; + namespace uWS { /* A Subscriber is an extension of a socket */ @@ -65,8 +75,12 @@ struct Topic { }; struct TopicTree { + + /* Sender holes */ + std::map> senderHoles; + private: - std::function)> cb; + std::function cb; Topic *root = new Topic; @@ -187,7 +201,7 @@ private: public: - TopicTree(std::function)> cb) { + TopicTree(std::function cb) { this->cb = cb; } @@ -251,7 +265,13 @@ public: } } - void publish(std::string_view topic, std::pair message) { + 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); messageId++; } @@ -340,6 +360,7 @@ public: numTriggeredTopics = numFilteredTriggeredTopics; if (!numTriggeredTopics) { + senderHoles.clear(); return; } @@ -355,7 +376,7 @@ public: if (min != (Subscriber *)UINTPTR_MAX) { /* Up to 64 triggered Topics per batch */ - std::map> intersectionCache; + std::map*/ Intersection> intersectionCache; /* Loop over these here */ std::set::iterator it[64]; @@ -402,7 +423,7 @@ public: } /* Generate cache for intersection */ - if (intersectionCache[intersection].first.length() == 0) { + if (intersectionCache[intersection].dataChannels.first.length() == 0) { /* Build the union in order without duplicates */ std::map> complete; @@ -411,15 +432,39 @@ public: } /* Create the linear cache, {inflated, deflated} */ - std::pair res; + /*std::pair*/ Intersection res; + //std::string messageIds; // sorterade id:n för meddelanden + + //std::vector< + + for (auto &p : complete) { - res.first.append(p.second.first); - res.second.append(p.second.second); + 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} + 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); } + //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]); } @@ -434,6 +479,7 @@ public: triggeredTopics[i]->triggered = false; } numTriggeredTopics = 0; + senderHoles.clear(); } }; diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index db974d4..61c3e69 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -9,14 +9,14 @@ void testUnsubscribeInside() { uWS::TopicTree *topicTree; std::map> expectedResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair dataChannels) { - std::string_view data = dataChannels.second; + topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, Intersection &intersection) { + std::string_view data = intersection.dataChannels.second; /* Check for unexpected subscribers */ assert(expectedResult.find(s) != expectedResult.end()); /* Check for unexpected data */ - assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second); + assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second); /* This one causes mess-up */ topicTree->unsubscribeAll(s); @@ -67,13 +67,40 @@ void testPublisherHoles() { uWS::TopicTree *topicTree; std::map> expectedResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair dataChannels) { + topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, /*std::pair &dataChannels*/ Intersection &intersection) { + + + std::cout << "Subscriber: " << s << std::endl; + + std::vector &senderForMessages = topicTree->senderHoles[s]; + + for (unsigned int id : senderForMessages) { + std::cout << "We are sender for id: " << id << std::endl; + } + + // iterate holes + for (Hole h : intersection.holes) { + std::cout << h.messageId << std::endl; + + + + // todo: this linear search is not needed as ids are ordered! + if (std::find(senderForMessages.begin(), senderForMessages.end(), h.messageId) != senderForMessages.end()) { + std::cout << "WE ARE SENDER FOR THIS MESSAGE!" << std::endl; + } + + + + } + + + /* Check for unexpected subscribers */ assert(expectedResult.find(s) != expectedResult.end()); /* Check for unexpected data */ - assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second); + assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second); /* We actually don't use this one */ return 0; @@ -84,8 +111,8 @@ void testPublisherHoles() { /* Fill out expectedResult */ expectedResult = { - {s1, {"Två!", "Två!"}}, // todo: this one should not receive what he sent himself - {s2, {"Två!", "Två!"}} + {s1, {"Två!Två!", "Två!Två!"}}, // todo: this one should not receive what he sent himself + {s2, {"Två!Två!", "Två!Två!"}} }; /* Make sure s1 < s2 */ @@ -101,7 +128,8 @@ void testPublisherHoles() { /* This order matters, as it fills triggeredTopics array in order */ //topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")}); - topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}); + topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1); + topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1); topicTree->drain(); From 596d7fdfc085f42111963687de5bb45005d5e0ea Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 28 Jan 2021 16:49:09 +0100 Subject: [PATCH 3/9] Pass basic topicTree testcase --- src/TopicTree.h | 80 +++++++++++++++++++++--- tests/TopicTree.cpp | 149 +++++++++++++++----------------------------- 2 files changed, 120 insertions(+), 109 deletions(-) diff --git a/src/TopicTree.h b/src/TopicTree.h index 20250e8..bc6b920 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -28,15 +28,7 @@ #include #include -struct Hole { - std::pair lengths; - unsigned int messageId; -}; - -struct Intersection { - std::pair dataChannels; - std::vector holes; -}; +#include namespace uWS { @@ -74,6 +66,74 @@ struct Topic { std::set subs; }; +struct Hole { + std::pair lengths; + unsigned int messageId; +}; + +struct Intersection { + std::pair dataChannels; + std::vector holes; + + void forSubscriber(Subscriber *s, std::vector &senderForMessages, std::function)> cb) { + /* 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 = {}; + + /* This linear search is most probably very small - it could be made log2 if every hole + * knows about its previous accumulated length, which is easy to set up. However this + * log2 search will most likely never be a warranted perf. gain */ + for (; examinedHoles < holes.size(); examinedHoles++) { + if (holes[examinedHoles].messageId == id) { + toIgnore.first += holes[examinedHoles].lengths.first; + toIgnore.second += holes[examinedHoles].lengths.second; + examinedHoles++; + break; + } + /* We are not the sender of this message so we should emit it in this segment */ + toEmit.first += holes[examinedHoles].lengths.first; + toEmit.second += holes[examinedHoles].lengths.second; + } + + /* Emit this segment */ + 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), + }; + + cb(cutDataChannels); + } + + emitted.first += toEmit.first + toIgnore.first; + emitted.second += toEmit.second + toIgnore.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); + } +}; + struct TopicTree { /* Sender holes */ @@ -439,7 +499,7 @@ public: for (auto &p : complete) { - printf("messageId = %d\n", p.first); + //printf("messageId = %d\n", p.first); res.dataChannels.first.append(p.second.first); diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 61c3e69..b3b3894 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -3,23 +3,26 @@ #include #include -void testUnsubscribeInside() { - std::cout << "TestUnsubscribeInside" << std::endl; +/* Modifying the topicTree inside callback is not allowed, we had + * tests for this before but we never need this to work anyways. + * Closing a socket when reaching too much backpressure is done + * deferred to next event loop iteration so we never need to modify + * topicTree inside callback - removed this test */ + +/* This tests pretty much all features for obvious incorrectness */ +void testCorrectness() { + std::cout << "TestCorrectness" << std::endl; uWS::TopicTree *topicTree; std::map> expectedResult; + std::map> actualResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, Intersection &intersection) { - std::string_view data = intersection.dataChannels.second; + topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { - /* Check for unexpected subscribers */ - assert(expectedResult.find(s) != expectedResult.end()); - - /* Check for unexpected data */ - assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second); - - /* This one causes mess-up */ - topicTree->unsubscribeAll(s); + intersection.forSubscriber(s, topicTree->senderHoles[s], [s, &actualResult](std::pair dataChannels) { + actualResult[s].first += dataChannels.first; + actualResult[s].second += dataChannels.second; + }); /* We actually don't use this one */ return 0; @@ -28,110 +31,60 @@ void testUnsubscribeInside() { uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); - /* Fill out expectedResult */ - expectedResult = { - {s1, {"Ett!", "Ett!"}}, - {s2, {"Två!", "Två!"}} - }; - - /* Make sure s1 < s2 */ + /* Make sure s1 < s2 (for debugging) */ if (s2 < s1) { uWS::Subscriber *tmp = s1; s1 = s2; s2 = tmp; } - /* This order does not matter as it fills a tree */ - topicTree->subscribe("2", s2); - topicTree->subscribe("1", s1); + /* Publish to topic3 - nobody should see this */ + topicTree->publish("topic3", {std::string_view("Nobody"), std::string_view("should see")}, nullptr); - /* This order matters, as it fills triggeredTopics array in order */ - topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")}); - topicTree->publish("2", {std::string_view("Två!"), std::string_view("Ett!")}); + /* Subscribe s1 to topic3 - s1 should not see above message */ + topicTree->subscribe("topic3", s1); - topicTree->drain(); + /* 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); - /* Release resources */ - topicTree->unsubscribeAll(s1); - topicTree->unsubscribeAll(s2); + /* Subscribe s2 to topic3 - should not get any message */ + topicTree->subscribe("topic3", s2); - delete s1; - delete s2; + /* Publish to topic3 without sender - both should see */ + topicTree->publish("topic3", {std::string_view("Both"), std::string_view("should see")}, nullptr); - delete topicTree; -} + /* 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); -void testPublisherHoles() { - std::cout << "TestPublisherHoles" << std::endl; + /* 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); - uWS::TopicTree *topicTree; - std::map> expectedResult; + /* 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 = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, /*std::pair &dataChannels*/ Intersection &intersection) { - - - std::cout << "Subscriber: " << s << std::endl; - - std::vector &senderForMessages = topicTree->senderHoles[s]; - - for (unsigned int id : senderForMessages) { - std::cout << "We are sender for id: " << id << std::endl; - } - - // iterate holes - for (Hole h : intersection.holes) { - std::cout << h.messageId << std::endl; - - - - // todo: this linear search is not needed as ids are ordered! - if (std::find(senderForMessages.begin(), senderForMessages.end(), h.messageId) != senderForMessages.end()) { - std::cout << "WE ARE SENDER FOR THIS MESSAGE!" << std::endl; - } - - - - } - - - - - /* Check for unexpected subscribers */ - assert(expectedResult.find(s) != expectedResult.end()); - - /* Check for unexpected data */ - assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second); - - /* We actually don't use this one */ - return 0; - }); - - uWS::Subscriber *s1 = new uWS::Subscriber(nullptr); - uWS::Subscriber *s2 = new uWS::Subscriber(nullptr); + // todo: add more cases involving more topics and duplicates, etc /* Fill out expectedResult */ expectedResult = { - {s1, {"Två!Två!", "Två!Två!"}}, // todo: this one should not receive what he sent himself - {s2, {"Två!Två!", "Två!Två!"}} + {s1, {"Boths1Again, both", "should seeshould see, not s2should see this as well"}}, + {s2, {"Boths2Again, both", "should seeshould see, not s1should see this as well"}} }; - /* Make sure s1 < s2 */ - if (s2 < s1) { - uWS::Subscriber *tmp = s1; - s1 = s2; - s2 = tmp; - } - - /* This order does not matter as it fills a tree */ - topicTree->subscribe("1", s2); - topicTree->subscribe("1", s1); - - /* This order matters, as it fills triggeredTopics array in order */ - //topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")}); - topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1); - topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1); - + /* Compare result with expected result for every subscriber */ topicTree->drain(); + 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; + exit(1); + } + } /* Release resources */ topicTree->unsubscribeAll(s1); @@ -144,7 +97,5 @@ void testPublisherHoles() { } int main() { - //testUnsubscribeInside(); - - testPublisherHoles(); + testCorrectness(); } \ No newline at end of file From af8094268cd591960d3a7d28151b070f38aa3a7a Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 28 Jan 2021 17:15:05 +0100 Subject: [PATCH 4/9] Cleanups --- src/TopicTree.h | 58 ++++++++++++++++++++------------------------- tests/TopicTree.cpp | 2 +- 2 files changed, 27 insertions(+), 33 deletions(-) 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; }); From ac7c72698dec81847f105b32767c8565622f5c69 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 28 Jan 2021 17:26:02 +0100 Subject: [PATCH 5/9] Initial integration of new topicTree --- src/TopicTree.h | 4 ++-- src/WebSocketContextData.h | 8 +++++++- tests/TopicTree.cpp | 2 +- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/src/TopicTree.h b/src/TopicTree.h index 334f4b2..6d11c4d 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -76,13 +76,13 @@ struct Intersection { std::pair dataChannels; std::vector holes; - void forSubscriber(Subscriber *s, std::vector &senderForMessages, std::function)> cb) { + void forSubscriber(std::vector &senderForMessages, std::function)> cb) { /* How far we already emitted of the two dataChannels */ std::pair emitted = {}; /* 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; + unsigned int examinedHoles = 0; /* This is a slow path of sorts, most subscribers will be observers, not active senders */ for (unsigned int id : senderForMessages) { diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index db2a04f..e8d7dba 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -76,7 +76,13 @@ public: Loop::get()->removePreHandler(this); } - WebSocketContextData() : topicTree([this](Subscriber *s, std::pair data) -> int { + WebSocketContextData() : topicTree([this](Subscriber *s, Intersection &intersection) -> int { + + + std::pair data = intersection.dataChannels; + + + /* We rely on writing to regular asyncSockets */ auto *asyncSocket = (AsyncSocket *) s->user; diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 1148a19..8d1ef9f 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->getSenderFor(s), [s, &actualResult](std::pair dataChannels) { + intersection.forSubscriber(topicTree->getSenderFor(s), [s, &actualResult](std::pair dataChannels) { actualResult[s].first += dataChannels.first; actualResult[s].second += dataChannels.second; }); From dc423cfda2d3f4d28d29954eb719048c59303078 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 29 Jan 2021 12:21:46 +0100 Subject: [PATCH 6/9] Add "fin" to intersection iterator --- src/TopicTree.h | 7 ++++--- tests/TopicTree.cpp | 16 +++++++++++++++- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/src/TopicTree.h b/src/TopicTree.h index 6d11c4d..44cb4fb 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -76,7 +76,7 @@ struct Intersection { std::pair dataChannels; std::vector holes; - void forSubscriber(std::vector &senderForMessages, std::function)> cb) { + void forSubscriber(std::vector &senderForMessages, std::function, bool)> cb) { /* How far we already emitted of the two dataChannels */ std::pair emitted = {}; @@ -111,7 +111,8 @@ struct Intersection { std::string_view(dataChannels.second.data() + emitted.second, toEmit.second), }; - cb(cutDataChannels); + /* We only need to test the first data channel for "FIN" */ + cb(cutDataChannels, emitted.first + toEmit.first + toIgnore.first == dataChannels.first.length()); } emitted.first += toEmit.first + toIgnore.first; @@ -127,7 +128,7 @@ struct Intersection { std::string_view(dataChannels.second.data() + emitted.second, dataChannels.second.length() - emitted.second), }; - cb(cutDataChannels); + cb(cutDataChannels, true); } }; diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index 8d1ef9f..389d1aa 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -19,11 +19,25 @@ void testCorrectness() { topicTree = new uWS::TopicTree([&topicTree, &actualResult](uWS::Subscriber *s, uWS::Intersection &intersection) { - intersection.forSubscriber(topicTree->getSenderFor(s), [s, &actualResult](std::pair dataChannels) { + /* How many bytes we have in first data channel at time we get fin = true */ + unsigned int finAt = 0; + + 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; }); From f8800d3e091dda095e645b76a3995556472f90d2 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 29 Jan 2021 13:10:49 +0100 Subject: [PATCH 7/9] Proper topicTree integration --- src/WebSocketContextData.h | 130 ++++++++++++++++++++++--------------- 1 file changed, 78 insertions(+), 52 deletions(-) diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index e8d7dba..1488e5e 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -78,83 +78,109 @@ public: WebSocketContextData() : topicTree([this](Subscriber *s, Intersection &intersection) -> int { - - std::pair data = intersection.dataChannels; - - + /* 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; - /* Check if we now have too much backpressure (todo: don't buffer up before check) */ - if (!maxBackpressure || (unsigned int) asyncSocket->getBufferedAmount() < maxBackpressure) { - /* Pick uncompressed data track */ - std::string_view selectedData = data.first; + /* If we are corked, do not uncork - otherwise if we cork in here, uncork before leaving */ + bool wasCorked = asyncSocket->isCorked(); - /* Are we using compression? Fine, pick the compressed data track */ - WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); - if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { + /* Do we even have room for potential data? */ + if (!maxBackpressure || asyncSocket->getBufferedAmount() < maxBackpressure) { - /* This is used for both shared and dedicated paths */ - selectedData = data.second; + /* Roll over all our segments */ + intersection.forSubscriber(topicTree.getSenderFor(s), [asyncSocket, this](std::pair data, bool fin) { - /* However, dedicated compression has its own path */ - if (compression != SHARED_COMPRESSOR) { + /* 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(); + } - WebSocket *ws = (WebSocket *) asyncSocket; + /* Pick uncompressed data track */ + std::string_view selectedData = data.first; - /* We need to handle being corked, and corking here */ - bool needsUncorking = false; - if (!ws->isCorked() && ws->canCork()) { - asyncSocket->cork(); - needsUncorking = true; - } + /* Are we using compression? Fine, pick the compressed data track */ + WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); + if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { - 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); + /* This is used for both shared and dedicated paths */ + selectedData = data.second; - /* Skip this message if our backpressure is too high */ - if (maxBackpressure && ws->getBufferedAmount() > maxBackpressure) { - break; + /* 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(); } - /* Here we perform the actual compression and framing */ - ws->send(unframedMessage, mm.opCode, mm.compress); + 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); - /* Advance until empty */ - selectedData.remove_prefix(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; } - - /* Here we need to uncork or keep it as was */ - if (needsUncorking) { - asyncSocket->uncork(); - } - - /* See below */ - return 0; } - } - /* 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()); + /* 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) { + asyncSocket->timeout(this->idleTimeout); + } + } + }); + } + + /* 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) { asyncSocket->timeout(this->idleTimeout); } } - - /* Failing here 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 */ } - /* If we have too much backpressure, simply skip sending from here */ - - /* Also (defer) a close if we have too much backpressure if that is what we want */ + /* 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); } From af631e453428f6c79522e97a7957df114517d9b8 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 29 Jan 2021 13:17:56 +0100 Subject: [PATCH 8/9] Update topicTree fuzzing --- fuzzing/TopicTree.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/fuzzing/TopicTree.cpp b/fuzzing/TopicTree.cpp index 8bc1897..24a91d1 100644 --- a/fuzzing/TopicTree.cpp +++ b/fuzzing/TopicTree.cpp @@ -9,7 +9,10 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { /* Create topic tree */ - uWS::TopicTree topicTree([](uWS::Subscriber *s, std::pair message) { + 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. */ From 76487df709d7471f04431b7100bde0a008b26a96 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 29 Jan 2021 13:27:04 +0100 Subject: [PATCH 9/9] Hook up WebSocket::publish as sender --- src/HttpResponse.h | 2 +- src/WebSocket.h | 11 +++++++++-- src/WebSocketContextData.h | 10 +++++----- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/src/HttpResponse.h b/src/HttpResponse.h index d91fe64..dc31f4f 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -86,7 +86,7 @@ private: #ifndef UWS_HTTPRESPONSE_NO_WRITEMARK if (!Super::getLoopData()->noMark) { /* We only expose major version */ - writeHeader("uWebSockets", "18"); + writeHeader("uWebSockets", "19"); } #endif } diff --git a/src/WebSocket.h b/src/WebSocket.h index fcf40d1..6a69ea5 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -230,8 +230,15 @@ public: WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); - /* Is the same as publishing per websocket context */ - webSocketContextData->publish(topic, message, opCode, compress); + + /* Make us a subscriber if we aren't yet (important for allocating a sender address) */ + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); + if (!webSocketData->subscriber) { + webSocketData->subscriber = new Subscriber(this); + } + + /* Publish as sender, does not receive its own messages even if subscribed to relevant topics */ + webSocketContextData->publish(topic, message, opCode, compress, webSocketData->subscriber); } }; diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 1488e5e..99c2f07 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -200,7 +200,7 @@ public: } /* Helper for topictree publish, common path from app and ws */ - void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress) { + void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress, Subscriber *sender = nullptr) { /* 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); @@ -208,7 +208,7 @@ public: /* If compression is disabled */ if (compression == DISABLED) { /* Leave second field empty as nobody will ever read it */ - topicTree.publish(topic, {std::string_view(dst, dst_length), {}}); + 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) { @@ -225,7 +225,7 @@ public: 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 */ - 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)}); + 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); @@ -246,14 +246,14 @@ public: 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. */ - topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}); + topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}, sender); } }