Merge pull request #485 from FederatedAI/feature-1.11.2-deepspeed_log

Fix the issue of thread accumulation due to session cleanup timeouts.

Author: zhihuiwan

python/fate_flow/db/runtime_config.py

@@ -38,6 +38,7 @@ class RuntimeConfig(ReloadConfigBase):
     SERVICE_DB = None
     LOAD_COMPONENT_REGISTRY = False
     LOAD_CONFIG_MANAGER = False
+    SESSION_LIST = []
 
     @classmethod
     def init_config(cls, **kwargs):

python/fate_flow/detection/detector.py

@@ -225,19 +225,23 @@ def detect_expired_session(cls):
         detect_logger().info(f'start detect expired session by ttl {ttl/1000} s')
         try:
             session_records = Session.query_sessions(create_time=[None, current_timestamp() - ttl])
-            manager_session_id_list = []
             for session_record in session_records:
                 manager_session_id = session_record.f_manager_session_id
-                if manager_session_id in manager_session_id_list:
+                if manager_session_id in RuntimeConfig.SESSION_LIST:
                     continue
-                manager_session_id_list.append(manager_session_id)
+                else:
+                    RuntimeConfig.SESSION_LIST.append(manager_session_id)
                 detect_logger().info(f'start destroy session {manager_session_id}')
                 try:
                     sess = Session(session_id=manager_session_id, options={"logger": detect_logger()})
                     sess.destroy_all_sessions()
                 except Exception as e:
                     detect_logger().error(f'stop session {manager_session_id} error', e)
                 finally:
+                    try:
+                        RuntimeConfig.SESSION_LIST.remove(manager_session_id)
+                    except:
+                        pass
                     detect_logger().info(f'stop session {manager_session_id} successfully')
         except Exception as e:
             detect_logger().error('detect expired session error', e)