diff --git a/src/dhcp_devman.cpp b/src/dhcp_devman.cpp index 9fa951e3c..88fea920a 100644 --- a/src/dhcp_devman.cpp +++ b/src/dhcp_devman.cpp @@ -11,6 +11,8 @@ #include #include +#include + #include "dhcp_devman.h" @@ -205,11 +207,13 @@ bool dhcp_devman_is_tracked_interface(const std::string &ifname) * @param none * @return none */ -static void update_vlan_mapping() +static void update_vlan_mapping(const std::shared_ptr &config_db, + std::unordered_map &vlan_mapping, + std::unordered_map> &reverse_vlan_mapping) { syslog(LOG_INFO, "Updating vlan mapping from VLAN_MEMBER"); auto match_pattern = std::string("VLAN_MEMBER|*"); - auto keys = mConfigDbPtr->keys(match_pattern); + auto keys = config_db->keys(match_pattern); std::string all_ifname; std::string all_skipped_ifname; for (const auto &key : keys) { @@ -221,8 +225,8 @@ static void update_vlan_mapping() all_skipped_ifname += "<" + ifname + ", " + vlan + ">, "; continue; } - vlan_map[ifname] = vlan; - rev_vlan_map[vlan].insert(ifname); + vlan_mapping[ifname] = vlan; + reverse_vlan_mapping[vlan].insert(ifname); all_ifname += "<" + ifname + ", " + vlan + ">, "; } syslog(LOG_INFO, "Added vlan member interface mappings: %s", all_ifname.c_str()); @@ -236,11 +240,13 @@ static void update_vlan_mapping() * @param none * @return none */ -static void update_portchannel_mapping() +static void update_portchannel_mapping(const std::shared_ptr &config_db, + std::unordered_map &portchannel_mapping, + std::unordered_map> &reverse_portchannel_mapping) { syslog(LOG_INFO, "Updating port-channel mapping from PORTCHANNEL_MEMBER"); auto match_pattern = std::string("PORTCHANNEL_MEMBER|*"); - auto keys = mConfigDbPtr->keys(match_pattern); + auto keys = config_db->keys(match_pattern); std::string all_ifname; std::string all_skipped_ifname; for (const auto &key : keys) { @@ -252,14 +258,31 @@ static void update_portchannel_mapping() all_skipped_ifname += "<" + ifname + ", " + portchannel + ">, "; continue; } - portchan_map[ifname] = portchannel; - rev_portchan_map[portchannel].insert(ifname); + portchannel_mapping[ifname] = portchannel; + reverse_portchannel_mapping[portchannel].insert(ifname); all_ifname += "<" + ifname + ", " + portchannel + ">, "; } syslog(LOG_INFO, "Added port-channel member interface mappings: %s", all_ifname.c_str()); syslog(LOG_INFO, "Skipped port-channel member interface mappings: %s", all_skipped_ifname.c_str()); } +void dhcp_devman_refresh_mappings() +{ + std::unordered_map new_vlan_map; + std::unordered_map new_portchan_map; + std::unordered_map> new_rev_vlan_map; + std::unordered_map> new_rev_portchan_map; + auto config_db = std::make_shared("CONFIG_DB", 0); + + update_vlan_mapping(config_db, new_vlan_map, new_rev_vlan_map); + update_portchannel_mapping(config_db, new_portchan_map, new_rev_portchan_map); + + vlan_map.swap(new_vlan_map); + portchan_map.swap(new_portchan_map); + rev_vlan_map.swap(new_rev_vlan_map); + rev_portchan_map.swap(new_rev_portchan_map); +} + int dhcp_devman_init() { syslog(LOG_INFO, "Initializing dhcp device manager"); @@ -297,8 +320,12 @@ int dhcp_devman_init() agg_dev_prefix = agg_dev_all + "-"; // vlan and its members, portchannel and its members are initialized regardless of whether they are in cmdline - update_vlan_mapping(); - update_portchannel_mapping(); + try { + dhcp_devman_refresh_mappings(); + } catch (const std::exception &e) { + syslog(LOG_ALERT, "Failed to initialize DHCP interface mappings: %s", e.what()); + return -1; + } syslog(LOG_INFO, "Dhcp device manager initialized successfully"); @@ -309,6 +336,8 @@ void dhcp_devman_free() { vlan_map.clear(); portchan_map.clear(); + rev_vlan_map.clear(); + rev_portchan_map.clear(); for (const auto &[ifname, context] : intfs) { dhcp_device_free(context); } diff --git a/src/dhcp_devman.h b/src/dhcp_devman.h index 73c1cd6f2..bc2885559 100644 --- a/src/dhcp_devman.h +++ b/src/dhcp_devman.h @@ -117,6 +117,15 @@ bool dhcp_devman_is_tracked_interface(const std::string &ifname); */ int dhcp_devman_init(); +/** + * @code dhcp_devman_refresh_mappings(); + * + * @brief rebuild VLAN and PortChannel membership mappings transactionally from CONFIG_DB. + * + * @return none + */ +void dhcp_devman_refresh_mappings(); + /** * @code dhcp_devman_free(); * diff --git a/src/dhcp_mon.cpp b/src/dhcp_mon.cpp index ef4f6623d..40a0139b7 100644 --- a/src/dhcp_mon.cpp +++ b/src/dhcp_mon.cpp @@ -5,8 +5,14 @@ */ #include +#include #include +#include #include +#include +#include +#include +#include #include #include #include @@ -47,6 +53,13 @@ static const char db_update_tag[] = "DB_UPDATE"; static std::chrono::steady_clock::time_point last_update_time{}; /** Default time point to check whether a time_point has been initialized or updated yet. */ static const std::chrono::steady_clock::time_point default_time_point{}; +static std::thread::id main_thread_id; +static bool topology_refresh_pending = false; +static bool config_subscribers_failed = false; +static std::shared_ptr vlan_member_subscriber; +static std::shared_ptr portchannel_member_subscriber; +static const char config_event_tag[] = "CONFIG_UPDATE"; +static int dhcp_mon_reconcile_topology(); std::shared_ptr mConfigDbPtr = std::make_shared ("CONFIG_DB", 0); std::shared_ptr mCountersDbPtr = std::make_shared ("COUNTERS_DB", 0); @@ -55,6 +68,62 @@ std::shared_ptr mStateDbMuxTablePtr = std::make_shared mStateDbPtr.get(), "HW_MUX_CABLE_TABLE" ); +static void config_update_callback(evutil_socket_t, short, void *arg) +{ + auto *subscriber = static_cast(arg); + try { + subscriber->readData(); + std::deque entries; + subscriber->pops(entries); + if (!entries.empty()) { + topology_refresh_pending = true; + } + } catch (const std::exception &e) { + syslog(LOG_ALERT, "Failed to read DHCP membership update: %s", e.what()); + config_subscribers_failed = true; + topology_refresh_pending = true; + main_event_mgr->suspend_all_events(config_event_tag); + } +} + +static void clear_config_events() +{ + main_event_mgr->del_all_events(config_event_tag); + vlan_member_subscriber.reset(); + portchannel_member_subscriber.reset(); +} + +static int register_config_events() +{ + clear_config_events(); + try { + vlan_member_subscriber = std::make_shared( + mConfigDbPtr.get(), "VLAN_MEMBER"); + portchannel_member_subscriber = std::make_shared( + mConfigDbPtr.get(), "PORTCHANNEL_MEMBER"); + } catch (const std::exception &e) { + syslog(LOG_ALERT, "Failed to initialize DHCP membership subscribers: %s", e.what()); + return -1; + } + + for (const auto &subscriber : {vlan_member_subscriber, portchannel_member_subscriber}) { + struct event *config_event = event_new(main_event_mgr->get_base(), subscriber->getFd(), + EV_READ | EV_PERSIST, config_update_callback, + subscriber.get()); + if (config_event == NULL || + main_event_mgr->add_event(config_event, NULL, config_event_tag) < 0) { + if (config_event != NULL) { + event_free(config_event); + } + syslog(LOG_ALERT, "Failed to register DHCP membership event"); + clear_config_events(); + return -1; + } + } + config_subscribers_failed = false; + return 0; +} + /** * @code recalculate_agg_counter(all_counters); * @@ -212,9 +281,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 +297,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 lock(db_sync_mutex); sock_mgr_pause_write_cache_to_db(); syslog(LOG_INFO, "Stopped writing to DB counter"); @@ -260,6 +338,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 lock(db_sync_mutex); // can only sync db to cache counter and db updater is paused, otherwise its unexpected @@ -393,6 +475,42 @@ static void timeout_callback(evutil_socket_t fd, short event, void *arg) { syslog_debug(LOG_INFO, "Received timeout signal for DHCP relay health check"); + bool subscribers_available = true; + if (config_subscribers_failed) { + if (register_config_events() < 0) { + topology_refresh_pending = true; + subscribers_available = false; + } else { + topology_refresh_pending = true; + } + } + + if (topology_refresh_pending && subscribers_available) { + if (sock_mgr_suspend_packet_handler() < 0) { + syslog(LOG_ALERT, "Failed to suspend packet handlers for topology refresh"); + dhcp_mon_stop(); + return; + } + int result = dhcp_mon_reconcile_topology(); + if (result == 0) { + sock_mgr_drain_sock_buffer(); + } + if (sock_mgr_resume_packet_handler() < 0) { + syslog(LOG_ALERT, "Failed to resume packet handlers after topology refresh"); + dhcp_mon_stop(); + return; + } + topology_refresh_pending = result != 0; + if (result == 0) { + syslog(LOG_INFO, "Refreshed DHCP interface membership from CONFIG_DB"); + return; + } + } + + 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); dhcp_devman_print_all_status_debug(DHCP_COUNTERS_CURRENT_V6); @@ -418,22 +536,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 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; + { + counter_state_write_lock counter_lock; + if (!counter_lock.owns_lock()) { return; } + lock = std::unique_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"); + 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"); } @@ -454,50 +581,135 @@ static void free_event_mgr(struct event_mgr *mgr) } /** - * @code initialize_all_intf_counters(); - * @brief Initialize all db counters and cache counters for all tracked interfaces - * @param none - * @return 0 upon success, negative upon failure + * @code reconcile_all_intf_counters(initialize_db); + * @brief Reconcile cache counters for all tracked interfaces + * @param initialize_db initialize missing database counters when true + * @return none */ -static void initialize_all_intf_counters() +static void reconcile_all_intf_counters(bool initialize_db) { - for (const auto &[vlan, intfs] : rev_vlan_map) { - for (const auto &ifname : intfs) { + std::unordered_set valid_ifnames; + auto ensure_interface = [&valid_ifnames, initialize_db](const std::string &ifname) { + valid_ifnames.insert(ifname); + if (initialize_db && !all_counters_initialized(ifname)) { initialize_all_counters(ifname); + } else if (!initialize_db && !sock_mgr_all_cache_counters_initialized(ifname)) { + sock_mgr_init_cache_counters(ifname, DHCP_MESSAGE_TYPE_COUNT, DHCPV6_MESSAGE_TYPE_COUNT); } - initialize_all_counters(vlan); - sock_mgr_init_cache_counters(agg_dev_prefix + vlan, DHCP_MESSAGE_TYPE_COUNT, DHCPV6_MESSAGE_TYPE_COUNT); - } + }; + auto ensure_aggregate = [&valid_ifnames](const std::string &ifname) { + valid_ifnames.insert(ifname); + if (!sock_mgr_all_cache_counters_initialized(ifname)) { + sock_mgr_init_cache_counters(ifname, DHCP_MESSAGE_TYPE_COUNT, DHCPV6_MESSAGE_TYPE_COUNT); + } + }; - for (const auto &[portchan, intfs] : rev_portchan_map) { - for (const auto &ifname : intfs) { - initialize_all_counters(ifname); + for (const auto &[vlan, members] : rev_vlan_map) { + for (const auto &ifname : members) { + ensure_interface(ifname); } - initialize_all_counters(portchan); - sock_mgr_init_cache_counters(agg_dev_prefix + portchan, DHCP_MESSAGE_TYPE_COUNT, DHCPV6_MESSAGE_TYPE_COUNT); + ensure_interface(vlan); + ensure_aggregate(agg_dev_prefix + vlan); } - // Now all vlan and portchannel related interfaces have entries in counters, now do the rest (uplink) - for (const auto &itr : intfs) { - if (!all_counters_initialized(itr.first)) { - initialize_all_counters(itr.first); + for (const auto &[portchan, members] : rev_portchan_map) { + for (const auto &ifname : members) { + ensure_interface(ifname); } + ensure_interface(portchan); + ensure_aggregate(agg_dev_prefix + portchan); + } + + for (const auto &entry : intfs) { + ensure_interface(entry.first); } - // also initialize mgmt and agg device counters if (mgmt_ifname.size() > 0) { - initialize_all_counters(mgmt_ifname); + ensure_interface(mgmt_ifname); } + ensure_aggregate(agg_dev_all); - sock_mgr_init_cache_counters(agg_dev_all, DHCP_MESSAGE_TYPE_COUNT, DHCPV6_MESSAGE_TYPE_COUNT); + sock_mgr_remove_cache_counters_except(valid_ifnames); + for (int sock : {rx_sock, tx_sock, rx_sock_v6, tx_sock_v6}) { + sock_info_t &sock_info = sock_mgr_get_sock_info(sock); + recalculate_agg_counter(sock_info.all_counters); + recalculate_agg_counter(sock_info.all_counters_snapshot); + } - // counter db (the interfaces) might be outdated, clean up stale entries to be in sync with current tracked interfaces - cleanup_stale_db_counters(); + if (initialize_db) { + cleanup_stale_db_counters(); + } +} + +static int dhcp_mon_reconcile_topology() +{ + if (std::this_thread::get_id() != main_thread_id) { + syslog(LOG_ALERT, "Topology reconciliation must run on the main event-loop thread"); + return -1; + } + if (packet_handlers_enabled.load(std::memory_order_acquire)) { + syslog(LOG_ALERT, "Topology reconciliation requires suspended packet handlers"); + return -1; + } + + std::lock_guard lock(db_sync_mutex); + if (!sock_mgr_pause_write_cache_to_db_all_cleared()) { + return 1; + } + + decltype(vlan_map) old_vlan_map; + decltype(portchan_map) old_portchan_map; + decltype(rev_vlan_map) old_rev_vlan_map; + decltype(rev_portchan_map) old_rev_portchan_map; + std::unordered_map> old_counters; + try { + old_vlan_map = vlan_map; + old_portchan_map = portchan_map; + old_rev_vlan_map = rev_vlan_map; + old_rev_portchan_map = rev_portchan_map; + for (int sock : {rx_sock, tx_sock, rx_sock_v6, tx_sock_v6}) { + sock_info_t &sock_info = sock_mgr_get_sock_info(sock); + old_counters[sock] = {sock_info.all_counters, sock_info.all_counters_snapshot}; + } + } catch (const std::exception &e) { + syslog(LOG_ALERT, "Failed to snapshot DHCP topology before reconciliation: %s", e.what()); + return -1; + } + + try { + dhcp_devman_refresh_mappings(); + reconcile_all_intf_counters(false); + mCountersDbPtr = std::make_shared("COUNTERS_DB", 0); + sock_mgr_update_db_counters(); + cleanup_stale_db_counters(); + sock_mgr_update_snapshot(); + } catch (const std::exception &e) { + syslog(LOG_ALERT, "Failed to reconcile DHCP interface membership: %s", e.what()); + vlan_map = std::move(old_vlan_map); + portchan_map = std::move(old_portchan_map); + rev_vlan_map = std::move(old_rev_vlan_map); + rev_portchan_map = std::move(old_rev_portchan_map); + for (auto &[sock, counters] : old_counters) { + sock_info_t &sock_info = sock_mgr_get_sock_info(sock); + sock_info.all_counters = std::move(counters.first); + sock_info.all_counters_snapshot = std::move(counters.second); + } + try { + mCountersDbPtr = std::make_shared("COUNTERS_DB", 0); + sock_mgr_update_db_counters(); + cleanup_stale_db_counters(); + } catch (const std::exception &rollback_error) { + syslog(LOG_ALERT, "Failed to restore COUNTERS_DB after topology rollback: %s", rollback_error.what()); + } + return -1; + } + return 0; } int dhcp_mon_init(size_t snaplen, int window_sec, int max_count, int db_update_interval) { int rv = -1; + main_thread_id = std::this_thread::get_id(); syslog(LOG_INFO, "Initializing dhcp monitor with snaplen %zu, window_sec %d, max_count %d, db_update_interval %d", snaplen, window_sec, max_count, db_update_interval); @@ -519,7 +731,7 @@ int dhcp_mon_init(size_t snaplen, int window_sec, int max_count, int db_update_i // deinitialization of counters is not our responsibility // cache counter will be cleanup by sock_mgr_free and the initialized db we intend to keep - initialize_all_intf_counters(); + reconcile_all_intf_counters(true); syslog(LOG_INFO, "Initialized all counters for tracked interfaces"); window_interval_sec = window_sec; @@ -667,6 +879,10 @@ static int register_main_events() break; } + if (register_config_events() < 0) { + break; + } + rv = 0; syslog(LOG_INFO, "Main events registered successfully"); @@ -693,6 +909,8 @@ static int register_main_events() int dhcp_mon_start() { int rv = -1; + int reconcile_result = -1; + int resume_result = -1; syslog(LOG_INFO, "Starting dhcp monitor in %s", debug_on ? "debug mode" : "normal mode"); @@ -711,6 +929,17 @@ int dhcp_mon_start() goto unregister_cache_counter_updater; } + topology_refresh_pending = true; + if (sock_mgr_suspend_packet_handler() < 0) { + goto unregister_main_events; + } + reconcile_result = dhcp_mon_reconcile_topology(); + resume_result = sock_mgr_resume_packet_handler(); + if (reconcile_result != 0 || resume_result < 0) { + goto unregister_main_events; + } + topology_refresh_pending = false; + sock_mgr_drain_sock_buffer(); // it could fail and we wouldnt know it because its in another thread diff --git a/src/event_mgr.cpp b/src/event_mgr.cpp index 6e1e0e76c..a794cee1a 100644 --- a/src/event_mgr.cpp +++ b/src/event_mgr.cpp @@ -69,18 +69,22 @@ 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 &tagless_set = this->event_map[""]; - std::unordered_set &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(); @@ -88,6 +92,54 @@ void event_mgr::del_all_events(const std::string &tag) syslog(LOG_INFO, "event_mgr: Deleted %d events of tag %s for %s", count, tag.c_str(), this->name.c_str()); } +void 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; + } + 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; + } + for (const auto &event : tagged_events->second) { + event_del(event); + } +} + +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 +149,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()); } diff --git a/src/event_mgr.h b/src/event_mgr.h index 90ff4a146..26a1b5135 100644 --- a/src/event_mgr.h +++ b/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=""); + void 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(); diff --git a/src/packet_handler.cpp b/src/packet_handler.cpp index 7ec5d07a3..aa564c1a0 100644 --- a/src/packet_handler.cpp +++ b/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)); diff --git a/src/sock_mgr.cpp b/src/sock_mgr.cpp index 8d3e48d81..53f54b019 100644 --- a/src/sock_mgr.cpp +++ b/src/sock_mgr.cpp @@ -11,6 +11,9 @@ #include #include #include +#include +#include +#include #include #include "sock_mgr.h" @@ -36,14 +39,131 @@ static const char dhcpv6_outbound_filter[] = "outbound and ip6 and udp and (port /** Tags for different events, so we can triiger only one type */ static const char packet_handler_tag[] = "PacketHandler"; static const char cache_counter_updater_tag[] = "CacheCounterUpdater"; +static const char keepalive_tag[] = "Keepalive"; + +static void keepalive_callback(evutil_socket_t, short, void *) +{ +} /* sock fd to sock_info mapping */ std::unordered_map sock_map; +std::shared_mutex packet_handler_quiesce_mutex; +std::atomic packet_handlers_enabled{true}; +std::atomic counter_state_writers_pending{0}; +static std::unique_lock packet_handler_quiesce_lock; +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; +static void set_packet_handlers_enabled(bool enabled) +{ + { + std::lock_guard wait_lock(counter_state_wait_mutex); + packet_handlers_enabled.store(enabled, std::memory_order_release); + } + counter_state_wait_cv.notify_all(); +} + +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); + } + try { + lock = std::unique_lock(packet_handler_quiesce_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(); + } + syslog(LOG_ALERT, "Failed to lock DHCP counter state: %s", e.what()); + } +} + +counter_state_write_lock::~counter_state_write_lock() +{ + if (!lock.owns_lock()) { + return; + } + lock.unlock(); + 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(); + } +} + +bool counter_state_write_lock::owns_lock() const +{ + return lock.owns_lock(); +} + +counter_state_read_lock::counter_state_read_lock() +{ + if (packet_handlers_enabled.load(std::memory_order_acquire) && + counter_state_writers_pending.load(std::memory_order_acquire) == 0) { + try { + lock = std::shared_lock(packet_handler_quiesce_mutex, + std::try_to_lock); + } catch (const std::system_error &e) { + syslog(LOG_ALERT, "Failed to lock DHCP counter state for packet handling: %s", e.what()); + return; + } + if (lock.owns_lock() && + packet_handlers_enabled.load(std::memory_order_acquire) && + counter_state_writers_pending.load(std::memory_order_acquire) == 0) { + return; + } + if (lock.owns_lock()) { + lock.unlock(); + } + } + + while (packet_handlers_enabled.load(std::memory_order_acquire)) { + { + std::unique_lock wait_lock(counter_state_wait_mutex); + counter_state_wait_cv.wait(wait_lock, [] { + return !packet_handlers_enabled.load(std::memory_order_acquire) || + counter_state_writers_pending.load(std::memory_order_acquire) == 0; + }); + } + if (!packet_handlers_enabled.load(std::memory_order_acquire)) { + return; + } + try { + lock = std::shared_lock(packet_handler_quiesce_mutex); + } catch (const std::system_error &e) { + syslog(LOG_ALERT, "Failed to lock DHCP counter state for packet handling: %s", e.what()); + return; + } + if (!packet_handlers_enabled.load(std::memory_order_acquire)) { + lock.unlock(); + return; + } + if (counter_state_writers_pending.load(std::memory_order_acquire) == 0) { + return; + } + lock.unlock(); + } +} + +bool counter_state_read_lock::owns_lock() const +{ + return lock.owns_lock(); +} + /** * @code opensocket(); * @@ -385,6 +505,18 @@ int sock_mgr_init_event_mgr() sock_mgr_free_event_mgr(); return -1; } + struct event *keepalive_event = event_new(info.event_mgr_ptr->get_base(), -1, EV_PERSIST, + keepalive_callback, NULL); + struct timeval keepalive_interval = {.tv_sec = 3600, .tv_usec = 0}; + if (keepalive_event == NULL || + info.event_mgr_ptr->add_event(keepalive_event, &keepalive_interval, keepalive_tag) < 0) { + if (keepalive_event != NULL) { + event_free(keepalive_event); + } + syslog(LOG_ALERT, "Failed to initialize event manager keepalive %s", info.name); + sock_mgr_free_event_mgr(); + return -1; + } } return 0; @@ -432,6 +564,65 @@ void sock_mgr_unregister_packet_handler() } } +int sock_mgr_suspend_packet_handler() +{ + if (packet_handler_quiesce_lock.owns_lock()) { + syslog(LOG_ALERT, "Packet handlers are already suspended"); + return -1; + } + for (const auto &entry : sock_map) { + entry.second.event_mgr_ptr->suspend_all_events(packet_handler_tag); + } + set_packet_handlers_enabled(false); + try { + packet_handler_quiesce_lock = std::unique_lock(packet_handler_quiesce_mutex); + } catch (const std::system_error &e) { + syslog(LOG_ALERT, "Failed to quiesce packet handlers: %s", e.what()); + set_packet_handlers_enabled(true); + int restore_result = 0; + for (const auto &entry : sock_map) { + if (entry.second.event_mgr_ptr->resume_all_events(packet_handler_tag) < 0) { + restore_result = -1; + } + } + if (restore_result < 0) { + set_packet_handlers_enabled(false); + for (const auto &entry : sock_map) { + entry.second.event_mgr_ptr->suspend_all_events(packet_handler_tag); + } + syslog(LOG_ALERT, "Failed to restore packet handlers after quiesce failure"); + } + return -1; + } + return 0; +} + +int sock_mgr_resume_packet_handler() +{ + if (!packet_handler_quiesce_lock.owns_lock()) { + syslog(LOG_ALERT, "Packet handlers are not suspended"); + return -1; + } + set_packet_handlers_enabled(true); + packet_handler_quiesce_lock.unlock(); + + for (const auto &entry : sock_map) { + if (entry.second.event_mgr_ptr->resume_all_events(packet_handler_tag) < 0) { + set_packet_handlers_enabled(false); + for (const auto &suspended_entry : sock_map) { + suspended_entry.second.event_mgr_ptr->suspend_all_events(packet_handler_tag); + } + try { + packet_handler_quiesce_lock = std::unique_lock(packet_handler_quiesce_mutex); + } catch (const std::system_error &e) { + syslog(LOG_ALERT, "Failed to restore packet quiesce lock after resume failure: %s", e.what()); + } + return -1; + } + } + return 0; +} + int sock_mgr_register_cache_counter_updater(event_callback_fn callback) { syslog(LOG_INFO, "Registering cache counter updater for all sockets"); @@ -605,7 +796,8 @@ void sock_mgr_init_cache_counters(const std::string &ifname, uint8_t dhcp_messag bool sock_mgr_all_cache_counters_initialized(const std::string &ifname) { - for (const auto &[sock, info] : sock_map) { + for (const auto &entry : sock_map) { + const auto &info = entry.second; auto itr = info.all_counters.find(ifname); if (itr == info.all_counters.end()) { return false; @@ -614,17 +806,53 @@ bool sock_mgr_all_cache_counters_initialized(const std::string &ifname) return true; } -void sock_mgr_update_db_counters() +void sock_mgr_remove_cache_counters_except(const std::unordered_set &valid_ifnames) { - syslog_debug(LOG_INFO, "Updating all cache counters to DB counters"); + for (auto &entry : sock_map) { + auto &info = entry.second; + for (auto itr = info.all_counters.begin(); itr != info.all_counters.end();) { + if (valid_ifnames.find(itr->first) == valid_ifnames.end()) { + itr = info.all_counters.erase(itr); + } else { + itr++; + } + } + for (auto itr = info.all_counters_snapshot.begin(); itr != info.all_counters_snapshot.end();) { + if (valid_ifnames.find(itr->first) == valid_ifnames.end()) { + itr = info.all_counters_snapshot.erase(itr); + } else { + itr++; + } + } + } +} +socket_counters_t sock_mgr_copy_cache_counters() +{ + socket_counters_t counters_by_socket; 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, 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; @@ -639,4 +867,9 @@ void sock_mgr_update_db_counters() syslog_debug(LOG_INFO, "Skipped aggregated device counter entry of %sfor downstream vlan %s", all_skipped_ifname.c_str(), downstream_ifname.c_str()); } +} + +void sock_mgr_update_db_counters() +{ + sock_mgr_update_db_counters(sock_mgr_copy_cache_counters()); } \ No newline at end of file diff --git a/src/sock_mgr.h b/src/sock_mgr.h index 9619a9526..61908c415 100644 --- a/src/sock_mgr.h +++ b/src/sock_mgr.h @@ -9,9 +9,13 @@ #ifndef SOCKET_MANAGER_H_ #define SOCKET_MANAGER_H_ +#include +#include #include +#include #include #include +#include #include #include @@ -19,6 +23,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 +46,34 @@ 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; +/** Guards in-flight packet callbacks while topology and counters are reconciled */ +extern std::shared_mutex packet_handler_quiesce_mutex; +extern std::atomic packet_handlers_enabled; +extern std::atomic counter_state_writers_pending; + +class counter_state_write_lock +{ + public: + counter_state_write_lock(); + ~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; +}; + +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); @@ -59,6 +92,12 @@ int sock_mgr_register_packet_handler(); /** Unregister packet handler for socket manager */ void sock_mgr_unregister_packet_handler(); +/** Temporarily suspend registered packet handlers */ +int sock_mgr_suspend_packet_handler(); + +/** Resume registered packet handlers */ +int sock_mgr_resume_packet_handler(); + /** Register cache counter updater callback for socket manager */ int sock_mgr_register_cache_counter_updater(event_callback_fn callback); @@ -104,7 +143,14 @@ 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); +/** Remove cache counters that are not present in the valid interface set */ +void sock_mgr_remove_cache_counters_except(const std::unordered_set &valid_ifnames); + /** Update database counters from cache counters for all sockets */ void sock_mgr_update_db_counters(); +void sock_mgr_update_db_counters(const socket_counters_t &counters_by_socket); + +/** Copy cache counters for all sockets */ +socket_counters_t sock_mgr_copy_cache_counters(); #endif /* SOCKET_MANAGER_H_ */