Fix several deadlocks

One fix is client destruction happening on the correct thread (when kicking another one off with existing client ID). This caused deadlocks on the subscriptions lock during a race condition when doKeepAliveCheck() also ran. A related deadlock was that the queued functions were executed while holding the lock taskQueueMutex. Together with the subscriptions lock, that was executed in the $SYS topic function, this also caused deadlocks.

Fix several deadlocks
One fix is client destruction happening on the correct thread (when kicking another one off with existing client ID). This caused deadlocks on the subscriptions lock during a race condition when doKeepAliveCheck() also ran. A related deadlock was that the queued functions were executed while holding the lock taskQueueMutex. Together with the subscriptions lock, that was executed in the $SYS topic function, this also caused deadlocks.
Wiebe Cazemier
1 parent af190a63
Showing 7 changed files with 123 additions and 18 deletions
client.cpp
client.h
mqttpacket.cpp
subscriptionstore.cpp
threaddata.cpp
threaddata.h
threadloop.cpp
@@ -222,7 +222,7 @@ void Client::writeMqttPacketAndBlameThisClient(const MqttPacket &amp;packet, const c
     }
     catch (std::exception &ex)
     {
-        threadData->removeClient(fd);
+        threadData->removeClientQueued(fd);
     }
 }
@@ -322,6 +322,11 @@ void Client::resetBuffersIfEligible()
     writebuf.resetSizeIfEligable(initialBufferSize);
 }
+void Client::setCleanSession(bool val)
+{
+    this->cleanSession = val;
+}
+
 #ifndef NDEBUG
 /**
  * @brief IoWrapper::setFakeUpgraded().
@@ -134,6 +134,8 @@ public:
     std::string getKeepAliveInfoString() const;
     void resetBuffersIfEligible();
+    void setCleanSession(bool val);
+
 #ifndef NDEBUG
     void setFakeUpgraded();
 #endif
@@ -412,7 +412,7 @@ void MqttPacket::handleDisconnect()
     sender->setDisconnectReason("MQTT Disconnect received.");
     sender->markAsDisconnecting();
     sender->clearWill();
-    sender->getThreadData()->removeClient(sender);
+    sender->getThreadData()->removeClientQueued(sender);
 }
 void MqttPacket::handleSubscribe()
@@ -202,6 +202,7 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
     if (client->getClientId().empty())
         throw ProtocolError("Trying to store client without an ID.");
+    bool originalClientDemandsSessionDestruction = false;
     std::shared_ptr<Session> session;
     auto session_it = sessionsById.find(client->getClientId());
     if (session_it != sessionsById.end())
@@ -215,14 +216,22 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
             if (cl)
             {
                 logger->logf(LOG_NOTICE, "Disconnecting existing client with id '%s'", cl->getClientId().c_str());
+                cl->setDisconnectReason("Another client with this ID connected");
+
+                // We have to set session to false, because it's no longer up to the destruction of that client
+                // to destroy the session. We either do it in this function, or not at all.
+                originalClientDemandsSessionDestruction = cl->getCleanSession();
+                cl->setCleanSession(false);
+
                 cl->setReadyForDisconnect();
-                cl->getThreadData()->removeClient(cl);
+                cl->getThreadData()->removeClientQueued(cl);
                 cl->markAsDisconnecting();
             }
+
         }
     }
-    if (!session || client->getCleanSession())
+    if (!session || client->getCleanSession() || originalClientDemandsSessionDestruction)
     {
         session.reset(new Session());
@@ -131,6 +131,36 @@ void ThreadData::publishStat(const std::string &amp;topic, uint64_t n)
     subscriptionStore->setRetainedMessage(topic, subtopics, payload, 0);
 }
+void ThreadData::removeQueuedClients()
+{
+    std::vector<int> fds;
+    fds.reserve(1024); // 1024 is arbitrary...
+
+    {
+        std::lock_guard<std::mutex> lck2(clientsToRemoveMutex);
+
+        for (const std::weak_ptr<Client> &c : clientsQueuedForRemoving)
+        {
+            std::shared_ptr<Client> client = c.lock();
+            if (client)
+            {
+                int fd = client->getFd();
+                fds.push_back(fd);
+            }
+        }
+
+        clientsQueuedForRemoving.clear();
+    }
+
+    {
+        std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
+        for(int fd : fds)
+        {
+            clients_by_fd.erase(fd);
+        }
+    }
+}
+
 void ThreadData::giveClient(std::shared_ptr<Client> client)
 {
     clients_by_fd_mutex.lock();
@@ -151,25 +181,70 @@ std::shared_ptr&lt;Client&gt; ThreadData::getClient(int fd)
     return this->clients_by_fd[fd];
 }
-void ThreadData::removeClient(std::shared_ptr<Client> client)
+void ThreadData::removeClientQueued(const std::shared_ptr<Client> &client)
 {
-    client->markAsDisconnecting();
+    bool wakeUpNeeded = true;
-    std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
-    clients_by_fd.erase(client->getFd());
+    {
+        std::lock_guard<std::mutex> locker(clientsToRemoveMutex);
+        wakeUpNeeded = clientsQueuedForRemoving.empty();
+        clientsQueuedForRemoving.push_front(client);
+    }
+
+    if (wakeUpNeeded)
+    {
+        auto f = std::bind(&ThreadData::removeQueuedClients, this);
+        std::lock_guard<std::mutex> lockertaskQueue(taskQueueMutex);
+        taskQueue.push_front(f);
+
+        wakeUpThread();
+    }
 }
-void ThreadData::removeClient(int fd)
+void ThreadData::removeClientQueued(int fd)
 {
-    std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
-    auto client_it = this->clients_by_fd.find(fd);
-    if (client_it != this->clients_by_fd.end())
+    bool wakeUpNeeded = true;
+    std::shared_ptr<Client> clientFound;
+
+    {
+        std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
+        auto client_it = this->clients_by_fd.find(fd);
+        if (client_it != this->clients_by_fd.end())
+        {
+            clientFound = client_it->second;
+        }
+    }
+
+    if (clientFound)
     {
-        client_it->second->markAsDisconnecting();
-        this->clients_by_fd.erase(fd);
+        {
+            std::lock_guard<std::mutex> locker(clientsToRemoveMutex);
+            wakeUpNeeded = clientsQueuedForRemoving.empty();
+            clientsQueuedForRemoving.push_front(clientFound);
+        }
+
+        if (wakeUpNeeded)
+        {
+            auto f = std::bind(&ThreadData::removeQueuedClients, this);
+            std::lock_guard<std::mutex> lockertaskQueue(taskQueueMutex);
+            taskQueue.push_front(f);
+
+            wakeUpThread();
+        }
     }
 }
+void ThreadData::removeClient(std::shared_ptr<Client> client)
+{
+    // This function is only for same-thread calling.
+    assert(pthread_self() == thread.native_handle());
+
+    client->markAsDisconnecting();
+
+    std::lock_guard<std::mutex> lck(clients_by_fd_mutex);
+    clients_by_fd.erase(client->getFd());
+}
+
 std::shared_ptr<SubscriptionStore> &ThreadData::getSubscriptionStore()
 {
     return subscriptionStore;
@@ -56,6 +56,8 @@ class ThreadData
     uint64_t sentMessageCountPrevious = 0;
     std::chrono::time_point<std::chrono::steady_clock> sentMessagePreviousTime = std::chrono::steady_clock::now();
+    std::mutex clientsToRemoveMutex;
+    std::forward_list<std::weak_ptr<Client>> clientsQueuedForRemoving;
     void reload(std::shared_ptr<Settings> settings);
     void wakeUpThread();
@@ -64,6 +66,8 @@ class ThreadData
     void publishStatsOnDollarTopic(std::vector<std::shared_ptr<ThreadData>> &threads);
     void publishStat(const std::string &topic, uint64_t n);
+    void removeQueuedClients();
+
 public:
     Settings settingsLocalCopy; // Is updated on reload, within the thread loop.
     Authentication authentication;
@@ -84,8 +88,9 @@ public:
     void giveClient(std::shared_ptr<Client> client);
     std::shared_ptr<Client> getClient(int fd);
+    void removeClientQueued(const std::shared_ptr<Client> &client);
+    void removeClientQueued(int fd);
     void removeClient(std::shared_ptr<Client> client);
-    void removeClient(int fd);
     std::shared_ptr<SubscriptionStore> &getSubscriptionStore();
     void initAuthPlugin();
@@ -64,12 +64,21 @@ void do_thread_work(ThreadData *threadData)
                     uint64_t eventfd_value = 0;
                     check<std::runtime_error>(read(fd, &eventfd_value, sizeof(uint64_t)));
-                    std::lock_guard<std::mutex> locker(threadData->taskQueueMutex);
-                    for(auto &f : threadData->taskQueue)
+                    std::forward_list<std::function<void()>> copiedTasks;
+
+                    {
+                        std::lock_guard<std::mutex> locker(threadData->taskQueueMutex);
+                        for(auto &f : threadData->taskQueue)
+                        {
+                            copiedTasks.push_front(std::move(f));
+                        }
+                        threadData->taskQueue.clear();
+                    }
+
+                    for(auto &f : copiedTasks)
                     {
                         f();
                     }
-                    threadData->taskQueue.clear();
                     continue;
                 }