Fix pipe handler timeout in task exchanger and launcher executor (#2495)

NVIDIA · Apr 12, 2024 · 47b2feb · 47b2feb
1 parent f4df50f
commit 47b2feb
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/nvflare/app_common/executors/launcher_executor.py b/nvflare/app_common/executors/launcher_executor.py
@@ -44,7 +44,7 @@ def __init__(
         last_result_transfer_timeout: float = 300.0,
         external_pre_init_timeout: float = 60.0,
         peer_read_timeout: Optional[float] = 60.0,
-        monitor_interval: float = 1.0,
+        monitor_interval: float = 0.1,
         read_interval: float = 0.5,
         heartbeat_interval: float = 5.0,
         heartbeat_timeout: float = 60.0,
@@ -373,8 +373,8 @@ def _monitor_launcher(self, fl_ctx: FLContext):
                 if self.launcher is None:
                     break
 
+                # no task is running
                 if self._current_task is None:
-                    self.pause_pipe_handler()
                     continue
 
                 task_name = self._current_task
@@ -390,16 +390,19 @@ def _monitor_launcher(self, fl_ctx: FLContext):
                     continue
 
                 elif run_status == LauncherRunStatus.NOT_RUNNING:
+                    # pause pipe handler because external process is not running
                     self.pause_pipe_handler()
                     continue
 
                 elif run_status == LauncherRunStatus.RUNNING:
+                    # resume pipe handler when external process is running
                     self.resume_pipe_handler()
                     continue
 
                 elif (
                     run_status == LauncherRunStatus.COMPLETE_FAILED or run_status == LauncherRunStatus.COMPLETE_SUCCESS
                 ):
+                    # pause pipe handler because external process is completed
                     self.pause_pipe_handler()
                     if not self._launcher_finish:
                         self._launcher_finish_time = time.time()

diff --git a/nvflare/app_common/executors/task_exchanger.py b/nvflare/app_common/executors/task_exchanger.py
@@ -111,7 +111,6 @@ def handle_event(self, event_type: str, fl_ctx: FLContext):
             )
             self.pipe_handler.set_status_cb(self._pipe_status_cb)
             self.pipe.open(self.pipe_channel_name)
-        elif event_type == EventType.BEFORE_TASK_EXECUTION:
             self.pipe_handler.start()
         elif event_type == EventType.ABOUT_TO_END_RUN:
             self.log_info(fl_ctx, "Stopping pipe handler")