fix(xcvrd): handle port add/remove events in CmisManagerTask

This commit fixes DPB support with CMIS transceivers.

CmisManagerTask's port_dict and port_mapping must be updated
according to the port add/remove events.
Sinse CmisManagerTask uses PortChangeObserver, which emits
PORT_SET/PORT_DEL events, not PORT_ADD/PORT_REMOVE,
PortMapping is updated to handle these events.

fixes sonic-net/sonic-buildimage#18893

Signed-off-by: Wataru Ishida <[email protected]>

sonic-xcvrd/tests/test_xcvrd.py

@@ -204,7 +204,10 @@ def test_CmisManagerTask_get_xcvr_api_exception(self, mock_platform_chassis, moc
         # Case 1: get_xcvr_api() raises an exception
         task.on_port_update_event(port_change_event)
         mock_sfp.get_xcvr_api = MagicMock(side_effect=NotImplementedError)
-        task.task_worker()
+        assert len(task.port_dict) == 1
+        for lport, info in task.port_dict.items():
+            task.handle_cmis_state_machine(lport, info, False)
+
         assert mock_log_exception_traceback.call_count == 1
         assert get_cmis_state_from_state_db('Ethernet0', task.xcvr_table_helper.get_status_tbl(task.port_mapping.get_asic_id_for_logical_port('Ethernet0'))) == CMIS_STATE_FAILED
 
@@ -214,7 +217,9 @@ def test_CmisManagerTask_get_xcvr_api_exception(self, mock_platform_chassis, moc
         mock_xcvr_api.is_flat_memory = MagicMock(side_effect=AttributeError)
         task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True])
         task.on_port_update_event(port_change_event)
-        task.task_worker()
+        assert len(task.port_dict) == 1
+        for lport, info in task.port_dict.items():
+            task.handle_cmis_state_machine(lport, info, False)
         assert get_cmis_state_from_state_db('Ethernet0', task.xcvr_table_helper.get_status_tbl(task.port_mapping.get_asic_id_for_logical_port('Ethernet0'))) == CMIS_STATE_READY
 
         # Case 3: get_cmis_application_desired() raises an exception
@@ -224,7 +229,9 @@ def test_CmisManagerTask_get_xcvr_api_exception(self, mock_platform_chassis, moc
         task.task_stopping_event.is_set = MagicMock(side_effect=[False, False, True])
         task.on_port_update_event(port_change_event)
         task.get_cmis_host_lanes_mask = MagicMock()
-        task.task_worker()
+        assert len(task.port_dict) == 1
+        for lport, info in task.port_dict.items():
+            task.handle_cmis_state_machine(lport, info, False)
         assert mock_log_exception_traceback.call_count == 2
         assert get_cmis_state_from_state_db('Ethernet0', task.xcvr_table_helper.get_status_tbl(task.port_mapping.get_asic_id_for_logical_port('Ethernet0'))) == CMIS_STATE_FAILED
         assert task.get_cmis_host_lanes_mask.call_count == 0
@@ -1470,22 +1477,42 @@ def test_CmisManagerTask_handle_port_change_event(self):
         task.on_port_update_event(port_change_event)
         assert task.isPortConfigDone
 
+        assert len(task.port_dict) == 0
+        assert len(task.deleted_ports) == 0
+        # CmisManagerTask doesn't handle PORT_ADD event.
+        # No change on port_dict
         port_change_event = PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_ADD)
         task.on_port_update_event(port_change_event)
         assert len(task.port_dict) == 0
+        assert len(task.deleted_ports) == 0
 
+        # CmisManagerTask doesn't handle PORT_REMOVE event.
+        # No change on port_dict
         port_change_event = PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_REMOVE)
         task.on_port_update_event(port_change_event)
         assert len(task.port_dict) == 0
+        assert len(task.deleted_ports) == 0
 
+        # PORT_DEL event before handling PORT_SET
+        # No change on port_dict
         port_change_event = PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_DEL)
         task.on_port_update_event(port_change_event)
-        assert len(task.port_dict) == 1
+        assert len(task.port_dict) == 0
+        assert len(task.deleted_ports) == 0
 
         port_change_event = PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_SET)
         task.on_port_update_event(port_change_event)
         assert len(task.port_dict) == 1
+        assert len(task.deleted_ports) == 0
+        assert len(task.port_mapping.logical_port_list) == 1
 
+        # PORT_DEL event after handling PORT_SET
+        # No change on port_dict
+        # deleted_ports should have been updated
+        port_change_event = PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_DEL)
+        task.on_port_update_event(port_change_event)
+        assert len(task.port_dict) == 1
+        assert len(task.deleted_ports) == 1
 
     @patch('xcvrd.xcvrd.XcvrTableHelper')
     def test_CmisManagerTask_get_configured_freq(self, mock_table_helper):

sonic-xcvrd/xcvrd/xcvrd.py

@@ -829,6 +829,7 @@ def __init__(self, namespaces, port_mapping, main_thread_stop_event, skip_cmis_m
         self.main_thread_stop_event = main_thread_stop_event
         self.port_dict = {}
         self.port_mapping = copy.deepcopy(port_mapping)
+        self.deleted_ports = {}
         self.xcvr_table_helper = XcvrTableHelper(namespaces)
         self.isPortInitDone = False
         self.isPortConfigDone = False
@@ -860,6 +861,7 @@ def on_port_update_event(self, port_change_event):
             return
 
         lport = port_change_event.port_name
+        # 'index' can be -1 if STATE_DB|PORT_TABLE
         pport = port_change_event.port_index
 
         if lport in ['PortInitDone']:
@@ -878,15 +880,14 @@ def on_port_update_event(self, port_change_event):
         if pport is None:
             return
 
-        # Skip if the port/cage type is not a CMIS
-        # 'index' can be -1 if STATE_DB|PORT_TABLE
-        if lport not in self.port_dict:
-            self.port_dict[lport] = {}
+        if port_change_event.event_type == port_change_event.PORT_SET:
+            if lport not in self.port_dict:
+                self.port_dict[lport] = {}
+                self.port_mapping.handle_port_change_event(port_change_event)
 
-        if port_change_event.port_dict is None:
-            return
+            if port_change_event.port_dict is None:
+                return
 
-        if port_change_event.event_type == port_change_event.PORT_SET:
             if pport >= 0:
                 self.port_dict[lport]['index'] = pport
             if 'speed' in port_change_event.port_dict and port_change_event.port_dict['speed'] != 'N/A':
@@ -906,7 +907,13 @@ def on_port_update_event(self, port_change_event):
 
             self.force_cmis_reinit(lport, 0)
         else:
+            # PORT_DEL event for the same lport happens 3 times because
+            # we are subscribing to CONFIG_DB, STATE_DB|TRANSCEIVER_INFO, and STATE_DB|PORT_TABLE.
+            # We only handle the first one and ignore the rest.
+            if lport in self.deleted_ports or lport not in self.port_dict:
+                return
             self.update_port_transceiver_status_table_sw_cmis_state(lport, CMIS_STATE_REMOVED)
+            self.deleted_ports[lport] = port_change_event
 
     def get_cmis_dp_init_duration_secs(self, api):
         return api.get_datapath_init_duration()/1000
@@ -1647,6 +1654,12 @@ def task_worker(self):
 
                 self.handle_cmis_state_machine(lport, info, is_fast_reboot)
 
+            for event in self.deleted_ports.values():
+                self.port_mapping.handle_port_change_event(event)
+                self.port_dict.pop(event.port_name)
+
+            self.deleted_ports = {}
+
         self.log_notice("Stopped")
 
     def run(self):

sonic-xcvrd/xcvrd/xcvrd_utilities/port_event_helper.py

@@ -214,9 +214,9 @@ def __init__(self):
         self.logical_to_asic = {}
 
     def handle_port_change_event(self, port_change_event):
-        if port_change_event.event_type == PortChangeEvent.PORT_ADD:
+        if port_change_event.event_type in [PortChangeEvent.PORT_ADD, PortChangeEvent.PORT_SET]:
             self._handle_port_add(port_change_event)
-        elif port_change_event.event_type == PortChangeEvent.PORT_REMOVE:
+        elif port_change_event.event_type in [PortChangeEvent.PORT_REMOVE, PortChangeEvent.PORT_DEL]:
             self._handle_port_remove(port_change_event)
 
     def _handle_port_add(self, port_change_event):