Continued with my tree-based subscriptions

Wiebe Cazemier
1 parent 4881eaf9
Showing 5 changed files with 69 additions and 50 deletions
CMakeLists.txt
rwlockguard.cpp
rwlockguard.h
subscriptionstore.cpp
subscriptionstore.h
@@ -17,6 +17,7 @@ add_executable(FlashMQ
     exceptions.cpp
     types.cpp
     subscriptionstore.cpp
+    rwlockguard.cpp
     )
 target_link_libraries(FlashMQ pthread)
+#include "rwlockguard.h"
+
+RWLockGuard::RWLockGuard(pthread_rwlock_t *rwlock) :
+    rwlock(rwlock)
+{
+
+}
+
+RWLockGuard::~RWLockGuard()
+{
+    pthread_rwlock_unlock(rwlock);
+}
+
+void RWLockGuard::wrlock()
+{
+    pthread_rwlock_wrlock(rwlock);
+}
+
+void RWLockGuard::rdlock()
+{
+    pthread_rwlock_wrlock(rwlock);
+}
+#ifndef RWLOCKGUARD_H
+#define RWLOCKGUARD_H
+
+
+#include <pthread.h>
+
+class RWLockGuard
+{
+    pthread_rwlock_t *rwlock = NULL;
+public:
+    RWLockGuard(pthread_rwlock_t *rwlock);
+    ~RWLockGuard();
+    void wrlock();
+    void rdlock();
+};
+
+#endif // RWLOCKGUARD_H
@@ -2,6 +2,8 @@
 #include "cassert"
+#include "rwlockguard.h"
+
 SubscriptionNode::SubscriptionNode(const std::string &subtopic) :
     subtopic(subtopic)
@@ -18,7 +20,9 @@ SubscriptionStore::SubscriptionStore() :
 void SubscriptionStore::addSubscription(Client_p &client, std::string &topic)
 {
     const std::list<std::string> subtopics = split(topic, '/');
-    std::lock_guard<std::mutex> lock(subscriptionsMutex);
+
+    RWLockGuard lock_guard(&subscriptionsRwlock);
+    lock_guard.wrlock();
     SubscriptionNode *deepestNode = subscriptions2.get();
     for(const std::string &subtopic : subtopics)
@@ -37,71 +41,44 @@ void SubscriptionStore::addSubscription(Client_p &amp;client, std::string &amp;topic)
     {
         deepestNode->subscribers.insert(client->getClientId());
     }
+
+    clients_by_id[client->getClientId()] = client;
 }
 void SubscriptionStore::removeClient(const Client_p &client)
 {
-
+    RWLockGuard lock_guard(&subscriptionsRwlock);
+    lock_guard.wrlock();
+    clients_by_id.erase(client->getClientId());
 }
 void SubscriptionStore::queueAtClientsTemp(std::string &topic, const MqttPacket &packet, const Client_p &sender)
 {
     const std::list<std::string> subtopics = split(topic, '/');
+    const auto &clients = clients_by_id;
-    // TODO: temp. I want to work with read copies of the subscription store, to avoid frequent lock contention.
-    std::lock_guard<std::mutex> lock(subscriptionsMutex);
+    RWLockGuard lock_guard(&subscriptionsRwlock);
+    lock_guard.rdlock();
-    SubscriptionNode *deepestNode = subscriptions2.get();
+    const SubscriptionNode *deepestNode = subscriptions2.get();
     for(const std::string &subtopic : subtopics)
     {
-        SubscriptionNode &nodeRef = *deepestNode;
-
-        if (nodeRef.children.count(subtopic) == 0)
+        auto sub_iter = deepestNode->children.find(subtopic);
+        if (sub_iter == deepestNode->children.end())
             return;
-        std::unique_ptr<SubscriptionNode> &node = nodeRef.children[subtopic];
-
-        assert(node); // because any empty unique_ptr's is a bug
-
-        for (const std::string &client_id : node->subscribers)
-        {
-
-        }
-
-        deepestNode = node.get();
-
+        const std::unique_ptr<SubscriptionNode> &sub_node = sub_iter->second;
+        assert(sub_node); // because any empty unique_ptr's is a bug
+        deepestNode = sub_node.get();
     }
-    /*
-    for(const std::string &subtopic : subtopics)
-    {
-        std::unique_ptr<SubscriptionNode> &node = subscriptions2[subtopic];
-
-        if (!node)
-        {
-            subscriptions2
-        }
-    }*/
-
-    /*
-    for(const Client_p &client : subscriptions2[topic])
+    for (const std::string &client_id : deepestNode->subscribers)
     {
-        client->writeMqttPacket(packet);
-
-
-        if (client->getThreadData()->threadnr == sender->getThreadData()->threadnr)
-        {
-            client->writeMqttPacket(packet); // TODO: with my current hack way, this is wrong. Not using a lock only works with my previous idea of queueing.
-        }
-        else
-        {
-            // Or keep a list of queued messages in the store, per client?
-
-            //client->writeMqttPacketLocked(packet);
-            //client->getThreadData()->addToReadyForDequeuing(client);
-            //client->getThreadData()->wakeUpThread();
-        }
-    }*/
+        std::cout << "Publishing to " << client_id << std::endl;
+        auto client_it = clients.find(client_id);
+        if (client_it != clients.end())
+            client_it->second->writeMqttPacket(packet);
+    }
 }
@@ -5,6 +5,7 @@
 #include <unordered_set>
 #include <list>
 #include <mutex>
+#include <pthread.h>
 #include "forward_declarations.h"
@@ -20,7 +21,7 @@ public:
     SubscriptionNode(const SubscriptionNode &node) = delete;
     SubscriptionNode(SubscriptionNode &&node) = delete;
-    std::unordered_set<std::string> subscribers;
+    std::unordered_set<std::string> subscribers; // The idea is to store subscriptions by client id, to support persistent sessions.
     std::unordered_map<std::string, std::unique_ptr<SubscriptionNode>> children;
 };
@@ -29,7 +30,8 @@ public:
 class SubscriptionStore
 {
     std::unique_ptr<SubscriptionNode> subscriptions2;
-    std::mutex subscriptionsMutex;
+    pthread_rwlock_t subscriptionsRwlock = PTHREAD_RWLOCK_INITIALIZER;
+    std::unordered_map<std::string, Client_p> clients_by_id;
 public:
     SubscriptionStore();