[dhcpmon] Protect counters from concurrent packet and health updates

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");
@@ -223,6 +228,10 @@ static void signal_callback(evutil_socket_t fd, short event, void *arg)
     if (fd == SIGUSR1) {
         // we need to sync cache counter from COUNTERS_DB
         syslog(LOG_INFO, "Received signal to stop writing to DB counter");
+        counter_state_write_lock counter_lock;
+        if (!counter_lock.owns_lock()) {
+            return;
+        }
         std::lock_guard<std::mutex> lock(db_sync_mutex);
         sock_mgr_pause_write_cache_to_db();
         syslog(LOG_INFO, "Stopped writing to DB counter");
@@ -260,6 +269,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);
 
+    counter_state_write_lock counter_lock;
+    if (!counter_lock.owns_lock()) {
+        return;
+    }
     std::lock_guard<std::mutex> lock(db_sync_mutex);
 
     // can only sync db to cache counter and db updater is paused, otherwise its unexpected
@@ -392,6 +405,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,6 +435,10 @@ 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");
+    counter_state_write_lock counter_lock;
+    if (!counter_lock.owns_lock()) {
+        return;
+    }
     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

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();
  *

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>
@@ -41,6 +44,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);
 

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 ' '