[dhcpmon] Write COUNTERS_DB without blocking packet handling

src/dhcp_mon.cpp

@@ -212,9 +212,14 @@ static void cleanup_stale_db_counters()
 static void signal_callback(evutil_socket_t fd, short event, void *arg)
 {
     syslog(LOG_INFO, "Received signal: %s", strsignal(fd));
-
-    dhcp_devman_print_all_status(DHCP_COUNTERS_CURRENT);
-    dhcp_devman_print_all_status(DHCP_COUNTERS_CURRENT_V6);
+
+    {
+        counter_state_write_lock counter_lock(std::try_to_lock);
+        if (counter_lock.owns_lock()) {
+            dhcp_devman_print_all_status(DHCP_COUNTERS_CURRENT);
+            dhcp_devman_print_all_status(DHCP_COUNTERS_CURRENT_V6);
+        }
+    }
 
     if ((fd == SIGTERM) || (fd == SIGINT)) {
         syslog(LOG_INFO, "Received signal to stop dhcpmon");
@@ -224,7 +229,13 @@ static void signal_callback(evutil_socket_t fd, short event, void *arg)
         // we need to sync cache counter from COUNTERS_DB
         syslog(LOG_INFO, "Received signal to stop writing to DB counter");
         std::lock_guard<std::mutex> lock(db_sync_mutex);
-        sock_mgr_pause_write_cache_to_db();
+        {
+            counter_state_write_lock counter_lock;
+            if (!counter_lock.owns_lock()) {
+                return;
+            }
+            sock_mgr_pause_write_cache_to_db();
+        }
         syslog(LOG_INFO, "Stopped writing to DB counter");
         mStateDbPtr->hset(STATE_DB_COUNTER_UPDATE_PREFIX + downstream_ifname, "pause_write_to_db", "done");
         mStateDbPtr->hset(STATE_DB_COUNTER_UPDATE_V6_PREFIX + downstream_ifname, "pause_write_to_db", "done");
@@ -261,6 +272,10 @@ static void update_cache_counter_callback(evutil_socket_t fd, short event, void
     syslog(LOG_INFO, "Start updating %s cache counter from DB counter", sock_info.name);
 
     std::lock_guard<std::mutex> lock(db_sync_mutex);
+    counter_state_write_lock counter_lock;
+    if (!counter_lock.owns_lock()) {
+        return;
+    }
 
     // can only sync db to cache counter and db updater is paused, otherwise its unexpected
     if (!sock_info.pause_write_cache_to_db) {
@@ -344,9 +359,9 @@ static void update_cache_counter_callback(evutil_socket_t fd, short event, void
 
         updated_intfs.insert(ifname);
     }
-    syslog(LOG_INFO, "Processing DB entry of %sfor downstream vlan %s",
+    syslog(LOG_INFO, "Processing DB entry of %s for downstream vlan %s",
              all_ifname.c_str(), downstream_ifname.c_str());
-    syslog(LOG_INFO, "Skipped DB entry of %sbecause we are only interested in %s",
+    syslog(LOG_INFO, "Skipped DB entry of %s because we are only interested in %s",
              all_skipped_ifname.c_str(), downstream_ifname.c_str());
 
     // log any cache counter entry not appearing in db counter. This is highly unexpected.
@@ -392,6 +407,10 @@ static void update_cache_counter_callback(evutil_socket_t fd, short event, void
 static void timeout_callback(evutil_socket_t fd, short event, void *arg)
 {
     syslog_debug(LOG_INFO, "Received timeout signal for DHCP relay health check");
+    counter_state_write_lock counter_lock;
+    if (!counter_lock.owns_lock()) {
+        return;
+    }
 
     dhcp_devman_print_all_status_debug(DHCP_COUNTERS_CURRENT);
     dhcp_devman_print_all_status_debug(DHCP_COUNTERS_SNAPSHOT);
@@ -418,22 +437,30 @@ static void db_update_callback(evutil_socket_t fd, short event, void *arg)
 {
     syslog_debug(LOG_INFO, "Received db update signal");
     syslog_debug(LOG_INFO, "Sync cache counter to DB counter");
-    std::lock_guard<std::mutex> lock(db_sync_mutex);
-    // If there is clear counter going on and its been longer than expected
-    // consider the clear counter operation failed so we don't block db update forever
-    if (!sock_mgr_pause_write_cache_to_db_all_cleared() && last_update_time != default_time_point) {
-        auto now = std::chrono::steady_clock::now();
-        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - last_update_time);
-        if (elapsed.count() >= clear_counter_timeout) {
-            syslog(LOG_WARNING, "Clear counter going on for too long, abort clear counter");
-            sock_mgr_clear_pause_write_cache_to_db();
-        } else {
-            syslog(LOG_INFO, "Clear counter is ongoing, skip syncing write cache counter to DB counter");
+    socket_counters_t counters_by_socket;
+    std::unique_lock<std::mutex> lock(db_sync_mutex);
+    {
+        counter_state_write_lock counter_lock;
+        if (!counter_lock.owns_lock()) {
             return;
         }
+        // If there is clear counter going on and its been longer than expected
+        // consider the clear counter operation failed so we don't block db update forever
+        if (!sock_mgr_pause_write_cache_to_db_all_cleared() && last_update_time != default_time_point) {
+            auto now = std::chrono::steady_clock::now();
+            auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - last_update_time);
+            if (elapsed.count() >= clear_counter_timeout) {
+                syslog(LOG_WARNING, "Clear counter going on for too long, abort clear counter");
+                sock_mgr_clear_pause_write_cache_to_db();
+            } else {
+                syslog(LOG_INFO, "Clear counter is ongoing, skip syncing write cache counter to DB counter");
+                return;
+            }
+        }
+        counters_by_socket = sock_mgr_copy_cache_counters(counter_lock);
     }
     last_update_time = std::chrono::steady_clock::now();
-    sock_mgr_update_db_counters();
+    sock_mgr_update_db_counters(counters_by_socket);
     cleanup_stale_db_counters();
     syslog_debug(LOG_INFO, "Successfully synced cache counter to DB counter");
 }

src/packet_handler.cpp

@@ -16,6 +16,9 @@
 #include "dhcp_check_profile.h"   /** to get dhcp/v6 check profile */
 #include "util.h"
 
+// Bound one callback to keep health/DB events responsive during packet bursts.
+static constexpr int MAX_PACKETS_PER_CALLBACK = 64;
+
 /**
  * @code _increase_cache_counter(ifname, sock, type);
  * @brief helper function to increase cache counter. Simple increase of counter, no complications. In the event of
@@ -859,13 +862,22 @@ void packet_handler_v6(int sock, const std::string &ifname, const dhcp_device_co
 
 void callback_common(int fd, short event, void *arg)
 {
+    counter_state_read_lock counter_lock;
+    if (!counter_lock.owns_lock()) {
+        return;
+    }
     ssize_t buffer_sz;
     struct sockaddr_ll sll;
     socklen_t slen = sizeof(sll);
     sock_info_t &sock_info = sock_mgr_get_sock_info(fd);
 
-    while ((buffer_sz = recvfrom(fd, sock_info.buffer, sock_info.snaplen, MSG_DONTWAIT, (struct sockaddr *)&sll, &slen)) > 0) 
-    {
+    for (int packet_count = 0; packet_count < MAX_PACKETS_PER_CALLBACK; packet_count++) {
+        slen = sizeof(sll);
+        buffer_sz = recvfrom(fd, sock_info.buffer, sock_info.snaplen, MSG_DONTWAIT,
+                             (struct sockaddr *)&sll, &slen);
+        if (buffer_sz <= 0) {
+            break;
+        }
         char ifname_buf[IF_NAMESIZE];
         if (if_indextoname(sll.sll_ifindex, ifname_buf) == NULL) {
             syslog_debug(LOG_WARNING, "if_indextoname: invalid input interface index %d %s", sll.sll_ifindex, strerror(errno));

src/sock_mgr.cpp

@@ -11,7 +11,15 @@
 #include <unistd.h>
 #include <linux/if_ether.h>
 #include <linux/if_packet.h>
+#include <condition_variable>
+#include <mutex>
+#include <system_error>
 #include <sys/socket.h>
+#include <utility>
+
+#if __cplusplus < 201703L
+#error "dhcpmon counter synchronization requires C++17 or newer"
+#endif
 
 #include "sock_mgr.h"
 
@@ -40,10 +48,109 @@ static const char cache_counter_updater_tag[] = "CacheCounterUpdater";
 /* sock fd to sock_info mapping */
 std::unordered_map<int, sock_info_t> sock_map;
 
+static std::shared_timed_mutex counter_state_mutex;
+static std::atomic<unsigned int> counter_state_writers_pending{0};
+static std::mutex counter_state_wait_mutex;
+static std::condition_variable counter_state_wait_cv;
+
 extern std::shared_ptr<swss::DBConnector> mCountersDbPtr;
 
 extern std::string downstream_ifname;
 
+counter_state_write_lock::counter_state_write_lock()
+{
+    {
+        std::lock_guard<std::mutex> wait_lock(counter_state_wait_mutex);
+        counter_state_writers_pending.fetch_add(1, std::memory_order_acq_rel);
+        registered_writer = true;
+    }
+    try {
+        lock = std::unique_lock<std::shared_timed_mutex>(counter_state_mutex);
+    } catch (const std::system_error &e) {
+        bool notify = false;
+        {
+            std::lock_guard<std::mutex> wait_lock(counter_state_wait_mutex);
+            notify = counter_state_writers_pending.fetch_sub(1, std::memory_order_acq_rel) == 1;
+        }
+        if (notify) {
+            counter_state_wait_cv.notify_all();
+        }
+        registered_writer = false;
+        syslog(LOG_ALERT, "Failed to lock DHCP counter state: %s", e.what());
+    }
+}
+
+counter_state_write_lock::counter_state_write_lock(std::try_to_lock_t)
+{
+    try {
+        lock = std::unique_lock<std::shared_timed_mutex>(counter_state_mutex, std::try_to_lock);
+    } catch (const std::system_error &e) {
+        syslog(LOG_ALERT, "Failed to try-lock DHCP counter state: %s", e.what());
+    }
+}
+
+counter_state_write_lock::~counter_state_write_lock()
+{
+    if (!lock.owns_lock()) {
+        return;
+    }
+    bool notify = false;
+    if (registered_writer) {
+        std::lock_guard<std::mutex> wait_lock(counter_state_wait_mutex);
+        notify = counter_state_writers_pending.fetch_sub(1, std::memory_order_acq_rel) == 1;
+    }
+    if (notify) {
+        counter_state_wait_cv.notify_all();
+    }
+}
+
+bool counter_state_write_lock::owns_lock() const
+{
+    return lock.owns_lock();
+}
+
+counter_state_read_lock::counter_state_read_lock()
+{
+    if (counter_state_writers_pending.load(std::memory_order_acquire) == 0) {
+        try {
+            std::shared_lock<std::shared_timed_mutex> candidate(counter_state_mutex,
+                                                                std::try_to_lock);
+            if (candidate.owns_lock() &&
+                counter_state_writers_pending.load(std::memory_order_acquire) == 0) {
+                lock = std::move(candidate);
+                return;
+            }
+        } catch (const std::system_error &e) {
+            syslog(LOG_ALERT, "Failed to lock DHCP counter state for packet handling: %s", e.what());
+            return;
+        }
+    }
+
+    while (true) {
+        {
+            std::unique_lock<std::mutex> wait_lock(counter_state_wait_mutex);
+            counter_state_wait_cv.wait(wait_lock, [] {
+                return counter_state_writers_pending.load(std::memory_order_acquire) == 0;
+            });
+        }
+        try {
+            std::shared_lock<std::shared_timed_mutex> candidate(counter_state_mutex);
+            if (counter_state_writers_pending.load(std::memory_order_acquire) == 0) {
+                lock = std::move(candidate);
+                return;
+            }
+        } catch (const std::system_error &e) {
+            syslog(LOG_ALERT, "Failed to lock DHCP counter state for packet handling: %s", e.what());
+            return;
+        }
+    }
+}
+
+bool counter_state_read_lock::owns_lock() const
+{
+    return lock.owns_lock();
+}
+
 /**
  * @code opensocket();
  *
@@ -614,17 +721,37 @@ bool sock_mgr_all_cache_counters_initialized(const std::string &ifname)
     return true;
 }
 
-void sock_mgr_update_db_counters()
+socket_counters_t sock_mgr_copy_cache_counters(const counter_state_write_lock &counter_lock)
+{
+    socket_counters_t counters_by_socket;
+    if (!counter_lock.owns_lock()) {
+        syslog(LOG_ALERT, "Cannot copy DHCP counters without the counter-state writer lock");
+        return counters_by_socket;
+    }
+    counters_by_socket.reserve(sock_map.size());
+    for (const auto &[sock, info] : sock_map) {
+        counters_by_socket.emplace(sock, info.all_counters);
+    }
+    return counters_by_socket;
+}
+
+void sock_mgr_update_db_counters(const socket_counters_t &counters_by_socket)
 {
     syslog_debug(LOG_INFO, "Updating all cache counters to DB counters");
 
-    for (const auto &[sock, info] : sock_map) {
+    for (const auto &[sock, all_counters] : counters_by_socket) {
+        const auto sock_info = sock_map.find(sock);
+        if (sock_info == sock_map.end()) {
+            syslog(LOG_WARNING, "Skip DB counter snapshot for unknown socket %d", sock);
+            continue;
+        }
+        const sock_info_t &info = sock_info->second;
         syslog_debug(LOG_INFO, "Start updating socket %d %s DB counter from cache counter", sock, info.name);
         int msg_type_count = info.is_v6 ? DHCPV6_MESSAGE_TYPE_COUNT : DHCP_MESSAGE_TYPE_COUNT;
         const std::string *msg_type_name = info.is_v6 ? db_counter_name_v6 : db_counter_name;
         std::string all_ifname;
         std::string all_skipped_ifname;
-        for (const auto &[ifname, counter] : info.all_counters) {
+        for (const auto &[ifname, counter] : all_counters) {
             if (is_agg_counter(ifname) == true) {
                 all_skipped_ifname += ifname + ", ";
                 continue;
@@ -634,9 +761,9 @@ void sock_mgr_update_db_counters()
             std::string table_name = construct_counter_db_table_key(ifname, info.is_v6);
             mCountersDbPtr->hset(table_name, info.is_rx ? "RX" : "TX", value);
         }
-        syslog_debug(LOG_INFO, "Processing cache counter entry of %sfor downstream vlan %s",
+        syslog_debug(LOG_INFO, "Processing cache counter entry of %s for downstream vlan %s",
                      all_ifname.c_str(), downstream_ifname.c_str());
-        syslog_debug(LOG_INFO, "Skipped aggregated device counter entry of %sfor downstream vlan %s",
+        syslog_debug(LOG_INFO, "Skipped aggregated device counter entry of %s for downstream vlan %s",
                      all_skipped_ifname.c_str(), downstream_ifname.c_str());
     }
-}
+}

src/sock_mgr.h

@@ -9,7 +9,10 @@
 #ifndef SOCKET_MANAGER_H_
 #define SOCKET_MANAGER_H_
 
+#include <atomic>
+#include <mutex>
 #include <stdint.h>
+#include <shared_mutex>
 #include <string>
 #include <unordered_map>
 #include <linux/filter.h>
@@ -19,6 +22,7 @@
 
 typedef std::unordered_map<uint8_t, uint64_t> counter_t;
 typedef std::unordered_map<std::string, counter_t> all_counters_t;
+typedef std::unordered_map<int, all_counters_t> socket_counters_t;
 
 /** struct for socket information */
 typedef struct {
@@ -41,6 +45,31 @@ typedef struct {
 /** sock file descriptors, serve as the identifier of all related information described in sock_info_t */
 extern int rx_sock, tx_sock, rx_sock_v6, tx_sock_v6;
 
+class counter_state_write_lock
+{
+    public:
+        counter_state_write_lock();
+        explicit counter_state_write_lock(std::try_to_lock_t);
+        ~counter_state_write_lock();
+        bool owns_lock() const;
+        counter_state_write_lock(const counter_state_write_lock &) = delete;
+        counter_state_write_lock &operator=(const counter_state_write_lock &) = delete;
+
+    private:
+        std::unique_lock<std::shared_timed_mutex> lock;
+        bool registered_writer = false;
+};
+
+class counter_state_read_lock
+{
+    public:
+        counter_state_read_lock();
+        bool owns_lock() const;
+
+    private:
+        std::shared_lock<std::shared_timed_mutex> lock;
+};
+
 /** Initialize socket manager with given snaplen */
 int sock_mgr_init(uint32_t snaplen);
 
@@ -104,7 +133,10 @@ void sock_mgr_init_cache_counters(const std::string &ifname, uint8_t dhcp_messag
 /** Check if cache counters are initialized for given ifname for all sockets */
 bool sock_mgr_all_cache_counters_initialized(const std::string &ifname);
 
-/** Update database counters from cache counters for all sockets */
-void sock_mgr_update_db_counters();
+/** Write a previously locked, immutable cache-counter snapshot */
+void sock_mgr_update_db_counters(const socket_counters_t &counters_by_socket);
+
+/** Copy cache counters for all sockets while holding the supplied writer guard */
+socket_counters_t sock_mgr_copy_cache_counters(const counter_state_write_lock &counter_lock);
 
 #endif /* SOCKET_MANAGER_H_ */

src/subdir.mk

@@ -42,6 +42,6 @@ C_DEPS += \
 src/%.o: src/%.cpp
 	@echo 'Building file: $<'
 	@echo 'Invoking: GCC C Compiler'
-	$(CC) -O3 -g3 -Wall -I/usr/include/swss -c -fmessage-length=0 -MMD -MP -MF"$(@:%.o=%.d)" -MT"$(@)" -o "$@" "$<"
+	$(CC) -std=gnu++17 -O3 -g3 -Wall -I/usr/include/swss -c -fmessage-length=0 -MMD -MP -MF"$(@:%.o=%.d)" -MT"$(@)" -o "$@" "$<"
 	@echo 'Finished building: $<'
 	@echo ' '