Refactor/user function handling modules + Manager can run additional worker on thread

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):

libensemble/alloc_funcs/start_persistent_local_opt_gens.py

@@ -54,7 +54,7 @@ def start_persistent_local_opt_gens(W, H, sim_specs, gen_specs, alloc_specs, per
             Work[wid] = support.gen_work(wid, gen_specs["persis_in"], last_ind, persis_info[wid], persistent=True)
             persis_info[wid]["run_order"].append(last_ind)
 
-    for wid in support.avail_worker_ids(persistent=False):
+    for wid in support.avail_worker_ids(persistent=False, gen_workers=True):
         # Find candidates to start local opt runs if a sample has been evaluated
         if np.any(np.logical_and(~H["local_pt"], H["sim_ended"], ~H["cancel_requested"])):
             n = len(H["x"][0])
@@ -78,7 +78,8 @@ def start_persistent_local_opt_gens(W, H, sim_specs, gen_specs, alloc_specs, per
             persis_info[wid]["run_order"] = [ind]
             gen_count += 1
 
-        elif 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 from existing runs
             q_inds_logical = np.logical_and(points_to_evaluate, H["local_pt"])
             if not np.any(q_inds_logical):
@@ -89,10 +90,13 @@ def start_persistent_local_opt_gens(W, H, sim_specs, gen_specs, alloc_specs, per
             except InsufficientFreeResources:
                 break
             points_to_evaluate[sim_ids_to_send] = False
+            if not np.any(points_to_evaluate):
+                break
 
-        elif gen_count == 0 and not np.any(np.logical_and(W["active"] == EVAL_GEN_TAG, W["persis_state"] == 0)):
-            # Finally, generate points since there is nothing else to do (no resource sets req.)
-            Work[wid] = support.gen_work(wid, gen_specs.get("in", []), [], persis_info[wid], rset_team=[])
-            gen_count += 1
+    if gen_count == 0 and not np.any(np.logical_and(W["active"] == EVAL_GEN_TAG, W["persis_state"] == 0)):
+        # Finally, generate points since there is nothing else to do (no resource sets req.)
+        wid = support.avail_worker_ids(persistent=False, gen_workers=True)[0]
+        Work[wid] = support.gen_work(wid, gen_specs.get("in", []), [], persis_info[wid], rset_team=[])
+        gen_count += 1
 
     return Work, persis_info