Use ordered map for session removals

This is way faster.

Use ordered map for session removals
This is way faster.
Wiebe Cazemier · Wiebe Cazemier
1 parent c5f8e527
Showing 2 changed files with 20 additions and 55 deletions
subscriptionstore.cpp
subscriptionstore.h
@@ -666,19 +666,23 @@ void SubscriptionStore::removeExpiredSessionsClients()
         auto it = queuedSessionRemovals.begin();
         while (it != queuedSessionRemovals.end())
         {
-            QueuedSessionRemoval &qsr = *it;
-            std::shared_ptr<Session> session = (*it).getSession();
-            if (session)
+            const std::chrono::time_point<std::chrono::steady_clock> &removeAt = it->first;
+
+            if (removeAt > now)
             {
-                if (qsr.getExpiresAt() > now)
-                {
-                    break;
-                }
+                break;
+            }
+
+            std::vector<std::weak_ptr<Session>> &sessionsFromSlot = it->second;
+
+            for (std::weak_ptr<Session> ses : sessionsFromSlot)
+            {
+                std::shared_ptr<Session> lockedSession = ses.lock();
                 // A session could have been picked up again, so we have to verify its expiration status.
-                if (!session->hasActiveClient())
+                if (lockedSession && !lockedSession->hasActiveClient())
                 {
-                    removeSession(session);
+                    removeSession(lockedSession);
                     removedSessions++;
                 }
             }
@@ -690,7 +694,7 @@ void SubscriptionStore::removeExpiredSessionsClients()
         queuedRemovalsLeft = queuedSessionRemovals.size();
     }
-    logger->logf(LOG_DEBUG, "Processed %d queued session removals, resuling in %d deleted expired sessions. %d queued removals in the future.",
+    logger->logf(LOG_DEBUG, "Processed %d queued session removals, resulting in %d deleted expired sessions. %d queued removals in the future.",
                  processedRemovals, removedSessions, queuedRemovalsLeft);
     if (lastTreeCleanup + std::chrono::minutes(30) < now)
@@ -705,7 +709,7 @@ void SubscriptionStore::removeExpiredSessionsClients()
 }
 /**
- * @brief SubscriptionStore::queueSessionRemoval places session efficiently in a sorted list that is periodically dequeued.
+ * @brief SubscriptionStore::queueSessionRemoval places session efficiently in a sorted map that is periodically dequeued.
  * @param session
  */
 void SubscriptionStore::queueSessionRemoval(const std::shared_ptr<Session> &session)
@@ -713,18 +717,11 @@ void SubscriptionStore::queueSessionRemoval(const std::shared_ptr&lt;Session&gt; &amp;sess
     if (!session)
         return;
-    QueuedSessionRemoval qsr(session);
-
-    auto comp = [](const QueuedSessionRemoval &a, const QueuedSessionRemoval &b)
-    {
-        return a.getExpiresAt() < b.getExpiresAt();
-    };
-
+    std::chrono::time_point<std::chrono::steady_clock> removeAt = std::chrono::steady_clock::now() + std::chrono::seconds(session->getSessionExpiryInterval());
     session->setQueuedRemovalAt();
     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);
+    queuedSessionRemovals[removeAt].push_back(session);
 }
 int64_t SubscriptionStore::getRetainedMessageCount() const
@@ -1026,23 +1023,6 @@ RetainedMessageNode *RetainedMessageNode::getChildren(const std::string &amp;subtopi
     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();
-}
-
 QueuedWill::QueuedWill(const std::shared_ptr<WillPublish> &will, const std::shared_ptr<Session> &session) :
     will(will),
     session(session),
@@ -22,6 +22,8 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include <forward_list>
 #include <list>
 #include <mutex>
+#include <map>
+#include <vector>
 #include <pthread.h>
 #include "forward_declarations.h"
@@ -84,23 +86,6 @@ class RetainedMessageNode
     RetainedMessageNode *getChildren(const std::string &subtopic) const;
 };
-/**
- * @brief A QueuedSessionRemoval is a sort of delayed request for removal. They are kept in a sorted list for fast insertion,
- * and fast dequeueing of expired entries from the start.
- *
- * You can have multiple of these in the pending list. If a client has picked up the session again, the removal is not executed.
- */
-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 QueuedWill
 {
     std::weak_ptr<WillPublish> will;
@@ -128,7 +113,7 @@ class SubscriptionStore
     const std::unordered_map<std::string, std::shared_ptr<Session>> &sessionsByIdConst;
     std::mutex queuedSessionRemovalsMutex;
-    std::list<QueuedSessionRemoval> queuedSessionRemovals;
+    std::map<std::chrono::time_point<std::chrono::steady_clock>, std::vector<std::weak_ptr<Session>>> queuedSessionRemovals;
     pthread_rwlock_t retainedMessagesRwlock = PTHREAD_RWLOCK_INITIALIZER;
     RetainedMessageNode retainedMessagesRoot;