Move ntf_queue to proper NotificationQueue class (sonic-net#539)

* Move ntf_queue to proper NotificationQueue class

* add extra aspell exception

* Address comments

Author: kcudnik

syncd/Makefile.am

@@ -22,7 +22,8 @@ syncd_SOURCES = \
 				syncd_notifications.cpp \
 				syncd_applyview.cpp \
 				syncd_flex_counter.cpp \
-				TimerWatchdog.cpp
+				TimerWatchdog.cpp \
+				NotificationQueue.cpp
 
 syncd_CPPFLAGS = $(DBGFLAGS) $(AM_CPPFLAGS) $(CFLAGS_COMMON) $(SAIFLAGS)
 syncd_LDADD = -lhiredis -lswsscommon $(SAILIB) -lpthread -L$(top_srcdir)/meta/.libs -lsaimetadata -lsaimeta -ldl
@@ -48,7 +49,8 @@ tests_SOURCES = \
 				syncd_notifications.cpp \
 				syncd_applyview.cpp \
 				syncd_flex_counter.cpp \
-				TimerWatchdog.cpp
+				TimerWatchdog.cpp \
+				NotificationQueue.cpp
 
 tests_CPPFLAGS = $(DBGFLAGS) $(AM_CPPFLAGS) $(CFLAGS_COMMON)
 tests_LDADD = -lhiredis -lswsscommon -lpthread -L$(top_srcdir)/lib/src/.libs -lsairedis -L$(top_srcdir)/meta/.libs -lsaimetadata -lsaimeta

syncd/NotificationQueue.cpp

@@ -0,0 +1,83 @@
+#include "NotificationQueue.h"
+
+#define NOTIFICATION_QUEUE_DROP_COUNT_INDICATOR (1000)
+
+NotificationQueue::NotificationQueue(
+        _In_ size_t queueLimit):
+    m_queueSizeLimit(queueLimit),
+    m_dropCount(0)
+{
+    SWSS_LOG_ENTER();
+
+    // empty;
+}
+
+NotificationQueue::~NotificationQueue()
+{
+    SWSS_LOG_ENTER();
+
+    // empty
+}
+
+bool NotificationQueue::enqueue(
+        _In_ const swss::KeyOpFieldsValuesTuple& item)
+{
+    SWSS_LOG_ENTER();
+
+    /*
+     * If the queue exceeds the limit, then drop all further FDB events This is
+     * a temporary solution to handle high memory usage by syncd and the
+     * notification queue keeps growing. The permanent solution would be to
+     * make this stateful so that only the *latest* event is published.
+     */
+
+    if (getQueueSize() < m_queueSizeLimit || kfvOp(item) != "fdb_event") // TODO use enum instead of strings
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+
+        m_queue.push(item);
+
+        return true;
+    }
+
+    m_dropCount++;
+
+    if (!(m_dropCount % NOTIFICATION_QUEUE_DROP_COUNT_INDICATOR))
+    {
+        SWSS_LOG_NOTICE(
+                "Too many messages in queue (%zu), dropped %zu FDB events!",
+                getQueueSize(),
+                m_dropCount);
+    }
+
+    return false;
+}
+
+bool NotificationQueue::tryDequeue(
+        _Out_ swss::KeyOpFieldsValuesTuple& item)
+{
+    SWSS_LOG_ENTER();
+
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    if (m_queue.empty())
+    {
+        return false;
+    }
+
+    item = m_queue.front();
+
+    m_queue.pop();
+
+    return true;
+}
+
+size_t NotificationQueue::getQueueSize()
+{
+    SWSS_LOG_ENTER();
+
+    std::lock_guard<std::mutex> lock(m_mutex);
+
+    return m_queue.size();
+}
+

syncd/NotificationQueue.h

@@ -0,0 +1,51 @@
+#pragma once
+
+extern "C" {
+#include <sai.h>
+}
+
+#include "swss/table.h"
+
+#include <queue>
+#include <mutex>
+
+/**
+ * @brief Default notification queue size limit.
+ *
+ * Value based on typical L2 deployment with 256k MAC entries and
+ * some extra buffer for other events like port-state, q-deadlock etc
+ *
+ * TODO: move to config, also this limit only applies to fdb notifications
+ */
+#define DEFAULT_NOTIFICATION_QUEUE_SIZE_LIMIT (300000)
+
+class NotificationQueue
+{
+    public:
+
+        NotificationQueue(
+                _In_ size_t limit = DEFAULT_NOTIFICATION_QUEUE_SIZE_LIMIT);
+
+        virtual ~NotificationQueue();
+
+    public:
+
+        bool enqueue(
+                _In_ const swss::KeyOpFieldsValuesTuple& msg);
+
+        bool tryDequeue(
+                _Out_ swss::KeyOpFieldsValuesTuple& msg);
+
+        size_t getQueueSize();
+
+    private:
+
+        std::mutex m_mutex;
+
+        std::queue<swss::KeyOpFieldsValuesTuple> m_queue;
+
+        size_t m_queueSizeLimit;
+
+        size_t m_dropCount;
+};
+

syncd/syncd_notifications.cpp

@@ -6,6 +6,8 @@
 #include <memory>
 #include <condition_variable>
 
+#include "NotificationQueue.h"
+
 void send_notification(
         _In_ std::string op,
         _In_ std::string data,
@@ -602,95 +604,14 @@ void processNotification(
 
 std::condition_variable cv;
 
-class ntf_queue_t
-{
-public:
-    ntf_queue_t() { }
-    bool enqueue(swss::KeyOpFieldsValuesTuple msg);
-    bool tryDequeue(swss::KeyOpFieldsValuesTuple& msg);
-    size_t queueStats()
-    {
-        SWSS_LOG_ENTER();
-
-        std::lock_guard<std::mutex> lock(queue_mutex);
-
-        return ntf_queue.size();
-    }
-
-private:
-    std::mutex queue_mutex;
-    std::queue<swss::KeyOpFieldsValuesTuple> ntf_queue;
-
-    /*
-     * Value based on typical L2 deployment with 256k MAC entries and
-     * some extra buffer for other events like port-state, q-deadlock etc
-     */
-    const size_t limit = 300000;
-};
-
-
 /*
  * Make sure that notification queue pointer is populated before we start
  * thread, and before we create_switch, since at switch_create we can start
  * receiving fdb_notifications which will arrive on different thread and
- * will call queueStats() when queue pointer could be null (this=0x0).
+ * will call getQueueSize() when queue pointer could be null (this=0x0).
  */
 
-static std::unique_ptr<ntf_queue_t> ntf_queue_hdlr = std::unique_ptr<ntf_queue_t>(new ntf_queue_t);
-
-bool ntf_queue_t::tryDequeue(
-        _Out_ swss::KeyOpFieldsValuesTuple &item)
-{
-    std::lock_guard<std::mutex> lock(queue_mutex);
-
-    SWSS_LOG_ENTER();
-
-    if (ntf_queue.empty())
-    {
-        return false;
-    }
-
-    item = ntf_queue.front();
-
-    ntf_queue.pop();
-
-    return true;
-}
-
-bool ntf_queue_t::enqueue(
-        _In_ swss::KeyOpFieldsValuesTuple item)
-{
-    SWSS_LOG_ENTER();
-
-    std::string notification = kfvKey(item);
-
-    /*
-     * If the queue exceeds the limit, then drop all further FDB events This is
-     * a temporary solution to handle high memory usage by syncd and the
-     * notification queue keeps growing. The permanent solution would be to
-     * make this stateful so that only the *latest* event is published.
-     */
-
-    if (queueStats() < limit || notification != "fdb_event")
-    {
-        std::lock_guard<std::mutex> lock(queue_mutex);
-
-        ntf_queue.push(item);
-        return true;
-    }
-
-    static uint32_t log_count = 0;
-
-    if (!(log_count % 1000))
-    {
-        SWSS_LOG_NOTICE("Too many messages in queue (%zu), dropped %u FDB events!",
-                         queueStats(), (log_count+1));
-    }
-
-    ++log_count;
-
-    return false;
-}
+static auto g_notificationQueue = std::make_shared<NotificationQueue>();
 
 void enqueue_notification(
         _In_ std::string op,
@@ -703,7 +624,7 @@ void enqueue_notification(
 
     swss::KeyOpFieldsValuesTuple item(op, data, entry);
 
-    if(ntf_queue_hdlr->enqueue(item))
+    if (g_notificationQueue->enqueue(item))
     {
         cv.notify_all();
     }
@@ -808,7 +729,7 @@ void ntf_process_function()
 
         swss::KeyOpFieldsValuesTuple item;
 
-        while (ntf_queue_hdlr->tryDequeue(item))
+        while (g_notificationQueue->tryDequeue(item))
         {
             processNotification(item);
         }

tests/Makefile.am

@@ -18,7 +18,8 @@ vssyncd_SOURCES = \
 				../syncd/syncd_notifications.cpp \
 				../syncd/syncd_applyview.cpp \
 				../syncd/syncd_flex_counter.cpp \
-				../syncd/TimerWatchdog.cpp
+				../syncd/TimerWatchdog.cpp \
+				../syncd/NotificationQueue.cpp
 
 vssyncd_CPPFLAGS = $(DBGFLAGS) $(AM_CPPFLAGS) $(CFLAGS_COMMON) $(SAIFLAGS)
 vssyncd_LDADD = -lhiredis -lswsscommon $(SAILIB) -lpthread -L$(top_srcdir)/meta/.libs -lsaimetadata -lsaimeta -ldl

tests/aspell.en.pws

@@ -71,6 +71,7 @@ FIXME
 FlexCounter
 genetlink
 getInstance
+getQueueSize
 getSwitchId
 getVid
 github
@@ -91,8 +92,8 @@ INSEG
 ip
 IP
 IPG
-IPGs
 ipgs
+IPGs
 ipmc
 IPv
 isobjectid