Topic splitting with SSE instructions

Wiebe Cazemier
1 parent c9f3af29
Showing 8 changed files with 91 additions and 9 deletions
CMakeLists.txt
FlashMQTests/FlashMQTests.pro
FlashMQTests/tst_maintests.cpp
mqttpacket.cpp
mqttpacket.h
session.cpp
threaddata.cpp
threaddata.h
@@ -7,7 +7,7 @@ add_definitions(-DOPENSSL_API_COMPAT=0x10100000L)
 set(CMAKE_CXX_STANDARD 11)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
-SET(CMAKE_CXX_FLAGS "-rdynamic")
+SET(CMAKE_CXX_FLAGS "-rdynamic -msse4.2")
 add_compile_options(-Wall)
@@ -79,3 +79,4 @@ HEADERS += \
 LIBS += -ldl -lssl -lcrypto
 QMAKE_LFLAGS += -rdynamic
+QMAKE_CXXFLAGS += -msse4.2
@@ -26,6 +26,7 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include "mainapp.h"
 #include "mainappthread.h"
 #include "twoclienttestcontext.h"
+#include "threaddata.h"
 // Dumb Qt version gives warnings when comparing uint with number literal.
 template <typename T1, typename T2>
@@ -75,6 +76,8 @@ private slots:
     void test_acl_patterns_username();
     void test_acl_patterns_clientid();
+    void test_sse_split();
+
 };
 MainTests::MainTests()
@@ -587,6 +590,31 @@ void MainTests::test_acl_patterns_clientid()
     QCOMPARE(aclTree.findPermission(splitToVector("d/clientid_one/f/A/B", '/'), AclGrant::Read, "foo", "clientid_one"), AuthResult::success);
 }
+void MainTests::test_sse_split()
+{
+    std::shared_ptr<SubscriptionStore> store(new SubscriptionStore);
+    std::shared_ptr<Settings> settings(new Settings);
+    ThreadData data(0, store, settings);
+
+    std::list<std::string> topics;
+    topics.push_back("one/two/threeabcasdfasdf/koe");
+    topics.push_back("/two/threeabcasdfasdf/koe"); // Test empty component.
+    topics.push_back("//two/threeabcasdfasdf/koe"); // Test two empty components.
+    topics.push_back("//1234567890abcde/bla/koe"); // Test two empty components, 15 char topic (one byte short of 16 alignment).
+    topics.push_back("//1234567890abcdef/bla/koe"); // Test two empty components, 16 char topic
+    topics.push_back("//1234567890abcdefg/bla/koe"); // Test two empty components, 17 char topic
+    topics.push_back("//1234567890abcdefg/1234567890abcdefg/koe"); // Test two empty components, two 17 char topics
+    topics.push_back("//1234567890abcdef/1234567890abcdefg/koe"); // Test two empty components, 16 and 17 char
+    topics.push_back("//1234567890abcdef/1234567890abcdefg/koe/");
+    topics.push_back("//1234567890abcdef/1234567890abcdefg/koe//");
+    topics.push_back("//1234567890abcdef/1234567890abcdef/");
+
+    for (const std::string &t : topics)
+    {
+        QCOMPARE(*data.splitTopic(t), splitToVector(t, '/'));
+    }
+}
+
 QTEST_GUILESS_MAIN(MainTests)
 #include "tst_maintests.moc"
@@ -102,7 +102,7 @@ MqttPacket::MqttPacket(const Publish &amp;publish) :
     }
     this->topic = publish.topic;
-    this->subtopics = splitToVector(publish.topic, '/');
+    this->subtopics = sender->getThreadData()->splitTopic(this->topic);
     packetType = PacketType::PUBLISH;
     this->qos = publish.qos;
@@ -439,7 +439,7 @@ void MqttPacket::handlePublish()
         throw ProtocolError("Duplicate flag is set for QoS 0 packet. This is illegal.");
     topic = std::string(readBytes(variable_header_length), variable_header_length);
-    subtopics = splitToVector(topic, '/');
+    subtopics = sender->getThreadData()->splitTopic(topic);
     if (!isValidUtf8(topic, true))
     {
@@ -464,7 +464,7 @@ void MqttPacket::handlePublish()
         sender->writeMqttPacket(response);
     }
-    if (sender->getThreadData()->authentication.aclCheck(sender->getClientId(), sender->getUsername(), topic, subtopics, AclAccess::write) == AuthResult::success)
+    if (sender->getThreadData()->authentication.aclCheck(sender->getClientId(), sender->getUsername(), topic, *subtopics, AclAccess::write) == AuthResult::success)
     {
         if (retain)
         {
@@ -479,7 +479,7 @@ void MqttPacket::handlePublish()
         bites[0] &= 0b11110110;
         // For the existing clients, we can just write the same packet back out, with our small alterations.
-        sender->getThreadData()->getSubscriptionStore()->queuePacketAtSubscribers(subtopics, *this);
+        sender->getThreadData()->getSubscriptionStore()->queuePacketAtSubscribers(*subtopics, *this);
     }
 }
@@ -569,7 +569,7 @@ const std::string &amp;MqttPacket::getTopic() const
     return this->topic;
 }
-const std::vector<std::string> &MqttPacket::getSubtopics() const
+const std::vector<std::string> *MqttPacket::getSubtopics() const
 {
     return this->subtopics;
 }
@@ -44,7 +44,7 @@ public:
 class MqttPacket
 {
     std::string topic;
-    std::vector<std::string> subtopics;
+    std::vector<std::string> *subtopics; // comes from local thread storage. See std::vector<std::string> *ThreadData::splitTopic(std::string &topic)
     std::vector<char> bites;
     size_t fixed_header_length = 0; // if 0, this packet does not contain the bytes of the fixed header.
     RemainingLength remainingLength;
@@ -94,7 +94,7 @@ public:
     const std::vector<char> &getBites() const { return bites; }
     char getQos() const { return qos; }
     const std::string &getTopic() const;
-    const std::vector<std::string> &getSubtopics() const;
+    const std::vector<std::string> *getSubtopics() const;
     std::shared_ptr<Client> getSender() const;
     void setSender(const std::shared_ptr<Client> &value);
     bool containsFixedHeader() const;
@@ -52,7 +52,7 @@ void Session::writePacket(const MqttPacket &amp;packet, char max_qos)
 {
     assert(max_qos <= 2);
-    if (thread->authentication.aclCheck(client_id, username, packet.getTopic(), packet.getSubtopics(), AclAccess::read) == AuthResult::success)
+    if (thread->authentication.aclCheck(client_id, username, packet.getTopic(), *packet.getSubtopics(), AclAccess::read) == AuthResult::success)
     {
         const char qos = std::min<char>(packet.getQos(), max_qos);
@@ -19,8 +19,12 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include <string>
 #include <sstream>
+#define TOPIC_MEMORY_LENGTH 65560
+
 ThreadData::ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore, std::shared_ptr<Settings> settings) :
     subscriptionStore(subscriptionStore),
+    subtopicParseMem(TOPIC_MEMORY_LENGTH),
+    topicCopy(TOPIC_MEMORY_LENGTH),
     settingsLocalCopy(*settings.get()),
     authentication(settingsLocalCopy),
     threadnr(threadnr)
@@ -155,6 +159,45 @@ void ThreadData::queuePasswdFileReload()
     wakeUpThread();
 }
+/**
+ * @brief ThreadData::splitTopic uses SSE4.2 to detect the '/' chars, 16 chars at a time, and returns a pointer to thread-local memory.
+ * @param topic string is altered: some extra space is reserved.
+ * @return Pointer to thread-owned vector of subtopics.
+ *
+ * Because it returns a pointer to the thread-local vector, only the current thread should touch it.
+ */
+std::vector<std::string> *ThreadData::splitTopic(const std::string &topic)
+{
+    subtopics.clear();
+
+    const int s = topic.size();
+    std::memcpy(topicCopy.data(), topic.c_str(), s+1);
+    std::memset(&topicCopy.data()[s], 0, 16);
+    int n = 0;
+    int carryi = 0;
+    while (n <= s)
+    {
+        const char *i = &topicCopy.data()[n];
+        __m128i loaded = _mm_loadu_si128((__m128i*)i);
+
+        int len_left = s - n;
+        int index = _mm_cmpestri(slashes, 1, loaded, len_left, 0);
+        std::memcpy(&subtopicParseMem[carryi], i, index);
+        carryi += std::min<int>(index, len_left);
+
+        n += index;
+
+        if (index < 16 || n >= s)
+        {
+            subtopics.emplace_back(subtopicParseMem.data(), carryi);
+            carryi = 0;
+            n++;
+        }
+    }
+
+    return &subtopics;
+}
+
 // TODO: profile how fast hash iteration is. Perhaps having a second list/vector is beneficial?
 void ThreadData::doKeepAliveCheck()
 {
@@ -18,6 +18,8 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #ifndef THREADDATA_H
 #define THREADDATA_H
+#include <immintrin.h>
+
 #include <thread>
 #include <sys/epoll.h>
@@ -47,6 +49,12 @@ class ThreadData
     std::shared_ptr<SubscriptionStore> subscriptionStore;
     Logger *logger;
+    // Topic parsing working memory
+    std::vector<std::string> subtopics;
+    std::vector<char> subtopicParseMem;
+    std::vector<char> topicCopy;
+    __m128i slashes = _mm_set1_epi8('/');
+
     void reload(std::shared_ptr<Settings> settings);
     void wakeUpThread();
     void doKeepAliveCheck();
@@ -82,6 +90,8 @@ public:
     void waitForQuit();
     void queuePasswdFileReload();
+    std::vector<std::string> *splitTopic(const std::string &topic);
+
 };
 #endif // THREADDATA_H