[swss-common] Enhanced NotificationConsumer queue to avoid memory leaks in some cases

Why I did it

In the current notification mechanism, the NotificationConsumer stores notification messages in bulk to an internal queue.
However, when the various Orch consumers receive notifications, they only retrieve one notification from the queue for processing at a time.
This approach has a potential risk of memory leaks. If the device is hit by a large number of FDB move or link up/down flapping events,
a significant increase in memory consumption by the orchagent can be observed,
eventually leading to a system crash or hang.
(Testing has been done on SONiC 202111, where the orchagent's memory consumption reached up to 10GB and continued to increase)

How I did it

Modified the internal queue of the NotificationConsumer class to replace std::queue.
Now, when processing existing duplicate messages, the NotificationConsumer will no longer enqueue them and
will instead move the existing message to the end of the queue (as a penalty for DDOS).
This significantly reduces the number of duplicate notify messages, and the time complexity for both lookup and insertion in the new queue is O(1).

Author: yongxin927

common/notificationconsumer.h

@@ -4,6 +4,8 @@
 #include <string>
 #include <vector>
 #include <queue>
+#include <unordered_map>
+#include <list>
 
 #include <hiredis/hiredis.h>
 
@@ -18,6 +20,54 @@ namespace swss {
 
 static constexpr size_t DEFAULT_NC_POP_BATCH_SIZE = 2048;
 
+template <typename T>
+class Queue
+{
+private:
+    std::list<T> dq;
+    std::unordered_map<T, typename std::list<T>::iterator> map;
+
+public:
+    void push(const T& value)
+    {
+        auto it = map.find(value);
+        if (it != map.end())
+            dq.erase(it->second);
+        dq.push_back(value);
+        map[value] = --dq.end();
+    }
+
+    void pop()
+    {
+        if (dq.empty())
+            return;
+        map.erase(dq.front());
+        dq.pop_front();
+    }
+
+    T front() const
+    {
+        if (dq.empty())
+            throw std::runtime_error("Queue is empty");
+        return dq.front();
+    }
+
+    size_t size() const {
+        return dq.size();
+    }
+
+    void swap(Queue& other)
+    {
+        std::swap(dq, other.dq);
+        std::swap(map, other.map);
+    }
+
+    bool empty() const
+    {
+        return dq.empty();
+    }
+};
+
 class NotificationConsumer : public Selectable
 {
 public:
@@ -55,7 +105,7 @@ class NotificationConsumer : public Selectable
     swss::DBConnector *m_db;
     swss::DBConnector *m_subscribe;
     std::string m_channel;
-    std::queue<std::string> m_queue;
+    Queue<std::string> m_queue;
 };
 
 }