diff --git a/sonic-xcvrd/tests/test_xcvrd.py b/sonic-xcvrd/tests/test_xcvrd.py index e3e3cd8..9b348f5 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 @@ -5169,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 @@ -5197,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')) @@ -5227,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) @@ -5266,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 @@ -5285,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 @@ -5293,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)) @@ -5313,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) @@ -5340,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) @@ -5367,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) @@ -5393,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) @@ -6610,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/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) 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: