[Mellanox] run module initialization when any SFP related API is called (#18930) (#18937)

Why I did it
Currently, there are a few issues related to module host management:

1. module initialization flow is triggered when chassis.get_change_event is called for the first time. It is too late which delays fast-reboot/warm-reboot convergence time

2. module initialization flow always send change event for all SFP objects to xcvrd even if there is no real module change event. It causes ports flapping during warm-reboot

3. legacy mode and module host management mode are mixed into the same code logic which makes it hard to maintain and extend

To address above issues, the PR introduces following changes:

1. module initialization flow is triggered when chassis SFP related API is called for the first time. It means that any of the following API will trigger module init: chassis.get_sfp, chassis.get_all_sfps, chassis.get_change_event. The init flow shall only execute once under xcvrd context.

2. module initialization flow does not send change event anymore. Change event shall be sent only if a real cable event is detected by chassis.get_change_event (plug-in/plug-out/error)

3. legacy mode and module host management mode are decoupled to avoid affecting each other

How to verify it
1. Unit test covered most of the new change
2. Full sonic-mgmt regression test make sure there is no degradation (test branch 202311)
3. Manual test

Author: Junchao-Mellanox

platform/mellanox/mlnx-platform-api/sonic_platform/chassis.py

@@ -28,13 +28,13 @@
     import os
     from functools import reduce
     from .utils import extract_RJ45_ports_index
+    from . import module_host_mgmt_initializer
     from . import utils
     from .device_data import DeviceDataManager
     import re
-    import queue
+    import select
     import threading
     import time
-    from sonic_platform import modules_mgmt
 except ImportError as e:
     raise ImportError (str(e) + "- required module not found")
 
@@ -132,9 +132,9 @@ def __init__(self):
 
         Chassis.chassis_instance = self
 
-        self.modules_mgmt_thread = threading.Thread()
-        self.modules_changes_queue = queue.Queue()
-        self.modules_mgmt_task_stopping_event = threading.Event()
+        self.module_host_mgmt_initializer = module_host_mgmt_initializer.ModuleHostMgmtInitializer()
+        self.poll_obj = None
+        self.registered_fds = None
 
         logger.log_info("Chassis loaded successfully")
 
@@ -344,8 +344,11 @@ def get_all_sfps(self):
         Returns:
             A list of objects derived from SfpBase representing all sfps
             available on this chassis
-        """
-        self.initialize_sfp()
+        """    
+        if DeviceDataManager.is_module_host_management_mode():
+            self.module_host_mgmt_initializer.initialize(self)
+        else:
+            self.initialize_sfp()
         return self._sfp_list
 
     def get_sfp(self, index):
@@ -362,7 +365,10 @@ def get_sfp(self, index):
             An object dervied from SfpBase representing the specified sfp
         """
         index = index - 1
-        self.initialize_single_sfp(index)
+        if DeviceDataManager.is_module_host_management_mode():
+            self.module_host_mgmt_initializer.initialize(self)
+        else:
+            self.initialize_single_sfp(index)
         return super(Chassis, self).get_sfp(index)
 
     def get_port_or_cage_type(self, index):
@@ -412,42 +418,223 @@ def get_change_event(self, timeout=0):
                       indicates that fan 0 has been removed, fan 2
                       has been inserted and sfp 11 has been removed.
         """
-        if not self.modules_mgmt_thread.is_alive():
-            # open new SFP change events thread
-            self.modules_mgmt_thread = modules_mgmt.ModulesMgmtTask(q=self.modules_changes_queue
-                                                    , main_thread_stop_event = self.modules_mgmt_task_stopping_event)
-            # Set the thread as daemon so when pmon/xcvrd are shutting down, modules_mgmt will shut down immedietly.
-            self.modules_mgmt_thread.daemon = True
-            self.modules_mgmt_thread.start()
-        self.initialize_sfp()
-        wait_for_ever = (timeout == 0)
+        if DeviceDataManager.is_module_host_management_mode():
+            self.module_host_mgmt_initializer.initialize(self)
+            return self.get_change_event_for_module_host_management_mode(timeout)
+        else:
+            self.initialize_sfp()
+            return self.get_change_event_legacy(timeout)
+            
+    def get_change_event_for_module_host_management_mode(self, timeout):
+        """Get SFP change event when module host management mode is enabled.
+
+        Args:
+            timeout: Timeout in milliseconds (optional). If timeout == 0,
+                this method will block until a change is detected.
+
+        Returns:
+            (bool, dict):
+                - True if call successful, False if not; - Deprecated, will always return True
+                - A nested dictionary where key is a device type,
+                  value is a dictionary with key:value pairs in the format of
+                  {'device_id':'device_event'},
+                  where device_id is the device ID for this device and
+                        device_event,
+                             status='1' represents device inserted,
+                             status='0' represents device removed.
+                  Ex. {'fan':{'0':'0', '2':'1'}, 'sfp':{'11':'0'}}
+                      indicates that fan 0 has been removed, fan 2
+                      has been inserted and sfp 11 has been removed.
+        """
+        if not self.poll_obj:
+            self.poll_obj = select.poll()
+            self.registered_fds = {}
+            for s in self._sfp_list:
+                fds = s.get_fds_for_poling()
+                for fd_type, fd in fds.items():
+                    self.poll_obj.register(fd, select.POLLERR | select.POLLPRI)
+                    self.registered_fds[fd.fileno()] = (s.sdk_index, fd, fd_type)
+
+            logger.log_debug(f'Registered SFP file descriptors for polling: {self.registered_fds}')
+                    
+        from . import sfp
+        
+        wait_forever = (timeout == 0)
+        # poll timeout should be no more than 1000ms to ensure fast shutdown flow
+        timeout = 1000.0 if timeout >= 1000 else float(timeout)
+        port_dict = {}
+        error_dict = {}
+        begin = time.time()
+        wait_ready_task = sfp.SFP.get_wait_ready_task()
+        
+        while True:        
+            fds_events = self.poll_obj.poll(timeout)
+            for fileno, _ in fds_events:
+                if fileno not in self.registered_fds:
+                    logger.log_error(f'Unknown file no {fileno} from poll event, registered files are {self.registered_fds}')
+                    continue
+                
+                sfp_index, fd, fd_type = self.registered_fds[fileno]
+                s = self._sfp_list[sfp_index]
+                fd.seek(0)
+                fd_value = int(fd.read().strip())
+
+                # Detecting dummy event
+                if s.is_dummy_event(fd_type, fd_value):
+                    # Ignore dummy event for the first poll, assume SDK only provide 1 dummy event
+                    logger.log_debug(f'Ignore dummy event {fd_type}:{fd_value} for SFP {sfp_index}')
+                    continue
+
+                logger.log_notice(f'Got SFP event: index={sfp_index}, type={fd_type}, value={fd_value}')
+                if fd_type == 'hw_present':
+                    # event could be EVENT_NOT_PRESENT or EVENT_PRESENT
+                    event = sfp.EVENT_NOT_PRESENT if fd_value == 0 else sfp.EVENT_PRESENT
+                    s.on_event(event)
+                elif fd_type == 'present':
+                    if str(fd_value) == sfp.SFP_STATUS_ERROR:
+                        # FW control cable got an error, no need trigger state machine
+                        sfp_status, error_desc = s.get_error_info_from_sdk_error_type()
+                        port_dict[sfp_index + 1] = sfp_status
+                        if error_desc: 
+                            error_dict[sfp_index + 1] = error_desc
+                        continue
+                    elif str(fd_value) == sfp.SFP_STATUS_INSERTED:
+                        # FW control cable got present, only case is that the cable is recovering
+                        # from an error. FW control cable has no transition from "Not Present" to "Present"
+                        # because "Not Present" cable is always "software control" and should always poll
+                        # hw_present sysfs instead of present sysfs.
+                        port_dict[sfp_index + 1] = sfp.SFP_STATUS_INSERTED
+                        continue
+                    else:
+                        s.on_event(sfp.EVENT_NOT_PRESENT)
+                else:
+                    # event could be EVENT_POWER_GOOD or EVENT_POWER_BAD
+                    event = sfp.EVENT_POWER_BAD if fd_value == 0 else sfp.EVENT_POWER_GOOD
+                    s.on_event(event)
+                    
+                if s.in_stable_state():
+                    s.fill_change_event(port_dict)
+                    s.refresh_poll_obj(self.poll_obj, self.registered_fds)
+                else:
+                    logger.log_debug(f'SFP {sfp_index} does not reach stable state, state={s.state}')
+                                        
+            ready_sfp_set = wait_ready_task.get_ready_set()
+            for sfp_index in ready_sfp_set:
+                s = self._sfp_list[sfp_index]
+                s.on_event(sfp.EVENT_RESET_DONE)
+                if s.in_stable_state():
+                    s.fill_change_event(port_dict)
+                    s.refresh_poll_obj(self.poll_obj, self.registered_fds)
+                else:
+                    logger.log_error(f'SFP {sfp_index} failed to reach stable state, state={s.state}')
+                    
+            if port_dict:
+                logger.log_notice(f'Sending SFP change event: {port_dict}, error event: {error_dict}')
+                self.reinit_sfps(port_dict)
+                return True, {
+                    'sfp': port_dict,
+                    'sfp_error': error_dict
+                }
+            else:
+                if not wait_forever:
+                    elapse = time.time() - begin
+                    if elapse * 1000 >= timeout:
+                        return True, {'sfp': {}}
+
+    def get_change_event_legacy(self, timeout):
+        """Get SFP change event when module host management is disabled.
+
+        Args:
+            timeout (int): polling timeout in ms
+
+        Returns:
+            (bool, dict):
+                - True if call successful, False if not; - Deprecated, will always return True
+                - A nested dictionary where key is a device type,
+                  value is a dictionary with key:value pairs in the format of
+                  {'device_id':'device_event'},
+                  where device_id is the device ID for this device and
+                        device_event,
+                             status='1' represents device inserted,
+                             status='0' represents device removed.
+                  Ex. {'fan':{'0':'0', '2':'1'}, 'sfp':{'11':'0'}}
+                      indicates that fan 0 has been removed, fan 2
+                      has been inserted and sfp 11 has been removed.
+        """
+        if not self.poll_obj:
+            self.poll_obj = select.poll()
+            self.registered_fds = {}
+            # SDK always sent event for the first time polling. Such event should not be sent to xcvrd.
+            # Store SFP state before first time polling so that we can detect dummy event.
+            self.sfp_states_before_first_poll = {}
+            for s in self._sfp_list:
+                fd = s.get_fd_for_polling_legacy()
+                self.poll_obj.register(fd, select.POLLERR | select.POLLPRI)
+                self.registered_fds[fd.fileno()] = (s.sdk_index, fd)
+                self.sfp_states_before_first_poll[s.sdk_index] = s.get_module_status()
+
+            logger.log_debug(f'Registered SFP file descriptors for polling: {self.registered_fds}')
+            
+        from . import sfp
+        
+        wait_forever = (timeout == 0)
         # poll timeout should be no more than 1000ms to ensure fast shutdown flow
         timeout = 1000.0 if timeout >= 1000 else float(timeout)
         port_dict = {}
         error_dict = {}
         begin = time.time()
-        i = 0
+        
         while True:
-            try:
-                logger.log_info(f'get_change_event() trying to get changes from queue on iteration {i}')
-                port_dict = self.modules_changes_queue.get(timeout=timeout / 1000)
-                logger.log_info(f'get_change_event() iteration {i} port_dict: {port_dict}')
-            except queue.Empty:
-                logger.log_info(f"failed to get item from modules changes queue on itertaion {i}")
+            fds_events = self.poll_obj.poll(timeout)
+            for fileno, _ in fds_events:
+                if fileno not in self.registered_fds:
+                    logger.log_error(f'Unknown file no {fileno} from poll event, registered files are {self.registered_fds}')
+                    continue
+                
+                sfp_index, fd = self.registered_fds[fileno]
+                fd.seek(0)
+                fd.read()
+                s = self._sfp_list[sfp_index]
+                sfp_status = s.get_module_status()
+
+                if sfp_index in self.sfp_states_before_first_poll:
+                    # Detecting dummy event
+                    sfp_state_before_poll = self.sfp_states_before_first_poll[sfp_index]
+                    self.sfp_states_before_first_poll.pop(sfp_index)
+                    if sfp_state_before_poll == sfp_status:
+                        # Ignore dummy event for the first poll, assume SDK only provide 1 dummy event
+                        logger.log_debug(f'Ignore dummy event {sfp_status} for SFP {sfp_index}')
+                        continue
+
+                logger.log_notice(f'Got SFP event: index={sfp_index}, value={sfp_status}')
+                if sfp_status == sfp.SFP_STATUS_UNKNOWN:
+                    # in the following sequence, STATUS_UNKNOWN can be returned.
+                    # so we shouldn't raise exception here.
+                    # 1. some sfp module is inserted
+                    # 2. sfp_event gets stuck and fails to fetch the change event instantaneously
+                    # 3. and then the sfp module is removed
+                    # 4. sfp_event starts to try fetching the change event
+                    logger.log_notice("unknown module state, maybe the port suffers two adjacent insertion/removal")
+                    continue
+
+                if sfp_status == sfp.SFP_STATUS_ERROR:
+                    sfp_status, error_desc = s.get_error_info_from_sdk_error_type()
+                    if error_desc:
+                        error_dict[sfp_index + 1] = error_desc
+                port_dict[sfp_index + 1] = sfp_status
 
             if port_dict:
+                logger.log_notice(f'Sending SFP change event: {port_dict}, error event: {error_dict}')
                 self.reinit_sfps(port_dict)
-                result_dict = {'sfp': port_dict}
-                result_dict['sfp_error'] = error_dict
-                return True, result_dict
+                return True, {
+                    'sfp': port_dict,
+                    'sfp_error': error_dict
+                }
             else:
-                if not wait_for_ever:
+                if not wait_forever:
                     elapse = time.time() - begin
-                    logger.log_info(f"get_change_event: wait_for_ever {wait_for_ever} elapse {elapse} iteartion {i}")
                     if elapse * 1000 >= timeout:
-                        logger.log_info(f"elapse {elapse} > timeout {timeout} iteartion {i} returning empty dict")
                         return True, {'sfp': {}}
-            i += 1
 
     def reinit_sfps(self, port_dict):
         """
@@ -457,7 +644,7 @@ def reinit_sfps(self, port_dict):
         """
         from . import sfp
         for index, status in port_dict.items():
-            if status == sfp.SFP_STATUS_INSERTED:
+            if status == sfp.SFP_STATUS_REMOVED:
                 try:
                     self._sfp_list[int(index) - 1].reinit()
                 except Exception as e:

platform/mellanox/mlnx-platform-api/sonic_platform/device_data.py

@@ -250,7 +250,7 @@ def get_cpld_component_list(cls):
 
     @classmethod
     @utils.read_only_cache()
-    def is_independent_mode(cls):
+    def is_module_host_management_mode(cls):
         from sonic_py_common import device_info
         _, hwsku_dir = device_info.get_paths_to_platform_and_hwsku_dirs()
         sai_profile_file = os.path.join(hwsku_dir, 'sai.profile')
@@ -266,7 +266,7 @@ def wait_platform_ready(cls):
         """
         conditions = []
         sysfs_nodes = ['power_mode', 'power_mode_policy', 'present', 'reset', 'status', 'statuserror']
-        if cls.is_independent_mode():
+        if cls.is_module_host_management_mode():
             sysfs_nodes.extend(['control', 'frequency', 'frequency_support', 'hw_present', 'hw_reset',
                                 'power_good', 'power_limit', 'power_on', 'temperature/input'])
         else: