Send all non-delayed wills on exit

This includes some logic to have threads finish their work before quitting.

Send all non-delayed wills on exit
This includes some logic to have threads finish their work before quitting.
Wiebe Cazemier
1 parent 3e13b436
Showing 6 changed files with 58 additions and 0 deletions
client.cpp
client.h
mainapp.cpp
mainapp.h
threaddata.cpp
threaddata.h
@@ -394,6 +394,19 @@ uint32_t Client::getMaxIncomingPacketSize() const
     return this->maxIncomingPacketSize;
 }
+void Client::sendOrQueueWill()
+{
+    if (!this->threadData)
+        return;
+
+    if (!this->willPublish)
+        return;
+
+    std::shared_ptr<SubscriptionStore> &store = this->threadData->getSubscriptionStore();
+    store->queueWillMessage(willPublish, session);
+    this->willPublish.reset();
+}
+
 #ifndef NDEBUG
 /**
  * @brief IoWrapper::setFakeUpgraded().
@@ -149,6 +149,8 @@ public:
     uint32_t getMaxIncomingPacketSize() const;
+    void sendOrQueueWill();
+
 #ifndef NDEBUG
     void setFakeUpgraded();
 #endif
@@ -285,6 +285,14 @@ void MainApp::queueRemoveExpiredSessions()
     }
 }
+void MainApp::waitForAllThreadsQueuedWills()
+{
+    while(std::any_of(threads.begin(), threads.end(), [](std::shared_ptr<ThreadData> t){ return !t->allWilssSentForExit; }))
+    {
+        usleep(1000);
+    }
+}
+
 void MainApp::saveState()
 {
     std::lock_guard<std::mutex> lg(saveStateMutex);
@@ -607,6 +615,14 @@ void MainApp::start()
         }
     }
+    logger->logf(LOG_DEBUG, "Having all client in all threads send or queue their will.");
+    for(std::shared_ptr<ThreadData> &thread : threads)
+    {
+        thread->queueSendAllWills();
+    }
+
+    waitForAllThreadsQueuedWills();
+
     oneInstanceLock.unlock();
     logger->logf(LOG_DEBUG, "Signaling threads to finish.");
@@ -93,6 +93,7 @@ class MainApp
     void saveStateInThread();
     void queueSendQueuedWills();
     void queueRemoveExpiredSessions();
+    void waitForAllThreadsQueuedWills();
     MainApp(const std::string &configFilePath);
 public:
@@ -162,6 +162,18 @@ void ThreadData::removeExpiredSessions()
     subscriptionStore->removeExpiredSessionsClients();
 }
+void ThreadData::sendAllWils()
+{
+    std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
+
+    for(auto pairs : clients_by_fd)
+    {
+        pairs.second->sendOrQueueWill();
+    }
+
+    allWilssSentForExit = true;
+}
+
 void ThreadData::removeQueuedClients()
 {
     std::vector<int> fds;
@@ -389,6 +401,16 @@ void ThreadData::authPluginPeriodicEvent()
     authentication.periodicEvent();
 }
+void ThreadData::queueSendAllWills()
+{
+    std::lock_guard<std::mutex> locker(taskQueueMutex);
+
+    auto f = std::bind(&ThreadData::sendAllWils, this);
+    taskQueue.push_front(f);
+
+    wakeUpThread();
+}
+
 // TODO: profile how fast hash iteration is. Perhaps having a second list/vector is beneficial?
 void ThreadData::doKeepAliveCheck()
 {
@@ -66,6 +66,7 @@ class ThreadData
     void publishStat(const std::string &topic, uint64_t n);
     void sendQueuedWills();
     void removeExpiredSessions();
+    void sendAllWils();
     void removeQueuedClients();
@@ -74,6 +75,7 @@ public:
     Authentication authentication;
     bool running = true;
     bool finished = false;
+    bool allWilssSentForExit = false;
     std::thread thread;
     int threadnr = 0;
     int epollfd = 0;
@@ -117,6 +119,8 @@ public:
     void queueAuthPluginPeriodicEvent();
     void authPluginPeriodicEvent();
+
+    void queueSendAllWills();
 };
 #endif // THREADDATA_H