[dhcpmon] Add packet event quiescing

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;
+        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,22 +435,31 @@ 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;
+    {
+        counter_state_write_lock counter_lock;
+        if (!counter_lock.owns_lock()) {
             return;
         }
+        lock = std::unique_lock<std::mutex>(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");
+                return;
+            }
+        }
+        counters_by_socket = sock_mgr_copy_cache_counters();
     }
     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/event_mgr.cpp

@@ -1,4 +1,5 @@
 #include <syslog.h>
+#include <vector>
 
 #include "event_mgr.h"
 
@@ -69,25 +70,97 @@ int event_mgr::add_event(struct event* event, const struct timeval *timeout, con
 void event_mgr::del_all_events(const std::string &tag)
 {
     int count = 0;
-    for (const auto &event : this->event_map[tag]) {
+    const auto tagged_events = this->event_map.find(tag);
+    if (tagged_events == this->event_map.end()) {
+        return;
+    }
+    auto all_events = this->event_map.find("");
+    for (const auto &event : tagged_events->second) {
+        int fd = event_get_fd(event);
+        if (!tag.empty() && all_events != this->event_map.end()) {
+            all_events->second.erase(event);
+        }
         event_del(event);
         event_free(event);
         count++;
-        syslog(LOG_INFO, "event_mgr: Deleted event (fd=%d) of tag %s from %s", event_get_fd(event), tag.c_str(), this->name.c_str());
+        syslog(LOG_INFO, "event_mgr: Deleted event (fd=%d) of tag %s from %s", fd, tag.c_str(), this->name.c_str());
     }
-    if (tag != "") {
-        std::unordered_set<struct event *> &tagless_set = this->event_map[""];
-        std::unordered_set<struct event *> &tagged_set = this->event_map[tag];
-        for (const auto &event : tagged_set) {
-            tagless_set.erase(event);
-        }
+    if (!tag.empty()) {
         this->event_map.erase(tag);
     } else {
         this->event_map.clear();
     }
     syslog(LOG_INFO, "event_mgr: Deleted %d events of tag %s for %s", count, tag.c_str(), this->name.c_str());
 }
 
+int event_mgr::suspend_all_events(const std::string &tag)
+{
+    if (tag.empty()) {
+        syslog(LOG_ALERT, "event_mgr: Refusing to suspend untagged events for %s",
+               this->name.c_str());
+        return -1;
+    }
+    const auto tagged_events = this->event_map.find(tag);
+    if (tagged_events == this->event_map.end()) {
+        syslog(LOG_ALERT, "event_mgr: Cannot suspend unknown tag %s for %s",
+               tag.c_str(), this->name.c_str());
+        return -1;
+    }
+    for (const auto &event : tagged_events->second) {
+        if (event_get_fd(event) < 0) {
+            syslog(LOG_ALERT, "event_mgr: Cannot suspend non-fd event with tag %s for %s",
+                   tag.c_str(), this->name.c_str());
+            return -1;
+        }
+    }
+    std::vector<struct event *> deleted_events;
+    for (const auto &event : tagged_events->second) {
+        if (event_del(event) < 0) {
+            bool restore_failed = false;
+            for (struct event *deleted_event : deleted_events) {
+                if (event_add(deleted_event, NULL) < 0) {
+                    restore_failed = true;
+                }
+            }
+            syslog(LOG_ALERT, "event_mgr: Failed to suspend event (fd=%d) with tag %s for %s",
+                   event_get_fd(event), tag.c_str(), this->name.c_str());
+            return restore_failed ? -2 : -1;
+        }
+        deleted_events.push_back(event);
+    }
+    return 0;
+}
+
+int event_mgr::resume_all_events(const std::string &tag)
+{
+    if (tag.empty()) {
+        syslog(LOG_ALERT, "event_mgr: Refusing to resume untagged events for %s",
+               this->name.c_str());
+        return -1;
+    }
+    const auto tagged_events = this->event_map.find(tag);
+    if (tagged_events == this->event_map.end()) {
+        syslog(LOG_ALERT, "event_mgr: Cannot resume unknown tag %s for %s",
+               tag.c_str(), this->name.c_str());
+        return -1;
+    }
+    for (const auto &event : tagged_events->second) {
+        if (event_get_fd(event) < 0) {
+            syslog(LOG_ALERT, "event_mgr: Cannot resume non-fd event with tag %s for %s",
+                   tag.c_str(), this->name.c_str());
+            this->suspend_all_events(tag);
+            return -1;
+        }
+        if (event_add(event, NULL) < 0) {
+            syslog(LOG_ALERT, "event_mgr: Failed to resume event (fd=%d) with tag %s for %s",
+                   event_get_fd(event), tag.c_str(), this->name.c_str());
+            this->suspend_all_events(tag);
+            return -1;
+        }
+    }
+    return 0;
+}
+
 /**
  * @code activate_all_events(tag, res);
  * 
@@ -97,7 +170,15 @@ void event_mgr::del_all_events(const std::string &tag)
  */
 void event_mgr::activate_all_events(const std::string &tag, int res)
 {
-    for (const auto &event : this->event_map[tag]) {
+    const auto tagged_events = this->event_map.find(tag);
+    if (tagged_events == this->event_map.end()) {
+        if (!tag.empty()) {
+            syslog(LOG_WARNING, "event_mgr: Cannot activate unknown tag %s for %s",
+                   tag.c_str(), this->name.c_str());
+        }
+        return;
+    }
+    for (const auto &event : tagged_events->second) {
         event_active(event, res, 0);
         syslog(LOG_INFO, "event_mgr: Activated event (fd=%d) of tag %s from %s", event_get_fd(event), tag.c_str(), this->name.c_str());
     }

src/event_mgr.h

@@ -12,6 +12,8 @@ class event_mgr {
         int init_base();
         int add_event(struct event* event, const struct timeval *timeout, const std::string &tag="");
         void del_all_events(const std::string &tag="");
+        int suspend_all_events(const std::string &tag);
+        int resume_all_events(const std::string &tag);
         void activate_all_events(const std::string &tag="", int res=0);
         void free();
         struct event_base* get_base();

src/packet_handler.cpp

@@ -16,6 +16,8 @@
 #include "dhcp_check_profile.h"   /** to get dhcp/v6 check profile */
 #include "util.h"
 
+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 +861,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));