Queue session removals

This replaces the old style checking of expired sessions once every 10 minutes or so.

Queue session removals
This replaces the old style checking of expired sessions once every 10 minutes or so.
Wiebe Cazemier
1 parent 4cd9300c
Showing 5 changed files with 94 additions and 52 deletions
client.cpp
subscriptionstore.cpp
subscriptionstore.h
types.cpp
types.h
@@ -69,10 +69,13 @@ Client::~Client()
         close(fd);
     }
-    // MQTT-3.1.2-6
     if (session->getDestroyOnDisconnect())
     {
-        store->removeSession(clientid);
+        store->removeSession(session);
+    }
+    else
+    {
+        store->queueSessionRemoval(session);
     }
 }
@@ -269,24 +269,9 @@ bool SubscriptionStore::sessionPresent(const std::string &amp;clientid)
     return result;
 }
-/**
- * @brief SubscriptionStore::purgeEmptyWills doesn't lock a mutex, because it's a helper for elsewhere.
- */
-void SubscriptionStore::purgeEmptyWills()
-{
-    auto it = pendingWillMessages.begin();
-    while (it != pendingWillMessages.end())
-    {
-        std::shared_ptr<Publish> p = (*it).lock();
-        if (!p)
-        {
-            it = pendingWillMessages.erase(it);
-        }
-    }
-}
-
 void SubscriptionStore::sendQueuedWillMessages()
 {
+    const auto now = std::chrono::steady_clock::now();
     std::lock_guard<std::mutex>(this->pendingWillsMutex);
     auto it = pendingWillMessages.begin();
@@ -295,7 +280,7 @@ void SubscriptionStore::sendQueuedWillMessages()
         std::shared_ptr<Publish> p = (*it).lock();
         if (p)
         {
-            if (p->createdAt + std::chrono::seconds(p->will_delay) > std::chrono::steady_clock::now())
+            if (p->createdAt + std::chrono::seconds(p->will_delay) > now)
                 break;
             logger->logf(LOG_DEBUG, "Sending delayed will on topic '%s'.", p->topic.c_str() );
@@ -306,14 +291,15 @@ void SubscriptionStore::sendQueuedWillMessages()
     }
 }
-void SubscriptionStore::queueWillMessage(std::shared_ptr<Publish> &willMessage)
+void SubscriptionStore::queueWillMessage(std::shared_ptr<Publish> &willMessage, bool forceNow)
 {
     if (!willMessage)
         return;
-    logger->logf(LOG_DEBUG, "Queueing will on topic '%s', with delay %d seconds.", willMessage->topic.c_str(), willMessage->will_delay );
+    const int delay = forceNow ? 0 : willMessage->will_delay;
+    logger->logf(LOG_DEBUG, "Queueing will on topic '%s', with delay %d seconds.", willMessage->topic.c_str(), delay );
-    if (willMessage->will_delay == 0)
+    if (delay == 0)
     {
         PublishCopyFactory factory(willMessage.get());
         queuePacketAtSubscribers(factory);
@@ -577,13 +563,20 @@ int SubscriptionNode::cleanSubscriptions()
     return subscribers.size() + subscribersLeftInChildren;
 }
-void SubscriptionStore::removeSession(const std::string &clientid)
+void SubscriptionStore::removeSession(const std::shared_ptr<Session> &session)
 {
+    const std::string &clientid = session->getClientId();
+    logger->logf(LOG_DEBUG, "Removing session of client '%s'.", clientid.c_str());
+
+    std::shared_ptr<Publish> &will = session->getWill();
+    if (will)
+    {
+        queueWillMessage(will, true);
+    }
+
     RWLockGuard lock_guard(&subscriptionsRwlock);
     lock_guard.wrlock();
-    logger->logf(LOG_DEBUG, "Removing session of client '%s'.", clientid.c_str());
-
     auto session_it = sessionsById.find(clientid);
     if (session_it != sessionsById.end())
     {
@@ -600,37 +593,59 @@ void SubscriptionStore::removeExpiredSessionsClients()
 {
     logger->logf(LOG_DEBUG, "Cleaning out old sessions");
-    RWLockGuard lock_guard(&subscriptionsRwlock);
-    lock_guard.wrlock();
+    const std::chrono::time_point<std::chrono::steady_clock> now = std::chrono::steady_clock::now();
-    auto session_it = sessionsById.begin();
-    while (session_it != sessionsById.end())
     {
-        std::shared_ptr<Session> &session = session_it->second;
+        std::lock_guard<std::mutex>(this->queuedSessionRemovalsMutex);
-        if (session->hasExpired())
+        auto it = queuedSessionRemovals.begin();
+        while (it != queuedSessionRemovals.end())
         {
-            logger->logf(LOG_DEBUG, "Removing expired session from store %s", session->getClientId().c_str());
-            std::shared_ptr<Publish> &will = session->getWill();
-            if (will)
+            QueuedSessionRemoval &qsr = *it;
+            std::shared_ptr<Session> session = (*it).getSession();
+            if (session)
             {
-                will->will_delay = 0;
-                queueWillMessage(will);
+                if (qsr.getExpiresAt() > now)
+                {
+                    logger->logf(LOG_DEBUG, "Breaking from sorted list of queued session removals. %d left in the future.", queuedSessionRemovals.size());
+                    break;
+                }
+
+                // A session could have been picked up again, so we have to verify its expiration status.
+                if (session->hasExpired())
+                {
+                    removeSession(session);
+                }
             }
-            session_it = sessionsById.erase(session_it);
+            it = queuedSessionRemovals.erase(it);
         }
-        else
-            session_it++;
     }
-    if (lastTreeCleanup + std::chrono::minutes(30) < std::chrono::steady_clock::now())
+    if (lastTreeCleanup + std::chrono::minutes(30) < now)
     {
+        RWLockGuard lock_guard(&subscriptionsRwlock);
+        lock_guard.wrlock();
+
         logger->logf(LOG_NOTICE, "Rebuilding subscription tree");
         root.cleanSubscriptions();
-        lastTreeCleanup = std::chrono::steady_clock::now();
+        lastTreeCleanup = now;
     }
 }
+void SubscriptionStore::queueSessionRemoval(const std::shared_ptr<Session> &session)
+{
+    QueuedSessionRemoval qsr(session);
+
+    auto comp = [](const QueuedSessionRemoval &a, const QueuedSessionRemoval &b)
+    {
+        return a.getExpiresAt() < b.getExpiresAt();
+    };
+
+    std::lock_guard<std::mutex>(this->queuedSessionRemovalsMutex);
+    auto pos = std::upper_bound(this->queuedSessionRemovals.begin(), this->queuedSessionRemovals.end(), qsr, comp);
+    this->queuedSessionRemovals.insert(pos, qsr);
+}
+
 int64_t SubscriptionStore::getRetainedMessageCount() const
 {
     return retainedMessageCount;
@@ -927,3 +942,20 @@ RetainedMessageNode *RetainedMessageNode::getChildren(const std::string &amp;subtopi
         return it->second.get();
     return nullptr;
 }
+
+QueuedSessionRemoval::QueuedSessionRemoval(const std::shared_ptr<Session> &session) :
+    session(session),
+    expiresAt(std::chrono::steady_clock::now() + std::chrono::seconds(session->getSessionExpiryInterval()))
+{
+
+}
+
+std::chrono::time_point<std::chrono::steady_clock> QueuedSessionRemoval::getExpiresAt() const
+{
+    return this->expiresAt;
+}
+
+std::shared_ptr<Session> QueuedSessionRemoval::getSession() const
+{
+    return session.lock();
+}
@@ -84,6 +84,17 @@ class RetainedMessageNode
     RetainedMessageNode *getChildren(const std::string &subtopic) const;
 };
+class QueuedSessionRemoval
+{
+    std::weak_ptr<Session> session;
+    std::chrono::time_point<std::chrono::steady_clock> expiresAt;
+
+public:
+    QueuedSessionRemoval(const std::shared_ptr<Session> &session);
+    std::chrono::time_point<std::chrono::steady_clock> getExpiresAt() const;
+    std::shared_ptr<Session> getSession() const;
+};
+
 class SubscriptionStore
 {
 #ifdef TESTING
@@ -96,6 +107,9 @@ class SubscriptionStore
     std::unordered_map<std::string, std::shared_ptr<Session>> sessionsById;
     const std::unordered_map<std::string, std::shared_ptr<Session>> &sessionsByIdConst;
+    std::mutex queuedSessionRemovalsMutex;
+    std::list<QueuedSessionRemoval> queuedSessionRemovals;
+
     pthread_rwlock_t retainedMessagesRwlock = PTHREAD_RWLOCK_INITIALIZER;
     RetainedMessageNode retainedMessagesRoot;
     RetainedMessageNode retainedMessagesRootDollar;
@@ -121,8 +135,6 @@ class SubscriptionStore
     void countSubscriptions(SubscriptionNode *this_node, int64_t &count) const;
     SubscriptionNode *getDeepestNode(const std::string &topic, const std::vector<std::string> &subtopics);
-
-    void purgeEmptyWills();
 public:
     SubscriptionStore();
@@ -133,14 +145,14 @@ public:
     bool sessionPresent(const std::string &clientid);
     void sendQueuedWillMessages();
-    void queueWillMessage(std::shared_ptr<Publish> &willMessage);
+    void queueWillMessage(std::shared_ptr<Publish> &willMessage, bool forceNow = false);
     void queuePacketAtSubscribers(PublishCopyFactory &copyFactory, bool dollar = false);
     uint64_t giveClientRetainedMessages(const std::shared_ptr<Client> &client, const std::shared_ptr<Session> &ses,
                                         const std::vector<std::string> &subscribeSubtopics, char max_qos);
     void setRetainedMessage(const std::string &topic, const std::vector<std::string> &subtopics, const std::string &payload, char qos);
-    void removeSession(const std::string &clientid);
+    void removeSession(const std::shared_ptr<Session> &session);
     void removeExpiredSessionsClients();
     int64_t getRetainedMessageCount() const;
@@ -152,6 +164,8 @@ public:
     void saveSessionsAndSubscriptions(const std::string &filePath);
     void loadSessionsAndSubscriptions(const std::string &filePath);
+
+    void queueSessionRemoval(const std::shared_ptr<Session> &session);
 };
 #endif // SUBSCRIPTIONSTORE_H
@@ -134,12 +134,6 @@ void PublishBase::setClientSpecificProperties()
         propertyBuilder->writeMessageExpiryInterval(newExpiresAfter.count());
 }
-bool PublishBase::hasExpired() const
-{
-    auto now = std::chrono::steady_clock::now();
-    return (createdAt + expiresAfter) > now;
-}
-
 Publish::Publish(const Publish &other) :
     PublishBase(other)
 {
@@ -209,7 +209,6 @@ public:
     PublishBase(const std::string &topic, const std::string &payload, char qos);
     size_t getLengthWithoutFixedHeader() const;
     void setClientSpecificProperties();
-    bool hasExpired() const;
 };
 class Publish : public PublishBase