Enhance launcher executor and subprocess launcher

nvflare/app_common/executors/launcher_executor.py

@@ -227,6 +227,7 @@ def _init_converter(self, fl_ctx: FLContext):
  def _initialize_external_execution(
  self, task_name: str, shareable: Shareable, fl_ctx: FLContext, abort_signal: Signal
  ) -> bool:
+ self.reset_peer_is_up_or_dead()
  with self._lock:
  self._abort_signal = abort_signal
  self._current_task = task_name
@@ -242,13 +243,15 @@ def _initialize_external_execution(
  abort_signal.trigger("launch task failed")
  return False
 
- self.log_info(fl_ctx, f"External execution for task ({task_name}) is launched.")
+ self.log_info(fl_ctx, f"launcher launch_task for ({task_name}) succeed.")
  # wait for external execution to set up their pipe_handler
  setup_success = self._wait_external_setup(task_name, fl_ctx, abort_signal)
  if not setup_success:
- self.log_error(fl_ctx, "External execution set up failed.")
- abort_signal.trigger("External execution set up failed.")
+ error = f"External execution set up for task ({task_name}) failed."
+ self.log_error(fl_ctx, error)
+ abort_signal.trigger(error)
  return False
+ self.log_info(fl_ctx, f"External execution set up for task ({task_name}) succeed.")
  return True
 
  def _execute_launcher_method_in_thread_executor(self, method_name: str, **kwargs) -> Any:

nvflare/app_common/executors/task_exchanger.py

@@ -250,6 +250,9 @@ def ask_peer_to_end(self, fl_ctx: FLContext) -> bool:
  def peer_is_up_or_dead(self) -> bool:
  return self.pipe_handler.peer_is_up_or_dead.is_set()
 
+ def reset_peer_is_up_or_dead(self):
+ self.pipe_handler.peer_is_up_or_dead.clear()
+
  def pause_pipe_handler(self):
  """Stops pipe_handler heartbeat."""
  self.pipe_handler.pause()