diff --git a/src/dhcp_mon.cpp b/src/dhcp_mon.cpp index ef4f6623d..c3aa1a82f 100644 --- a/src/dhcp_mon.cpp +++ b/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 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 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 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(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 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(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"); } diff --git a/src/packet_handler.cpp b/src/packet_handler.cpp index 7ec5d07a3..10409ca0b 100644 --- a/src/packet_handler.cpp +++ b/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)); diff --git a/src/sock_mgr.cpp b/src/sock_mgr.cpp index 8d3e48d81..845e0ade2 100644 --- a/src/sock_mgr.cpp +++ b/src/sock_mgr.cpp @@ -11,7 +11,15 @@ #include #include #include +#include +#include +#include #include +#include + +#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 sock_map; +static std::shared_timed_mutex counter_state_mutex; +static std::atomic counter_state_writers_pending{0}; +static std::mutex counter_state_wait_mutex; +static std::condition_variable counter_state_wait_cv; + extern std::shared_ptr mCountersDbPtr; extern std::string downstream_ifname; +counter_state_write_lock::counter_state_write_lock() +{ + { + std::lock_guard 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(counter_state_mutex); + } catch (const std::system_error &e) { + bool notify = false; + { + std::lock_guard 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(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 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 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 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 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()); } -} \ No newline at end of file +} diff --git a/src/sock_mgr.h b/src/sock_mgr.h index 9619a9526..85838acaa 100644 --- a/src/sock_mgr.h +++ b/src/sock_mgr.h @@ -9,7 +9,10 @@ #ifndef SOCKET_MANAGER_H_ #define SOCKET_MANAGER_H_ +#include +#include #include +#include #include #include #include @@ -19,6 +22,7 @@ typedef std::unordered_map counter_t; typedef std::unordered_map all_counters_t; +typedef std::unordered_map 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 lock; + bool registered_writer = false; +}; + +class counter_state_read_lock +{ + public: + counter_state_read_lock(); + bool owns_lock() const; + + private: + std::shared_lock 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_ */ diff --git a/src/subdir.mk b/src/subdir.mk index 4708a6fa5..e44d3f471 100644 --- a/src/subdir.mk +++ b/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 ' '