From 70e7844804af51bcb0f1cb7ef856ba62411d09e6 Mon Sep 17 00:00:00 2001 From: mssonicbld <79238446+mssonicbld@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:15:21 +1000 Subject: [PATCH 1/2] [cmis] Fix issue: Potential race condition between split ports while doing DPSM (#863) ## Why I did it Fix issue: https://github.com/sonic-net/sonic-buildimage/issues/27373 CMIS datapath deinitialization should follow the expected software deinit sequence. When forcing a datapath re-init because `host_tx_ready` is false or the port is administratively down, xcvrd should request datapath deinit before disabling media-side Tx. This keeps the forced re-init path consistent with the normal `CMIS_STATE_DP_DEINIT` flow and CMIS datapath state machine behavior. ## How I did it Updated the forced re-init path in `cmis_manager_task.py` to call `set_datapath_deinit()` before `tx_disable_channel()`. Updated the existing `test_CmisManagerTask_task_worker_host_tx_ready_false_to_true` unit test to record the mock API call order and verify that datapath deinit is issued before Tx disable. ## How to verify it Run the xcvrd unit test: ```bash python3 -m pytest sonic-xcvrd/tests/test_xcvrd.py -k test_CmisManagerTask_task_worker_host_tx_ready_false_to_true Signed-off-by: Sonic Build Admin --- sonic-xcvrd/tests/test_xcvrd.py | 11 +++++++++-- sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py | 2 ++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/sonic-xcvrd/tests/test_xcvrd.py b/sonic-xcvrd/tests/test_xcvrd.py index e3e3cd8..e5da197 100644 --- a/sonic-xcvrd/tests/test_xcvrd.py +++ b/sonic-xcvrd/tests/test_xcvrd.py @@ -4623,9 +4623,14 @@ def test_CmisManagerTask_task_worker_fastboot(self, mock_chassis, mock_get_statu def test_CmisManagerTask_task_worker_host_tx_ready_false_to_true(self, mock_chassis, mock_get_status_sw_tbl): mock_get_status_sw_tbl = Table("STATE_DB", TRANSCEIVER_STATUS_TABLE) mock_xcvr_api = MagicMock() - mock_xcvr_api.set_datapath_deinit = MagicMock(return_value=True) + dp_deinit_tx_disable_calls = [] + mock_xcvr_api.set_datapath_deinit = MagicMock( + side_effect=lambda *args, **kwargs: dp_deinit_tx_disable_calls.append('set_datapath_deinit') or True + ) mock_xcvr_api.set_datapath_init = MagicMock(return_value=True) - mock_xcvr_api.tx_disable_channel = MagicMock(return_value=True) + mock_xcvr_api.tx_disable_channel = MagicMock( + side_effect=lambda *args, **kwargs: dp_deinit_tx_disable_calls.append('tx_disable_channel') or True + ) mock_xcvr_api.set_lpmode = MagicMock(return_value=True) mock_xcvr_api.set_application = MagicMock(return_value=True) mock_xcvr_api.is_flat_memory = MagicMock(return_value=False) @@ -4789,7 +4794,9 @@ def test_CmisManagerTask_task_worker_host_tx_ready_false_to_true(self, mock_chas task.task_worker() assert task.post_port_active_apsel_to_db.call_count == 1 + assert mock_xcvr_api.set_datapath_deinit.call_count == 1 assert mock_xcvr_api.tx_disable_channel.call_count == 1 + assert dp_deinit_tx_disable_calls == ['set_datapath_deinit', 'tx_disable_channel'] assert common.get_cmis_state_from_state_db('Ethernet0', task.xcvr_table_helper.get_status_sw_tbl(task.get_asic_id('Ethernet0'))) == CMIS_STATE_READY assert task.port_dict['Ethernet0']['forced_tx_disabled'] == True diff --git a/sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py b/sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py index 148bf65..821aee6 100644 --- a/sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py +++ b/sonic-xcvrd/xcvrd/cmis/cmis_manager_task.py @@ -928,6 +928,8 @@ def handle_cmis_inserted_state(self, lport): if is_fast_reboot and self.check_datapath_state(api, host_lanes_mask, ['DataPathActivated']): self.log_notice("{} Skip datapath re-init in fast-reboot".format(lport)) else: + self.log_notice("{} DEINIT datapath".format(lport)) + api.set_datapath_deinit(host_lanes_mask) self.log_notice("{} Forcing Tx laser OFF".format(lport)) # Force DataPath re-init api.tx_disable_channel(media_lanes_mask, True) From 30f2f0b2bd7519eb7e4455ae634caedf47f5cda1 Mon Sep 17 00:00:00 2001 From: mssonicbld <79238446+mssonicbld@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:15:27 +1000 Subject: [PATCH 2/2] Adjust select timeouts during port update handling to allow for faster transceiver DOM polling (#864) #### Description This PR optimizes the DOM (Digital Optical Monitoring) polling loop in xcvrd by improving port update event handling and reducing unnecessary wait times. The changes include: 1. **Refactored port update handling** - Extracted port update event processing into a dedicated `check_port_update()` method for better code organization and reusability 2. **Optimized timeout strategy** - Introduced two-tier timeout mechanism: - 1000ms timeout when waiting for port updates before DOM polling begins and after when it completes until the next polling cycle - 100ms timeout during DOM polling to minimize delays while still handling port events 3. **Improved loop structure** - Restructured the main loop to handle port updates before entering the DOM polling phase, preventing unnecessary blocking 4. **Added comprehensive unit tests** - Created tests covering various scenarios including multiple ports, timing edge cases, and stop event handling fixes #759 and builds on https://github.com/sonic-net/sonic-platform-daemons/pull/757 #### Motivation and Context Before the change, the DOM monitoring loop would wait up to 1 second (`SELECT_TIMEOUT_MSECS`) for port update events during each iteration of the physical port loop. This caused significant delays in DOM data collection, especially on systems with many ports. **Problem:** With the 1-second timeout being called for every physical port, the DOM polling could take an excessive amount of time to complete, delaying DOM polling updates. **Solution:** By separating port update handling from DOM polling and using a shorter 100ms timeout during the polling phase, the loop can complete much faster while still being responsive to port change events. The 1-second timeout is only used when explicitly waiting for port updates before starting the next DOM polling cycle. #### How Has This Been Tested? 1. **Unit Tests Added** - Comprehensive test coverage for the new `check_port_update()` method including: - Scenario with no link change affected ports - Link change affected port with past timestamp (should trigger immediate update) - Link change affected port with future timestamp (should defer update) - Multiple ports with mixed ready/not-ready states - Stop event handling during processing 2. **CPU usage profiling**: Measured for 10 minutes after restarting `xcvrd` on a switch fully populated with optical transceivers. - Before the change (includes changes from https://github.com/sonic-net/sonic-platform-daemons/pull/757) image - After the change image The CPU usage is slightly higher during active polling as it spends less time waiting (100ms) between interfaces and then the loop spends time waiting for port change updates in 1s chunks. 3. **Measuring actual DOM update times** - Before change: We are not able to poll every 60 seconds (takes 90 sec+) on a switch fully populated with optical transceivers ``` $ while true; do sonic-db-dump -n STATE_DB -y -k "TRANSCEIVER_DOM_SENSOR|Ethernet112" | grep last_update_time; sleep 10; done | uniq "last_update_time": "Tue Feb 24 19:25:57 2026", "last_update_time": "Tue Feb 24 19:27:32 2026", "last_update_time": "Tue Feb 24 19:29:06 2026", "last_update_time": "Tue Feb 24 19:30:40 2026", "last_update_time": "Tue Feb 24 19:32:13 2026", "last_update_time": "Tue Feb 24 19:33:47 2026", ``` - After change: Updates happen every 60 seconds for a specified interface ``` $ while true; do sonic-db-dump -n STATE_DB -y -k "TRANSCEIVER_DOM_SENSOR|Ethernet112" | grep last_update_time; sleep 10; done | uniq "last_update_time": "Tue Feb 24 19:17:09 2026", "last_update_time": "Tue Feb 24 19:18:10 2026", "last_update_time": "Tue Feb 24 19:19:10 2026", "last_update_time": "Tue Feb 24 19:20:10 2026", "last_update_time": "Tue Feb 24 19:21:11 2026", "last_update_time": "Tue Feb 24 19:22:11 2026", ``` #### Additional Information (Optional) **Key Technical Changes:** - New constants: `PORT_UPDATE_EVENT_SELECT_TIMEOUT_MSECS` (1000ms) and `PORT_UPDATE_EVENT_SELECT_TIMEOUT_FAST_MSECS` (100ms) - Modified `PortChangeObserver.handle_port_update_event()` to accept a configurable timeout parameter - The periodic update interval calculation now uses `dom_loop_start_time` to maintain consistent intervals regardless of loop execution time **Backward Compatibility:** This change is fully backward compatible and does not affect the external API or configuration. #### Tested branch - [x] master - [x] 202605: already in Nexthop's internal 202605 branch Signed-off-by: Sonic Build Admin --- sonic-xcvrd/tests/test_xcvrd.py | 291 +++++++++++++++++- sonic-xcvrd/xcvrd/dom/dom_mgr.py | 70 +++-- .../xcvrd_utilities/port_event_helper.py | 4 +- 3 files changed, 326 insertions(+), 39 deletions(-) diff --git a/sonic-xcvrd/tests/test_xcvrd.py b/sonic-xcvrd/tests/test_xcvrd.py index e5da197..9b348f5 100644 --- a/sonic-xcvrd/tests/test_xcvrd.py +++ b/sonic-xcvrd/tests/test_xcvrd.py @@ -5176,7 +5176,7 @@ def test_DomInfoUpdateTask_task_worker(self, mock_post_pm_info, mock_cmis_manager = MagicMock() task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, False, False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, False, False, False, False, False, True]) task.get_dom_polling_from_config_db = MagicMock(return_value='enabled') task.is_port_in_cmis_terminal_state = MagicMock(return_value=False) mock_detect_error.return_value = True @@ -5204,7 +5204,7 @@ def test_DomInfoUpdateTask_task_worker(self, mock_post_pm_info, assert mock_post_pm_info.call_count == 0 mock_detect_error.return_value = False mock_select.return_value = (swsscommon.Select.TIMEOUT, None) - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, False, False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, False, False, False, False, False, True]) task.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) task.get_dom_polling_from_config_db = MagicMock(side_effect=('disabled', 'enabled')) @@ -5234,7 +5234,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_failure(self, mock_post_pm_info): mock_cmis_manager = MagicMock() task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task.port_mapping.logical_port_list = ['Ethernet0'] task.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) @@ -5273,7 +5273,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_failure(self, mock_post_pm_info): # Test the case where the VDM stats are successfully frozen but the VDM stats are not successfully unfrozen task.vdm_utils._freeze_vdm_stats_and_confirm.return_value = True task.vdm_utils._unfreeze_vdm_stats_and_confirm.return_value = False - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task.task_worker() assert task.vdm_utils._freeze_vdm_stats_and_confirm.call_count == 1 assert task.vdm_utils._unfreeze_vdm_stats_and_confirm.call_count == 1 @@ -5292,7 +5292,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_failure(self, mock_post_pm_info): # Step (c) COR flags still run (no continue), and PM already ran in step (a). task.vdm_utils._unfreeze_vdm_stats_and_confirm.return_value = True task.vdm_db_utils.post_port_vdm_real_values_from_dict_to_db.side_effect = TypeError - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task.task_worker() assert task.vdm_utils._freeze_vdm_stats_and_confirm.call_count == 1 assert task.vdm_utils._unfreeze_vdm_stats_and_confirm.call_count == 1 @@ -5300,6 +5300,74 @@ def test_DomInfoUpdateTask_task_worker_vdm_failure(self, mock_post_pm_info): assert task.vdm_db_utils.post_port_vdm_flags_to_db.call_count == 1 assert mock_post_pm_info.call_count == 1 + @patch('xcvrd.dom.dom_mgr.XcvrTableHelper', MagicMock()) + @patch('xcvrd.xcvrd_utilities.common._wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.sfp_status_helper.detect_port_in_error_status', MagicMock(return_value=False)) + @patch('xcvrd.dom.dom_mgr.DomInfoUpdateTask.post_port_sfp_firmware_info_to_db', MagicMock(return_value=True)) + @patch('swsscommon.swsscommon.Select.addSelectable', MagicMock()) + @patch('xcvrd.xcvrd_utilities.port_event_helper.subscribe_port_config_change', MagicMock(return_value=(None, None))) + @patch('xcvrd.xcvrd_utilities.port_event_helper.handle_port_config_change', MagicMock()) + @patch('xcvrd.dom.dom_mgr.DomInfoUpdateTask.post_port_pm_info_to_db') + def test_DomInfoUpdateTask_task_worker_stop_event_during_port_update_wait(self, mock_post_pm_info): + """ + This test simulates the scenario where task_stopping_event is set + while waiting for periodic DB update (during check_port_update loop) + """ + port_mapping = PortMapping() + mock_sfp_obj_dict = MagicMock() + stop_event = threading.Event() + mock_cmis_manager = MagicMock() + task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager) + task.xcvr_table_helper = MagicMock() + + # Set dom_update_interval to a large value so periodic update is not triggered + # This ensures we stay in the inner while loop + task.dom_update_interval = 1000 + + # Mock check_port_update to track if it's called + check_port_update_call_count = [0] + original_check_port_update = task.check_port_update + + def mock_check_port_update(port_change_observer, timeout): + check_port_update_call_count[0] += 1 + # Don't actually call the original to avoid complexity + pass + + task.check_port_update = mock_check_port_update + + # Mock log_notice to verify the specific log message is generated + log_messages = [] + original_log_notice = task.log_notice + + def mock_log_notice(message): + log_messages.append(message) + original_log_notice(message) + + task.log_notice = mock_log_notice + + # Set up task_stopping_event to be set after check_port_update is called once + # First False: outer loop check + # Second False: inner loop check for stopping_event_set + # Third True: inner loop check after check_port_update (breaks inner loop) + # Fourth True: post-inner-loop outer check (breaks outer loop) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True, True]) + + task.port_mapping.logical_port_list = ['Ethernet0'] + task.port_mapping.physical_to_logical = {'1': ['Ethernet0']} + task.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) + task.get_dom_polling_from_config_db = MagicMock(return_value='enabled') + task.is_port_in_cmis_terminal_state = MagicMock(return_value=False) + + # Run task_worker + task.task_worker() + + # Verify that check_port_update was called + assert check_port_update_call_count[0] >= 1, "check_port_update should have been called at least once" + + # Verify that the specific log message was generated + assert any("Stop event generated during DOM monitoring loop while checking port update" in msg for msg in log_messages), \ + "Expected log message about stop event during port update check was not found" + @patch('xcvrd.xcvrd.XcvrTableHelper', MagicMock()) @patch('xcvrd.xcvrd_utilities.common._wrapper_get_presence', MagicMock(return_value=True)) @patch('xcvrd.xcvrd_utilities.sfp_status_helper.detect_port_in_error_status', MagicMock(return_value=False)) @@ -5320,7 +5388,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_freeze_conditions(self, mock_post_pm_ # Expected: Skip freeze, only basic + flags, no PM task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task.port_mapping.logical_port_list = ['Ethernet0'] task.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) @@ -5347,7 +5415,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_freeze_conditions(self, mock_post_pm_ mock_post_pm_info.reset_mock() task2 = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task2.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task2.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task2.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task2.port_mapping.logical_port_list = ['Ethernet0'] task2.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task2.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) @@ -5374,7 +5442,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_freeze_conditions(self, mock_post_pm_ mock_post_pm_info.reset_mock() task3 = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task3.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task3.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task3.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task3.port_mapping.logical_port_list = ['Ethernet0'] task3.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task3.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) @@ -5400,7 +5468,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_freeze_conditions(self, mock_post_pm_ # Expected: Freeze happens, both basic and statistic values are captured, and PM info is captured task4 = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager, 0) task4.xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE) - task4.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True]) + task4.task_stopping_event.is_set = MagicMock(side_effect=[False, False, False, True]) task4.port_mapping.logical_port_list = ['Ethernet0'] task4.port_mapping.physical_to_logical = {'1': ['Ethernet0']} task4.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) @@ -6617,6 +6685,211 @@ def test_sfp_removal_from_dict(self, mock_platform_chassis, mock_update_status, assert mock_del_dom.call_count == 1 mock_sfp.remove_xcvr_api.assert_called_once() + @patch('xcvrd.dom.dom_mgr.XcvrTableHelper', MagicMock()) + def test_DomInfoUpdateTask_check_port_update(self): + """Test the check_port_update method with various scenarios""" + port_mapping = PortMapping() + mock_sfp_obj_dict = MagicMock() + stop_event = threading.Event() + mock_cmis_manager = MagicMock() + task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager) + task.xcvr_table_helper = MagicMock() + + # Create a mock port_change_observer + mock_port_change_observer = MagicMock() + mock_port_change_observer.handle_port_update_event = MagicMock() + + # Mock update_port_db_diagnostics_on_link_change + task.update_port_db_diagnostics_on_link_change = MagicMock() + + # Test 1: No link change affected ports + task.link_change_affected_ports = {} + task.check_port_update(mock_port_change_observer, 1000) + mock_port_change_observer.handle_port_update_event.assert_called_once_with(1000) + assert task.update_port_db_diagnostics_on_link_change.call_count == 0 + + # Test 2: Link change affected port with time in the past (should trigger update) + mock_port_change_observer.handle_port_update_event.reset_mock() + task.update_port_db_diagnostics_on_link_change.reset_mock() + past_time = datetime.datetime.now() - datetime.timedelta(seconds=5) + task.link_change_affected_ports = {0: past_time} + task.check_port_update(mock_port_change_observer, 100) + mock_port_change_observer.handle_port_update_event.assert_called_once_with(100) + task.update_port_db_diagnostics_on_link_change.assert_called_once_with(0) + assert 0 not in task.link_change_affected_ports + + # Test 3: Link change affected port with time in the future (should not trigger update) + mock_port_change_observer.handle_port_update_event.reset_mock() + task.update_port_db_diagnostics_on_link_change.reset_mock() + future_time = datetime.datetime.now() + datetime.timedelta(seconds=5) + task.link_change_affected_ports = {4: future_time} + task.check_port_update(mock_port_change_observer, 1000) + mock_port_change_observer.handle_port_update_event.assert_called_once_with(1000) + assert task.update_port_db_diagnostics_on_link_change.call_count == 0 + assert 4 in task.link_change_affected_ports + + # Test 4: Multiple link change affected ports, some ready, some not + mock_port_change_observer.handle_port_update_event.reset_mock() + task.update_port_db_diagnostics_on_link_change.reset_mock() + past_time1 = datetime.datetime.now() - datetime.timedelta(seconds=2) + past_time2 = datetime.datetime.now() - datetime.timedelta(seconds=1) + future_time = datetime.datetime.now() + datetime.timedelta(seconds=5) + task.link_change_affected_ports = { + 0: past_time1, + 8: past_time2, + 12: future_time + } + task.check_port_update(mock_port_change_observer, 1000) + mock_port_change_observer.handle_port_update_event.assert_called_once_with(1000) + assert task.update_port_db_diagnostics_on_link_change.call_count == 2 + # Check that the two past ports were processed + calls = [call[0][0] for call in task.update_port_db_diagnostics_on_link_change.call_args_list] + assert 0 in calls + assert 8 in calls + # Future port should still be in the dict + assert 12 in task.link_change_affected_ports + assert 0 not in task.link_change_affected_ports + assert 8 not in task.link_change_affected_ports + + # Test 5: Stop event is set during processing + mock_port_change_observer.handle_port_update_event.reset_mock() + task.update_port_db_diagnostics_on_link_change.reset_mock() + task.task_stopping_event.set() + past_time = datetime.datetime.now() - datetime.timedelta(seconds=1) + task.link_change_affected_ports = {16: past_time} + task.check_port_update(mock_port_change_observer, 1000) + mock_port_change_observer.handle_port_update_event.assert_called_once_with(1000) + # Should break early and not process the port + assert task.update_port_db_diagnostics_on_link_change.call_count == 0 + assert 16 in task.link_change_affected_ports + + @patch('xcvrd.dom.dom_mgr.XcvrTableHelper', MagicMock()) + @patch('xcvrd.xcvrd_utilities.common._wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.sfp_status_helper.detect_port_in_error_status', MagicMock(return_value=False)) + @patch('xcvrd.dom.dom_mgr.DomInfoUpdateTask.post_port_sfp_firmware_info_to_db', MagicMock(return_value=True)) + @patch('swsscommon.swsscommon.Select.addSelectable', MagicMock()) + @patch('xcvrd.xcvrd_utilities.port_event_helper.PortChangeObserver') + @patch('xcvrd.xcvrd_utilities.port_event_helper.subscribe_port_config_change', MagicMock(return_value=(None, None))) + @patch('xcvrd.xcvrd_utilities.port_event_helper.handle_port_config_change', MagicMock()) + @patch('xcvrd.dom.dom_mgr.DomInfoUpdateTask.post_port_pm_info_to_db') + def test_DomInfoUpdateTask_scheduling_uses_loop_start_time(self, mock_post_pm_info, mock_observer_class): + """ + Test that the scheduling logic uses the loop-start timestamp instead of loop-end timestamp. + This verifies that even if per-iteration processing takes a long time, the next update time + is based on when the loop started, not when it ended, preventing timing drift. + + The test simulates: + - Iteration 1: starts at T=0, processing takes 5 seconds (ends at T=5) + - If correct: next update scheduled at T=0+60=60 + - If wrong (using loop end): next update scheduled at T=5+60=65 + - Iteration 2: starts at T=61 + - If correct: update triggers (61 >= 60), processes, schedules next at T=61+60=121 + - If wrong: update doesn't trigger yet (61 < 65), which we can detect + """ + port_mapping = PortMapping() + mock_sfp_obj_dict = MagicMock() + stop_event = threading.Event() + mock_cmis_manager = MagicMock() + task = DomInfoUpdateTask(DEFAULT_NAMESPACE, port_mapping, mock_sfp_obj_dict, stop_event, mock_cmis_manager) + task.xcvr_table_helper = MagicMock() + + # Set a non-zero dom_update_interval to test the scheduling logic + task.dom_update_interval = 60 + + # Mock the port change observer + mock_observer_instance = MagicMock() + mock_observer_instance.handle_port_update_event = MagicMock() + mock_observer_class.return_value = mock_observer_instance + + # Setup port mapping with one port + task.port_mapping.physical_to_logical = {1: ['Ethernet0']} + task.port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0) + task.get_dom_polling_from_config_db = MagicMock(return_value='enabled') + task.is_port_in_cmis_initialization_process = MagicMock(return_value=False) + + # Mock all the DB update methods + task.dom_db_utils = MagicMock() + task.status_db_utils = MagicMock() + task.vdm_utils = MagicMock() + task.vdm_utils.is_transceiver_vdm_supported = MagicMock(return_value=False) + + # Strategy: Every call to now() advances time by 1 second. + # When the DOM polling function is called, we advance time by 30 seconds to simulate long processing. + # This lets us verify that scheduling uses loop start time, not loop end time: if loop end were used, + # the next iteration's DOM polling would be ~90s after the first, instead of ~60s. + base_time = datetime.datetime(2024, 1, 1, 12, 0, 0) + + current_time = [base_time] + first_dom_loop_start_time = [None] + second_dom_loop_start_time = [None] + dom_call_count = [0] + + def mock_now(): + # Advance time by 1 second on each call + current_time[0] = current_time[0] + datetime.timedelta(seconds=1) + return current_time[0] + + # Wrap the DOM polling function to track when it's called and simulate long processing + dom_sensor_mock = task.dom_db_utils.post_port_dom_sensor_info_to_db + + def dom_sensor_side_effect(logical_port_name): + dom_call_count[0] += 1 + + if dom_call_count[0] == 1: + # Approximate loop-start time for first iteration + first_dom_loop_start_time[0] = current_time[0] + # Simulate long processing: advance time by 30 seconds + current_time[0] = current_time[0] + datetime.timedelta(seconds=30) + elif dom_call_count[0] == 2: + # Approximate loop-start time for second iteration + second_dom_loop_start_time[0] = current_time[0] + # Simulate long processing again + current_time[0] = current_time[0] + datetime.timedelta(seconds=30) + + # We don't need to call the original MagicMock explicitly; returning None is fine. + return None + + dom_sensor_mock.side_effect = dom_sensor_side_effect + + # Patch datetime.datetime.now in the dom_mgr module + with patch('xcvrd.dom.dom_mgr.datetime.datetime') as mock_datetime: + mock_datetime.now = MagicMock(side_effect=mock_now) + mock_datetime.timedelta = datetime.timedelta + + # Stop the task after we've seen two DOM DB updates + def mock_is_set(): + return dom_call_count[0] >= 2 + + task.task_stopping_event.is_set = MagicMock(side_effect=mock_is_set) + + # Run the task worker + task.task_worker() + + # We expect two periodic DOM updates to have happened + assert dom_call_count[0] >= 2, \ + f"Expected at least 2 DOM sensor DB updates, got {dom_call_count[0]}" + + assert first_dom_loop_start_time[0] is not None and second_dom_loop_start_time[0] is not None, \ + "DOM sensor DB updates did not run twice as expected" + + delta = (second_dom_loop_start_time[0] - first_dom_loop_start_time[0]).total_seconds() + + # If scheduling uses loop-start time, the gap between iterations should be close to + # dom_update_interval (60s) and significantly less than 90s (which would + # include the simulated 30s processing time). + assert delta < 70, \ + f"Expected time between iterations to be based on loop-start time (~60s), got {delta} seconds" + + # Also verify that the other DOM-related DB methods ran at least twice + assert task.dom_db_utils.post_port_dom_sensor_info_to_db.call_count >= 2, \ + f"Expected at least 2 calls (one per iteration), got {task.dom_db_utils.post_port_dom_sensor_info_to_db.call_count}" + assert task.dom_db_utils.post_port_dom_flags_to_db.call_count >= 2, \ + f"Expected at least 2 calls, got {task.dom_db_utils.post_port_dom_flags_to_db.call_count}" + assert task.status_db_utils.post_port_transceiver_hw_status_to_db.call_count >= 2, \ + f"Expected at least 2 calls, got {task.status_db_utils.post_port_transceiver_hw_status_to_db.call_count}" + assert task.status_db_utils.post_port_transceiver_hw_status_flags_to_db.call_count >= 2, \ + f"Expected at least 2 calls, got {task.status_db_utils.post_port_transceiver_hw_status_flags_to_db.call_count}" + def test_DomInfoUpdateTask_dom_update_interval_parameter(self): """Test that DomInfoUpdateTask correctly handles dom_update_interval parameter""" port_mapping = PortMapping() diff --git a/sonic-xcvrd/xcvrd/dom/dom_mgr.py b/sonic-xcvrd/xcvrd/dom/dom_mgr.py index 46f753d..2ca37dc 100644 --- a/sonic-xcvrd/xcvrd/dom/dom_mgr.py +++ b/sonic-xcvrd/xcvrd/dom/dom_mgr.py @@ -33,6 +33,8 @@ raise ImportError(str(e) + " - required module not found in dom_mgr.py") SYSLOG_IDENTIFIER_DOMINFOUPDATETASK = "DomInfoUpdateTask" +PORT_UPDATE_EVENT_SELECT_TIMEOUT_MSECS = 1000 +PORT_UPDATE_EVENT_SELECT_TIMEOUT_FAST_MSECS = 10 class DomInfoUpdateBase(threading.Thread): @@ -262,6 +264,23 @@ def post_port_pm_info_to_db(self, logical_port_name, port_mapping, table, stop_e else: return xcvrd.SFP_EEPROM_NOT_READY + def check_port_update(self, port_change_observer, timeout): + # Process pending link change events and update diagnostic + # information in the database. Ensures timely handling of link + # change events and avoids duplicate updates in case of breakout ports. + port_change_observer.handle_port_update_event(timeout) + + # Process each port in the pending link change set based on the + # corresponding time to update the DB after the link change. + for link_changed_port in list(self.link_change_affected_ports.keys()): + if self.task_stopping_event.is_set(): + self.log_notice("Stop event generated during DOM link change event processing") + break + if self.link_change_affected_ports[link_changed_port] <= datetime.datetime.now(): + self.log_notice(f"Updating port db diagnostics post link change for port {link_changed_port}") + self.update_port_db_diagnostics_on_link_change(link_changed_port) + del self.link_change_affected_ports[link_changed_port] + def task_worker(self): self.log_notice("Start DOM monitoring loop") sel, asic_context = port_event_helper.subscribe_port_config_change(self.namespaces) @@ -277,42 +296,39 @@ def task_worker(self): # Adding dom_info_update_periodic_secs to allow xcvrd to initialize ports # before starting the periodic update next_periodic_db_update_time = datetime.datetime.now() + datetime.timedelta(seconds=dom_info_update_periodic_secs) - is_periodic_db_update_needed = False # Start loop to update dom info in DB periodically and handle port change events while not self.task_stopping_event.is_set(): - # Check if periodic db update is needed - now = datetime.datetime.now() - if next_periodic_db_update_time <= now: - is_periodic_db_update_needed = True - # Handle port change event from main thread port_event_helper.handle_port_config_change(sel, asic_context, self.task_stopping_event, self.port_mapping, self.helper_logger, self.on_port_config_change) + while True: + remaining_secs = (next_periodic_db_update_time - datetime.datetime.now()).total_seconds() + if remaining_secs <= 0: + break + # Cap select timeout at PORT_UPDATE_EVENT_SELECT_TIMEOUT_MSECS to remain responsive + # to port change events, but shrink it to the remaining time when smaller so the + # next periodic DOM update is not delayed by an entire select cycle. + select_timeout_msecs = min(PORT_UPDATE_EVENT_SELECT_TIMEOUT_MSECS, + max(1, int(remaining_secs * 1000))) + self.check_port_update(port_change_observer, select_timeout_msecs) + + if self.task_stopping_event.is_set(): + self.log_notice("Stop event generated during DOM monitoring loop while checking port update") + break + + # Exit outer loop immediately on stop event to avoid a partial DOM polling pass + if self.task_stopping_event.is_set(): + break + + dom_loop_start_time = datetime.datetime.now() for physical_port, logical_ports in self.port_mapping.physical_to_logical.items(): - # Process pending link change events and update diagnostic - # information in the database. Ensures timely handling of link - # change events and avoids duplicate updates in case of breakout ports. - port_change_observer.handle_port_update_event() - # Process each port in the pending link change set based on the - # corresponding time to update the DB after the link change. - for link_changed_port in list(self.link_change_affected_ports.keys()): - if self.task_stopping_event.is_set(): - self.log_notice("Stop event generated during DOM link change event processing") - break - if self.link_change_affected_ports[link_changed_port] <= datetime.datetime.now(): - self.log_notice(f"Updating port db diagnostics post link change for port {link_changed_port}") - self.update_port_db_diagnostics_on_link_change(link_changed_port) - del self.link_change_affected_ports[link_changed_port] + self.check_port_update(port_change_observer, PORT_UPDATE_EVENT_SELECT_TIMEOUT_FAST_MSECS) if self.task_stopping_event.is_set(): self.log_notice("Stop event generated during DOM monitoring loop") break - if not is_periodic_db_update_needed: - # If periodic db update is not needed, skip the rest of the loop - continue - # Get the first logical port name since it corresponds to the first subport # of the breakout group logical_port_name = logical_ports[0] @@ -400,10 +416,8 @@ def task_worker(self): except (KeyError, TypeError) as e: self.log_warning("Got exception {} while processing vdm flags for port {}, ignored".format(repr(e), logical_port_name)) - # Set the periodic db update time after all the ports are processed - if is_periodic_db_update_needed: - next_periodic_db_update_time = now + datetime.timedelta(seconds=dom_info_update_periodic_secs) - is_periodic_db_update_needed = False + # Schedule next poll from loop start time for consistent intervals + next_periodic_db_update_time = dom_loop_start_time + datetime.timedelta(seconds=dom_info_update_periodic_secs) self.log_notice("Stop DOM monitoring loop") diff --git a/sonic-xcvrd/xcvrd/xcvrd_utilities/port_event_helper.py b/sonic-xcvrd/xcvrd/xcvrd_utilities/port_event_helper.py index c2002bc..e6bc439 100644 --- a/sonic-xcvrd/xcvrd/xcvrd_utilities/port_event_helper.py +++ b/sonic-xcvrd/xcvrd/xcvrd_utilities/port_event_helper.py @@ -113,7 +113,7 @@ def subscribe_port_update_event(self): port_tbl, list(d.values())[0], namespace)) self.sel, self.asic_context = sel, asic_context - def handle_port_update_event(self): + def handle_port_update_event(self, timeout=SELECT_TIMEOUT_MSECS): """ Select PORT update events, notify the observers upon a port update in CONFIG_DB or a XCVR insertion/removal in STATE_DB @@ -123,7 +123,7 @@ def handle_port_update_event(self): """ has_event = False if not self.stop_event.is_set(): - (state, _) = self.sel.select(SELECT_TIMEOUT_MSECS) + (state, _) = self.sel.select(timeout) if state == swsscommon.Select.TIMEOUT: return has_event if state != swsscommon.Select.OBJECT: