[syncd] Make sure notification queue release memory when drained

syncd/NotificationQueue.cpp

@@ -18,7 +18,7 @@ NotificationQueue::NotificationQueue(
 {
     SWSS_LOG_ENTER();
 
-    // empty;
+    m_queue = std::make_shared<std::queue<swss::KeyOpFieldsValuesTuple>>();
 }
 
 NotificationQueue::~NotificationQueue()
@@ -34,7 +34,9 @@ bool NotificationQueue::enqueue(
     MUTEX;
 
     SWSS_LOG_ENTER();
+
     bool candidateToDrop = false;
+
     std::string currentEvent;
 
     /*
@@ -49,8 +51,10 @@ bool NotificationQueue::enqueue(
      * will also be dropped regardless of its event type to protect the device from crashing due to
      * running out of memory
      */
-    auto queueSize = m_queue.size();
+    auto queueSize = m_queue->size();
+
     currentEvent = kfvKey(item);
+
     if (currentEvent == m_lastEvent)
     {
         m_lastEventCount++;
@@ -60,12 +64,15 @@ bool NotificationQueue::enqueue(
         m_lastEventCount = 1;
         m_lastEvent = currentEvent;
     }
+
     if (queueSize >= m_queueSizeLimit)
     {
-        /* Too many queued up already check if notification fits condition to e dropped
+        /*
+         * Too many queued up already check if notification fits condition to e dropped
          * 1. All FDB events should be dropped at this point.
          * 2. All other notification events will start to drop if it reached the consecutive threshold limit
          */
+
         if (currentEvent == SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT)
         {
             candidateToDrop = true;
@@ -81,7 +88,7 @@ bool NotificationQueue::enqueue(
 
     if (!candidateToDrop)
     {
-        m_queue.push(item);
+        m_queue->push(item);
 
         return true;
     }
@@ -106,14 +113,38 @@ bool NotificationQueue::tryDequeue(
 
     SWSS_LOG_ENTER();
 
-    if (m_queue.empty())
+    if (m_queue->empty())
     {
         return false;
     }
 
-    item = m_queue.front();
+    item = m_queue->front();
 
-    m_queue.pop();
+    m_queue->pop();
+
+    if (m_queue->empty())
+    {
+        /*
+         * Since there could be burst of notifications, that allocated memory
+         * can be over 2GB, but when queue will be drained that memory will not
+         * be automatically released. Underlying deque container contains
+         * function shrink_to_fit but that is just a request, and usually this
+         * function does nothing.
+         *
+         * Make sure we will destroy queue and allocate new one. Assignment
+         * operator is not enough here, since internal deque container will not
+         * release memory under assignment. While making sure queue is deleted
+         * all memory will be released.
+         *
+         * Downside of this approach is that even if we will have steady stream
+         * of single notifications, each time we will allocate new queue.
+         * Partial solution for this could allocating new queue only when
+         * previous queue exceeded some size limit, for example 128 items.
+         */
+        m_queue = nullptr;
+
+        m_queue = std::make_shared<std::queue<swss::KeyOpFieldsValuesTuple>>();
+    }
 
     return true;
 }
@@ -124,5 +155,5 @@ size_t NotificationQueue::getQueueSize()
 
     SWSS_LOG_ENTER();
 
-    return m_queue.size();
+    return m_queue->size();
 }

syncd/NotificationQueue.h

@@ -2,13 +2,14 @@
 
 extern "C" {
 #include <sai.h>
-#include<saimetadata.h>
+#include <saimetadata.h>
 }
 
 #include "swss/table.h"
 
 #include <queue>
 #include <mutex>
+#include <memory>
 
 /**
  * @brief Default notification queue size limit.
@@ -54,7 +55,7 @@ namespace syncd
 
             std::mutex m_mutex;
 
-            std::queue<swss::KeyOpFieldsValuesTuple> m_queue;
+            std::shared_ptr<std::queue<swss::KeyOpFieldsValuesTuple>> m_queue;
 
             size_t m_queueSizeLimit;
 

tests/aspell.en.pws

@@ -477,3 +477,4 @@ TWAMP
 saiproxy
 submodule
 Enqueue
+deque

unittest/syncd/TestNotificationQueue.cpp

@@ -66,3 +66,19 @@ TEST(NotificationQueue, EnqueueLimitTest)
     }
 }
 
+TEST(NotificationQueue, tryDequeue)
+{
+    syncd::NotificationQueue nq(5, 3);
+
+    EXPECT_EQ(nq.getQueueSize(), 0);
+
+    swss::KeyOpFieldsValuesTuple item;
+
+    nq.enqueue(item);
+
+    EXPECT_EQ(nq.getQueueSize(), 1);
+
+    EXPECT_EQ(nq.tryDequeue(item), true);
+
+    EXPECT_EQ(nq.getQueueSize(), 0);
+}