Publish $SYS topics in a thread, because authentication is thread local

It was a mistake that it ran inside the timer thread, which came to light by asserting the thread local authentication pointer (see a few commits ago).

Publish $SYS topics in a thread, because authentication is thread local
It was a mistake that it ran inside the timer thread, which came to light by asserting the thread local authentication pointer (see a few commits ago).
Wiebe Cazemier
1 parent c20ba7c5
Showing 4 changed files with 73 additions and 37 deletions
mainapp.cpp
mainapp.h
threaddata.cpp
threaddata.h
@@ -195,9 +195,8 @@ MainApp::MainApp(const std::string &amp;configFilePath) :
     auto fPasswordFileReload = std::bind(&MainApp::queuePasswordFileReloadAllThreads, this);
     timer.addCallback(fPasswordFileReload, 2000, "Password file reload.");
-    auto fPublishStats = std::bind(&MainApp::publishStatsOnDollarTopic, this);
+    auto fPublishStats = std::bind(&MainApp::queuePublishStatsOnDollarTopic, this);
     timer.addCallback(fPublishStats, 10000, "Publish stats on $SYS");
-    publishStatsOnDollarTopic();
     if (settings->authPluginTimerPeriod > 0)
     {
@@ -340,44 +339,20 @@ void MainApp::setFuzzFile(const std::string &amp;fuzzFilePath)
     this->fuzzFilePath = fuzzFilePath;
 }
-void MainApp::publishStatsOnDollarTopic()
+/**
+ * @brief MainApp::queuePublishStatsOnDollarTopic publishes the dollar topics, on a thread that has thread local authentication.
+ */
+void MainApp::queuePublishStatsOnDollarTopic()
 {
-    uint nrOfClients = 0;
-    uint64_t receivedMessageCountPerSecond = 0;
-    uint64_t receivedMessageCount = 0;
-    uint64_t sentMessageCountPerSecond = 0;
-    uint64_t sentMessageCount = 0;
+    std::lock_guard<std::mutex> locker(eventMutex);
-    for (std::shared_ptr<ThreadData> &thread : threads)
+    if (!threads.empty())
     {
-        nrOfClients += thread->getNrOfClients();
+        auto f = std::bind(&ThreadData::queuePublishStatsOnDollarTopic, threads.front().get(), threads);
+        taskQueue.push_front(f);
-        receivedMessageCountPerSecond += thread->getReceivedMessagePerSecond();
-        receivedMessageCount += thread->getReceivedMessageCount();
-
-        sentMessageCountPerSecond += thread->getSentMessagePerSecond();
-        sentMessageCount += thread->getSentMessageCount();
+        wakeUpThread();
     }
-
-    publishStat("$SYS/broker/clients/total", nrOfClients);
-
-    publishStat("$SYS/broker/load/messages/received/total", receivedMessageCount);
-    publishStat("$SYS/broker/load/messages/received/persecond", receivedMessageCountPerSecond);
-
-    publishStat("$SYS/broker/load/messages/sent/total", sentMessageCount);
-    publishStat("$SYS/broker/load/messages/sent/persecond", sentMessageCountPerSecond);
-
-    publishStat("$SYS/broker/retained messages/count", subscriptionStore->getRetainedMessageCount());
-}
-
-void MainApp::publishStat(const std::string &topic, uint64_t n)
-{
-    std::vector<std::string> subtopics;
-    splitTopic(topic, subtopics);
-    const std::string payload = std::to_string(n);
-    Publish p(topic, payload, 0);
-    subscriptionStore->queuePacketAtSubscribers(subtopics, p, true);
-    subscriptionStore->setRetainedMessage(topic, subtopics, payload, 0);
 }
 void MainApp::saveState()
@@ -604,6 +579,9 @@ void MainApp::start()
         threads.push_back(t);
     }
+    // Populate the $SYS topics, otherwise you have to wait until the timer expires.
+    threads.front()->queuePublishStatsOnDollarTopic(threads);
+
     uint next_thread_index = 0;
     struct epoll_event events[MAX_EVENTS];
@@ -82,8 +82,7 @@ class MainApp
     void queuePasswordFileReloadAllThreads();
     void queueAuthPluginPeriodicEventAllThreads();
     void setFuzzFile(const std::string &fuzzFilePath);
-    void publishStatsOnDollarTopic();
-    void publishStat(const std::string &topic, uint64_t n);
+    void queuePublishStatsOnDollarTopic();
     void saveState();
     MainApp(const std::string &configFilePath);
@@ -73,6 +73,62 @@ void ThreadData::quit()
     running = false;
 }
+/**
+ * @brief ThreadData::queuePublishStatsOnDollarTopic makes this thread publish the $SYS topics.
+ * @param threads
+ *
+ * We want to do that in a thread because all authentication state is thread local.
+ */
+void ThreadData::queuePublishStatsOnDollarTopic(std::vector<std::shared_ptr<ThreadData>> &threads)
+{
+    std::lock_guard<std::mutex> locker(taskQueueMutex);
+
+    auto f = std::bind(&ThreadData::publishStatsOnDollarTopic, this, threads);
+    taskQueue.push_front(f);
+
+    wakeUpThread();
+}
+
+void ThreadData::publishStatsOnDollarTopic(std::vector<std::shared_ptr<ThreadData>> &threads)
+{
+    uint nrOfClients = 0;
+    uint64_t receivedMessageCountPerSecond = 0;
+    uint64_t receivedMessageCount = 0;
+    uint64_t sentMessageCountPerSecond = 0;
+    uint64_t sentMessageCount = 0;
+
+    for (const std::shared_ptr<ThreadData> &thread : threads)
+    {
+        nrOfClients += thread->getNrOfClients();
+
+        receivedMessageCountPerSecond += thread->getReceivedMessagePerSecond();
+        receivedMessageCount += thread->getReceivedMessageCount();
+
+        sentMessageCountPerSecond += thread->getSentMessagePerSecond();
+        sentMessageCount += thread->getSentMessageCount();
+    }
+
+    publishStat("$SYS/broker/clients/total", nrOfClients);
+
+    publishStat("$SYS/broker/load/messages/received/total", receivedMessageCount);
+    publishStat("$SYS/broker/load/messages/received/persecond", receivedMessageCountPerSecond);
+
+    publishStat("$SYS/broker/load/messages/sent/total", sentMessageCount);
+    publishStat("$SYS/broker/load/messages/sent/persecond", sentMessageCountPerSecond);
+
+    publishStat("$SYS/broker/retained messages/count", subscriptionStore->getRetainedMessageCount());
+}
+
+void ThreadData::publishStat(const std::string &topic, uint64_t n)
+{
+    std::vector<std::string> subtopics;
+    splitTopic(topic, subtopics);
+    const std::string payload = std::to_string(n);
+    Publish p(topic, payload, 0);
+    subscriptionStore->queuePacketAtSubscribers(subtopics, p, true);
+    subscriptionStore->setRetainedMessage(topic, subtopics, payload, 0);
+}
+
 void ThreadData::giveClient(std::shared_ptr<Client> client)
 {
     clients_by_fd_mutex.lock();
@@ -61,6 +61,8 @@ class ThreadData
     void wakeUpThread();
     void doKeepAliveCheck();
     void quit();
+    void publishStatsOnDollarTopic(std::vector<std::shared_ptr<ThreadData>> &threads);
+    void publishStat(const std::string &topic, uint64_t n);
 public:
     Settings settingsLocalCopy; // Is updated on reload, within the thread loop.
@@ -91,6 +93,7 @@ public:
     void queueQuit();
     void waitForQuit();
     void queuePasswdFileReload();
+    void queuePublishStatsOnDollarTopic(std::vector<std::shared_ptr<ThreadData>> &threads);
     int getNrOfClients() const;