temp commit for independent modules code

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

@@ -31,7 +31,10 @@
     from . import utils
     from .device_data import DeviceDataManager
     import re
+    import queue
+    import threading
     import time
+    from sonic_platform import modules_mgmt
 except ImportError as e:
     raise ImportError (str(e) + "- required module not found")
 
@@ -123,6 +126,12 @@ def __init__(self):
         self._RJ45_port_inited = False
         self._RJ45_port_list = None
 
+        self.threads = []
+        self.modules_mgmt_thread = threading.Thread()
+        self.modules_changes_queue = queue.Queue()
+        self.modules_queue_lock = threading.Lock()
+        self.modules_mgmt_task_stopping_event = threading.Event()
+
         logger.log_info("Chassis loaded successfully")
 
     def __del__(self):
@@ -133,6 +142,11 @@ def __del__(self):
             if self.sfp_module.SFP.shared_sdk_handle:
                 self.sfp_module.deinitialize_sdk_handle(self.sfp_module.SFP.shared_sdk_handle)
 
+        #self.modules_mgmt_task_stopping_event.set()
+        #logger.log_info('set modules_mgmt_task_stopping_event {self.modules_mgmt_task_stopping_event}')
+        #self.modules_mgmt_thread.join(timeout=10)
+        #logger.log_info('joined modules_mgmt_thread thread')
+
     @property
     def RJ45_port_list(self):
         if not self._RJ45_port_inited:
@@ -366,11 +380,11 @@ def get_change_event(self, timeout=0):
 
         Args:
             timeout: Timeout in milliseconds (optional). If timeout == 0,
-                this method will block until a change is detected.
+                this method will block until a change is detected. - Deprecated
 
         Returns:
             (bool, dict):
-                - True if call successful, False if not;
+                - 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'},
@@ -382,38 +396,38 @@ 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
+                                                    , l=self.modules_queue_lock
+                                                    , main_thread_stop_event = self.modules_mgmt_task_stopping_event)
+            self.modules_mgmt_thread.start()
+            self.threads.append(self.modules_mgmt_thread)
         self.initialize_sfp()
-        # Initialize SFP event first
-        if not self.sfp_event:
-            from .sfp_event import sfp_event
-            self.sfp_event = sfp_event(self.RJ45_port_list)
-            self.sfp_event.initialize()
-
-        wait_for_ever = (timeout == 0)
-        # select timeout should be no more than 1000ms to ensure fast shutdown flow
-        select_timeout = 1000.0 if timeout >= 1000 else float(timeout)
         port_dict = {}
         error_dict = {}
-        begin = time.time()
+        i = 0
         while True:
-            status = self.sfp_event.check_sfp_status(port_dict, error_dict, select_timeout)
-            if bool(port_dict):
-                break
+            logger.log_info('get_change_event() acquiring queue lock iteration {}'.format(i))
+            self.modules_queue_lock.acquire()
+            if self.modules_changes_queue.qsize() > 0:
+                if True:
+                    try:
+                        logger.log_info('get_change_event() trying to get changes from queue')
+                        port_dict = self.modules_changes_queue.get(timeout=1)
+                        logger.log_info ('get_change_event() port_dict: {}'.format(port_dict))
+                    except queue.Empty:
+                        logger.log_info("failed to get item from modules changes queue")
+            logger.log_info('get_change_event() releasing queue lock iteration {}'.format(i))
+            self.modules_queue_lock.release()
 
-            if not wait_for_ever:
-                elapse = time.time() - begin
-                if elapse * 1000 > timeout:
-                    break
-
-        if status:
             if port_dict:
                 self.reinit_sfps(port_dict)
-            result_dict = {'sfp': port_dict}
-            if error_dict:
+                result_dict = {'sfp': port_dict}
                 result_dict['sfp_error'] = error_dict
-            return True, result_dict
-        else:
-            return True, {'sfp': {}}
+                return True, result_dict
+            time.sleep(1)
+            i += 1
 
     def reinit_sfps(self, port_dict):
         """