debugging consecutive libE calls with gen_on_manager

Author: jlnav

libensemble/comms/comms.py

@@ -207,19 +207,6 @@ def result(self, timeout=None):
             raise RemoteException(self._exception.msg, self._exception.exc)
         return self._result
 
-    @staticmethod
-    def _qcomm_main(comm, main, *args, **kwargs):
-        """Main routine -- handles return values and exceptions."""
-        try:
-            if not kwargs.get("user_function"):
-                _result = main(comm, *args, **kwargs)
-            else:
-                _result = main(*args)
-            comm.send(CommResult(_result))
-        except Exception as e:
-            comm.send(CommResultErr(str(e), format_exc()))
-            raise e
-
     @property
     def running(self):
         """Check if the thread/process is running."""
@@ -233,6 +220,19 @@ def __exit__(self, etype, value, traceback):
         self.handle.join()
 
 
+def _qcomm_main(comm, main, *args, **kwargs):
+    """Main routine -- handles return values and exceptions."""
+    try:
+        if not kwargs.get("user_function"):
+            _result = main(comm, *args, **kwargs)
+        else:
+            _result = main(*args)
+        comm.send(CommResult(_result))
+    except Exception as e:
+        comm.send(CommResultErr(str(e), format_exc()))
+        raise e
+
+
 class QCommThread(QCommLocal):
     """Launch a user function in a thread with an attached QComm."""
 
@@ -241,7 +241,7 @@ def __init__(self, main, nworkers, *args, **kwargs):
         self.outbox = thread_queue.Queue()
         super().__init__(self, main, *args, **kwargs)
         comm = QComm(self.inbox, self.outbox, nworkers)
-        self.handle = Thread(target=QCommThread._qcomm_main, args=(comm, main) + args, kwargs=kwargs)
+        self.handle = Thread(target=_qcomm_main, args=(comm, main) + args, kwargs=kwargs)
 
     def terminate(self, timeout=None):
         """Terminate the thread.
@@ -265,7 +265,7 @@ def __init__(self, main, nworkers, *args, **kwargs):
         self.outbox = Queue()
         super().__init__(self, main, *args, **kwargs)
         comm = QComm(self.inbox, self.outbox, nworkers)
-        self.handle = Process(target=QCommProcess._qcomm_main, args=(comm, main) + args, kwargs=kwargs)
+        self.handle = Process(target=_qcomm_main, args=(comm, main) + args, kwargs=kwargs)
 
     def terminate(self, timeout=None):
         """Terminate the process."""

libensemble/comms/logs.py

@@ -203,6 +203,7 @@ def manager_logging_config(specs={}):
     def exit_logger():
         stat_timer.stop()
         stat_logger.info(f"Exiting ensemble at: {stat_timer.date_end} Time Taken: {stat_timer.elapsed}")
+        stat_logger.handlers[0].close()
 
         # If closing logs - each libE() call will log to a new file.
         # fh.close()

libensemble/manager.py

@@ -302,6 +302,9 @@ def _kill_workers(self) -> None:
         """Kills the workers"""
         for w in self.W["worker_id"]:
             self.wcomms[w].send(STOP_TAG, MAN_SIGNAL_FINISH)
+            if w == 0:
+                self.wcomms[0].result()
+                self.wcomms[0] = None
 
     # --- Checkpointing logic
 
@@ -691,6 +694,7 @@ def run(self, persis_info: dict) -> (dict, int, int):
         finally:
             # Return persis_info, exit_flag, elapsed time
             result = self._final_receive_and_kill(persis_info)
+            self.wcomms = None
             sys.stdout.flush()
             sys.stderr.flush()
         return result

libensemble/tests/functionality_tests/test_GPU_gen_resources.py

@@ -42,12 +42,6 @@
 from libensemble.sim_funcs.var_resources import gpu_variable_resources_from_gen as sim_f
 from libensemble.tools import add_unique_random_streams, parse_args
 
-# TODO: multiple libE calls with gen-on-manager currently not supported with spawn on macOS
-if sys.platform == "darwin":
-    from multiprocessing import set_start_method
-
-    set_start_method("fork", force=True)
-
 # from libensemble import logger
 # logger.set_level("DEBUG")  # For testing the test