Refactor/user function handling modules + Manager can run additional worker on thread (#1216)

** Commit Highlights: **

* first round of refactoring runners.py, Runner base class for normal in-place launches, but based on the contents of passed-in specs, instantiates the relevant subclass

* ThreadRunner uses comms.QCommThread, slightly modified, to launch its user function. corresponding unit test

* removing now-redundant content from manager, trying to see if we can start a temporary, local Worker for handling work

* use _Worker class to correctly index into W and wcomms. add initial option to libE_specs

* add "threaded" tentative option to sim/gen_specs

* fix ThreadRunner shutdown when that worker didn't launch a thread

* adjust alloc_support to not use w - 1 indexing

* add tentative gen_on_manager option, separate additional_worker_launch into function

* move _WorkerIndexer into libensemble.utils, also use within PersistentSupport

* manager also needs to send workflow_dir location to worker 0

* simply gen_workers parameter description for avail_worker_ids

* filter for gen_workers within avail_worker_ids, if set and there are gen_workers. solution resembles zrw, like shuds predicted all along!

* refactor give_sim_work_first for running on gen_workers if no points_to_evaluate. add test for mixed existing sample plus calling a gen

* it turns out that values set by validators are still considered "unset". So for updating purposes for libE_specs, we want to exclude fields that are still set to their defaults

* platform_specs sometimes seems to be at risk of disappearing when we convert LibeSpecs to dict, so lets save it and reinsert

* add libE_specs["gen_workers"] option, adjust ensure_one_active_gen so multiple gen work orders aren't given out at once

** Major features: **

* runners.py refactored to more easily develop additional methods to submit/run user functions

* User functions can be launched by workers on separate threads alternatively to calling in-place.

* Generator can be launched on the Manager - various refactors to make this possible

* gen_workers option to specify workers that should run generators only

* Various bugfixes and refactors of allocation functions

---------

Co-authored-by: shudson <[email protected]>
Co-authored-by: Jeffrey Larson <[email protected]>

Author: 3 people

docs/data_structures/libE_specs.rst

@@ -28,7 +28,11 @@ libEnsemble is primarily customized by setting options within a ``LibeSpecs`` cl
                     Manager/Worker communications mode: ``'mpi'``, ``'local'``, or ``'tcp'``.
 
                 **nworkers** [int]:
-                    Number of worker processes in ``"local"`` or ``"tcp"``.
+                    Number of worker processes in ``"local"``, ``"threads"``, or ``"tcp"``.
+
+                **gen_on_manager** Optional[bool] = False
+                    Instructs Manager process to run generator functions.
+                    This generator function can access/modify user objects by reference.
 
                 **mpi_comm** [MPI communicator] = ``MPI.COMM_WORLD``:
                     libEnsemble MPI communicator.
@@ -51,6 +55,10 @@ libEnsemble is primarily customized by setting options within a ``LibeSpecs`` cl
                 **disable_log_files** [bool] = ``False``:
                     Disable ``ensemble.log`` and ``libE_stats.txt`` log files.
 
+                **gen_workers** [list of ints]:
+                    List of workers that should only run generators. All other workers will only
+                    run simulator functions.
+
         .. tab-item:: Directories
 
             .. tab-set::

libensemble/alloc_funcs/fast_alloc.py

@@ -32,25 +32,26 @@ def give_sim_work_first(W, H, sim_specs, gen_specs, alloc_specs, persis_info, li
     Work = {}
     gen_in = gen_specs.get("in", [])
 
-    for wid in support.avail_worker_ids():
+    # Give sim work if possible
+    for wid in support.avail_worker_ids(gen_workers=False):
         persis_info = support.skip_canceled_points(H, persis_info)
-
-        # Give sim work if possible
         if persis_info["next_to_give"] < len(H):
             try:
                 Work[wid] = support.sim_work(wid, H, sim_specs["in"], [persis_info["next_to_give"]], [])
             except InsufficientFreeResources:
                 break
             persis_info["next_to_give"] += 1
 
-        elif gen_count < user.get("num_active_gens", gen_count + 1):
-            # Give gen work
-            return_rows = range(len(H)) if gen_in else []
-            try:
-                Work[wid] = support.gen_work(wid, gen_in, return_rows, persis_info.get(wid))
-            except InsufficientFreeResources:
-                break
-            gen_count += 1
-            persis_info["total_gen_calls"] += 1
+    # Give gen work if possible
+    if persis_info["next_to_give"] >= len(H):
+        for wid in support.avail_worker_ids(gen_workers=True):
+            if wid not in Work and gen_count < user.get("num_active_gens", gen_count + 1):
+                return_rows = range(len(H)) if gen_in else []
+                try:
+                    Work[wid] = support.gen_work(wid, gen_in, return_rows, persis_info.get(wid))
+                except InsufficientFreeResources:
+                    break
+                gen_count += 1
+                persis_info["total_gen_calls"] += 1
 
     return Work, persis_info

libensemble/alloc_funcs/fast_alloc_and_pausing.py

@@ -43,9 +43,10 @@ def give_sim_work_first(W, H, sim_specs, gen_specs, alloc_specs, persis_info, li
         for pt_id in persis_info["pt_ids"]:
             persis_info["inds_of_pt_ids"][pt_id] = H["pt_id"] == pt_id
 
-    idle_workers = support.avail_worker_ids()
+    idle_sim_workers = support.avail_worker_ids(gen_workers=False)
+    idle_gen_workers = support.avail_worker_ids(gen_workers=True)
 
-    while len(idle_workers):
+    while len(idle_sim_workers):
         pt_ids_to_pause = set()
 
         # Find indices of H that are not yet given out to be evaluated
@@ -106,15 +107,19 @@ def give_sim_work_first(W, H, sim_specs, gen_specs, alloc_specs, persis_info, li
 
             if len(persis_info["need_to_give"]) != 0:
                 next_row = persis_info["need_to_give"].pop()
-                i = idle_workers[0]
+                i = idle_sim_workers[0]
                 try:
                     Work[i] = support.sim_work(i, H, sim_specs["in"], [next_row], [])
                 except InsufficientFreeResources:
                     persis_info["need_to_give"].add(next_row)
                     break
-                idle_workers = idle_workers[1:]
+                idle_sim_workers = idle_sim_workers[1:]
 
-        elif gen_count < alloc_specs["user"].get("num_active_gens", gen_count + 1):
+        else:
+            break
+
+    while len(idle_gen_workers):
+        if gen_count < alloc_specs["user"].get("num_active_gens", gen_count + 1):
             lw = persis_info["last_worker"]
 
             last_size = persis_info.get("last_size")
@@ -126,18 +131,18 @@ def give_sim_work_first(W, H, sim_specs, gen_specs, alloc_specs, persis_info, li
                     break
 
             # Give gen work
-            i = idle_workers[0]
+            i = idle_gen_workers[0]
             try:
                 Work[i] = support.gen_work(i, gen_specs["in"], range(len(H)), persis_info[lw])
             except InsufficientFreeResources:
                 break
-            idle_workers = idle_workers[1:]
+            idle_gen_workers = idle_gen_workers[1:]
             gen_count += 1
             persis_info["total_gen_calls"] += 1
             persis_info["last_worker"] = i
             persis_info["last_size"] = len(H)
 
         elif gen_count >= alloc_specs["user"].get("num_active_gens", gen_count + 1):
-            idle_workers = []
+            idle_gen_workers = []
 
     return Work, persis_info

libensemble/alloc_funcs/give_pregenerated_work.py

@@ -23,7 +23,7 @@ def give_pregenerated_sim_work(W, H, sim_specs, gen_specs, alloc_specs, persis_i
     if persis_info["next_to_give"] >= len(H):
         return Work, persis_info, 1
 
-    for i in support.avail_worker_ids():
+    for i in support.avail_worker_ids(gen_workers=False):
         persis_info = support.skip_canceled_points(H, persis_info)
 
         # Give sim work

libensemble/alloc_funcs/give_sim_work_first.py

@@ -64,15 +64,19 @@ def give_sim_work_first(
     Work = {}
 
     points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"]
-    for wid in support.avail_worker_ids():
-        if np.any(points_to_evaluate):
+
+    if np.any(points_to_evaluate):
+        for wid in support.avail_worker_ids(gen_workers=False):
             sim_ids_to_send = support.points_by_priority(H, points_avail=points_to_evaluate, batch=batch_give)
             try:
                 Work[wid] = support.sim_work(wid, H, sim_specs["in"], sim_ids_to_send, persis_info.get(wid))
             except InsufficientFreeResources:
                 break
             points_to_evaluate[sim_ids_to_send] = False
-        else:
+            if not np.any(points_to_evaluate):
+                break
+    else:
+        for wid in support.avail_worker_ids(gen_workers=True):
             # Allow at most num_active_gens active generator instances
             if gen_count >= user.get("num_active_gens", gen_count + 1):
                 break

libensemble/alloc_funcs/inverse_bayes_allocf.py

@@ -42,8 +42,9 @@ def only_persistent_gens_for_inverse_bayes(W, H, sim_specs, gen_specs, alloc_spe
             Work[wid] = support.gen_work(wid, ["like"], inds_to_send_back, persis_info.get(wid), persistent=True)
 
     points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"]
-    for wid in support.avail_worker_ids(persistent=False):
-        if np.any(points_to_evaluate):
+    if np.any(points_to_evaluate):
+        for wid in support.avail_worker_ids(persistent=False, gen_workers=False):
+
             # perform sim evaluations (if any point hasn't been given).
             sim_subbatches = H["subbatch"][points_to_evaluate]
             sim_inds = sim_subbatches == np.min(sim_subbatches)
@@ -54,13 +55,11 @@ def only_persistent_gens_for_inverse_bayes(W, H, sim_specs, gen_specs, alloc_spe
             except InsufficientFreeResources:
                 break
             points_to_evaluate[sim_ids_to_send] = False
-
-        elif gen_count == 0:
-            # Finally, generate points since there is nothing else to do.
-            try:
-                Work[wid] = support.gen_work(wid, gen_specs["in"], [], persis_info.get(wid), persistent=True)
-            except InsufficientFreeResources:
+            if not np.any(points_to_evaluate):
                 break
-            gen_count += 1
+
+    elif gen_count == 0:
+        wid = support.avail_worker_ids(persistent=False, gen_workers=True)[0]
+        Work[wid] = support.gen_work(wid, gen_specs["in"], [], persis_info.get(wid), persistent=True)
 
     return Work, persis_info

libensemble/alloc_funcs/only_one_gen_alloc.py

@@ -21,27 +21,30 @@ def ensure_one_active_gen(W, H, sim_specs, gen_specs, alloc_specs, persis_info,
     gen_flag = True
     gen_in = gen_specs.get("in", [])
 
-    for wid in support.avail_worker_ids():
-        persis_info = support.skip_canceled_points(H, persis_info)
-
-        if persis_info["next_to_give"] < len(H):
+    if persis_info["next_to_give"] < len(H):
+        for wid in support.avail_worker_ids(gen_workers=False):
+            persis_info = support.skip_canceled_points(H, persis_info)
             try:
                 Work[wid] = support.sim_work(wid, H, sim_specs["in"], [persis_info["next_to_give"]], [])
             except InsufficientFreeResources:
                 break
             persis_info["next_to_give"] += 1
-
-        elif not support.test_any_gen() and gen_flag:
-            if not support.all_sim_ended(H):
+            if persis_info["next_to_give"] >= len(H):
                 break
 
-            # Give gen work
-            return_rows = range(len(H)) if gen_in else []
-            try:
-                Work[wid] = support.gen_work(wid, gen_in, return_rows, persis_info.get(wid))
-            except InsufficientFreeResources:
-                break
-            gen_flag = False
-            persis_info["total_gen_calls"] += 1
+    elif not support.test_any_gen() and gen_flag:
+        # Give gen work
+        return_rows = range(len(H)) if gen_in else []
+        wid = support.avail_worker_ids(gen_workers=True)[0]
+
+        if not support.all_sim_ended(H):
+            return Work, persis_info
+
+        try:
+            Work[wid] = support.gen_work(wid, gen_in, return_rows, persis_info.get(wid))
+        except InsufficientFreeResources:
+            return Work, persis_info
+        gen_flag = False
+        persis_info["total_gen_calls"] += 1
 
     return Work, persis_info

libensemble/alloc_funcs/persistent_aposmm_alloc.py

@@ -53,7 +53,7 @@ def persistent_aposmm_alloc(W, H, sim_specs, gen_specs, alloc_specs, persis_info
                 )
                 returned_but_not_given[point_ids] = False
 
-    for wid in support.avail_worker_ids(persistent=False):
+    for wid in support.avail_worker_ids(persistent=False, gen_workers=False):
         persis_info = support.skip_canceled_points(H, persis_info)
 
         if persis_info["next_to_give"] < len(H):
@@ -63,8 +63,11 @@ def persistent_aposmm_alloc(W, H, sim_specs, gen_specs, alloc_specs, persis_info
             except InsufficientFreeResources:
                 break
             persis_info["next_to_give"] += 1
+            if persis_info["next_to_give"] >= len(H):
+                break
 
-        elif persis_info.get("gen_started") is None:
+    if persis_info.get("gen_started") is None:
+        for wid in support.avail_worker_ids(persistent=False, gen_workers=True):
             # Finally, call a persistent generator as there is nothing else to do.
             persis_info.get(wid)["nworkers"] = len(W)
             try:
@@ -74,5 +77,6 @@ def persistent_aposmm_alloc(W, H, sim_specs, gen_specs, alloc_specs, persis_info
             except InsufficientFreeResources:
                 break
             persis_info["gen_started"] = True  # Must set after - in case break on resources
+            break
 
     return Work, persis_info

libensemble/alloc_funcs/start_fd_persistent.py

@@ -49,22 +49,21 @@ def finite_diff_alloc(W, H, sim_specs, gen_specs, alloc_specs, persis_info, libE
             )
 
     points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"]
-    for wid in support.avail_worker_ids(persistent=False):
-        if np.any(points_to_evaluate):
+    if np.any(points_to_evaluate):
+        for wid in support.avail_worker_ids(persistent=False, gen_workers=False):
             # perform sim evaluations (if they exist in History).
             sim_ids_to_send = np.nonzero(points_to_evaluate)[0][0]  # oldest point
             try:
                 Work[wid] = support.sim_work(wid, H, sim_specs["in"], sim_ids_to_send, persis_info.get(wid))
             except InsufficientFreeResources:
                 break
             points_to_evaluate[sim_ids_to_send] = False
-
-        elif gen_count == 0:
-            # Finally, call a persistent generator as there is nothing else to do.
-            try:
-                Work[wid] = support.gen_work(wid, gen_specs.get("in", []), [], persis_info.get(wid), persistent=True)
-            except InsufficientFreeResources:
+            if not np.any(points_to_evaluate):
                 break
-            gen_count += 1
+
+    if gen_count == 0:
+        wid = support.avail_worker_ids(persistent=False, gen_workers=True)[0]
+        Work[wid] = support.gen_work(wid, gen_specs.get("in", []), [], persis_info.get(wid), persistent=True)
+        gen_count += 1
 
     return Work, persis_info, 0

libensemble/alloc_funcs/start_only_persistent.py

@@ -89,7 +89,7 @@ def only_persistent_gens(W, H, sim_specs, gen_specs, alloc_specs, persis_info, l
 
     # Now the give_sim_work_first part
     points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"]
-    avail_workers = support.avail_worker_ids(persistent=False, zero_resource_workers=False)
+    avail_workers = support.avail_worker_ids(persistent=False, zero_resource_workers=False, gen_workers=False)
     if user.get("alt_type"):
         avail_workers = list(
             set(support.avail_worker_ids(persistent=False, zero_resource_workers=False))
@@ -115,7 +115,7 @@ def only_persistent_gens(W, H, sim_specs, gen_specs, alloc_specs, persis_info, l
 
     # Start persistent gens if no worker to give out. Uses zero_resource_workers if defined.
     if not np.any(points_to_evaluate):
-        avail_workers = support.avail_worker_ids(persistent=False, zero_resource_workers=True)
+        avail_workers = support.avail_worker_ids(persistent=False, zero_resource_workers=True, gen_workers=True)
 
         for wid in avail_workers:
             if gen_count < user.get("num_active_gens", 1):