From d4da77a2cd2ec15c6d7f4aa4222c6f57cf75a1d8 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 26 Jun 2025 17:00:58 -0500 Subject: [PATCH 01/38] initial commit, experimenting with pickling non-libE_field fields (x or f) out of a given update-index to file. --- libensemble/history.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/libensemble/history.py b/libensemble/history.py index 80f848cca..685170783 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -1,5 +1,7 @@ import logging +import pickle import time +from pathlib import Path import numpy as np import numpy.typing as npt @@ -123,6 +125,18 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: H_new[field][: len(self.H)] = self.H[field] self.H = H_new + def _shelf_longrunning_sims(self, cache_file, index): + """Cache any f values that ran for more than a second.""" + if 1: # self.H[index]['sim_ended_time'] - self.H[index]['sim_started_time'] > 1: + try: + cache = pickle.load(cache_file) + except EOFError: + cache = [] + entry = self.H[index] + presumptive_keys_to_cache = [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] + cache.append(entry[presumptive_keys_to_cache]) + pickle.dump(cache, cache_file) + def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: """ Updates the history after points have been evaluated @@ -135,6 +149,10 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: if returned_H is not None and any([field not in self.H.dtype.names for field in returned_H.dtype.names]): self._append_new_fields(returned_H) + cache_dir = Path.home() / ".libE" + cache_dir.mkdir(parents=True, exist_ok=True) + cache = open(cache_dir / "sims.pickle", "wb+") + for j, ind in enumerate(new_inds): for field in fields: if field in protected_libE_fields: @@ -159,6 +177,8 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: self.H["sim_ended"][ind] = True self.H["sim_ended_time"][ind] = time.time() self.sim_ended_count += 1 + self._shelf_longrunning_sims(cache, ind) + cache.close() if kill_canceled_sims: for j in range(self.last_ended + 1, np.max(new_inds) + 1): From 786c8b9d5c095ae4b70f1ac4d44380ff15c14301 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 28 Aug 2025 16:50:16 -0500 Subject: [PATCH 02/38] additional poking around and experimenting with history saving cache, and then seeing if values exist in that cache before sending sims... --- libensemble/history.py | 36 ++++++++++++++++++++++-------------- libensemble/manager.py | 8 ++++++++ 2 files changed, 30 insertions(+), 14 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 685170783..22ef41b0b 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -1,5 +1,4 @@ import logging -import pickle import time from pathlib import Path @@ -117,6 +116,11 @@ def __init__( self.last_started = -1 self.last_ended = -1 + self.cache_dir = Path.home() / ".libE" + self.cache_dir.mkdir(parents=True, exist_ok=True) + self.cache = open(self.cache_dir / "sims.pickle", "wb+") + self.cache_set = False + def _append_new_fields(self, H_f: npt.NDArray) -> None: dtype_new = np.dtype(list(set(self.H.dtype.descr + np.lib.recfunctions.repack_fields(H_f).dtype.descr))) H_new = np.zeros(len(self.H), dtype=dtype_new) @@ -125,17 +129,26 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: H_new[field][: len(self.H)] = self.H[field] self.H = H_new - def _shelf_longrunning_sims(self, cache_file, index): + def _shelf_longrunning_sims(self, index): """Cache any f values that ran for more than a second.""" if 1: # self.H[index]['sim_ended_time'] - self.H[index]['sim_started_time'] > 1: + presumptive_keys_to_cache = [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] + self.new_dtype_cache_keys = [(name, self.H.dtype.fields[name][0]) for name in presumptive_keys_to_cache] try: - cache = pickle.load(cache_file) + in_cache = np.load(self.cache, allow_pickle=True) except EOFError: - cache = [] - entry = self.H[index] - presumptive_keys_to_cache = [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] - cache.append(entry[presumptive_keys_to_cache]) - pickle.dump(cache, cache_file) + in_cache = np.zeros(1, dtype=self.new_dtype_cache_keys) + entry = self.H[index][presumptive_keys_to_cache] + in_cache = np.append(in_cache, entry) + np.save(self.cache, in_cache) + self.cache_set = True + + def get_shelved_sims(self) -> npt.NDArray: + try: + in_cache = np.load(self.cache, allow_pickle=True) + except EOFError: + in_cache = np.zeros(1, dtype=self.new_dtype_cache_keys) + return in_cache def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: """ @@ -149,10 +162,6 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: if returned_H is not None and any([field not in self.H.dtype.names for field in returned_H.dtype.names]): self._append_new_fields(returned_H) - cache_dir = Path.home() / ".libE" - cache_dir.mkdir(parents=True, exist_ok=True) - cache = open(cache_dir / "sims.pickle", "wb+") - for j, ind in enumerate(new_inds): for field in fields: if field in protected_libE_fields: @@ -177,8 +186,7 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: self.H["sim_ended"][ind] = True self.H["sim_ended_time"][ind] = time.time() self.sim_ended_count += 1 - self._shelf_longrunning_sims(cache, ind) - cache.close() + self._shelf_longrunning_sims(ind) if kill_canceled_sims: for j in range(self.last_ended + 1, np.max(new_inds) + 1): diff --git a/libensemble/manager.py b/libensemble/manager.py index 7995d2da9..ec0201731 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -437,6 +437,14 @@ def _send_work_order(self, Work: dict, w: int) -> None: for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) + if Work["tag"] == EVAL_SIM_TAG and self.hist.cache_set: + cached_H = self.hist.get_shelved_sims() + for entry in H_to_be_sent: + if np.allclose(entry[self.hist.new_dtype_cache_keys], cached_H, rtol=1e-8, atol=1e-8): + # probably figure out indexes for entries in H_to_be_sent that + # can simply be read back into History from cache? + pass + self.wcomms[w].send(0, H_to_be_sent) def _update_state_on_alloc(self, Work: dict, w: int): From 5ae2d723a356dd0e76a87a9fa0cb34396323ae62 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 29 Aug 2025 12:56:51 -0500 Subject: [PATCH 03/38] better making of .npy database, use History attributes created upon the cache being created. iterate over the cache_keys corresponding to those in gen_specs out, then check if they're close to a cache entry. grab those indexes so we can slot in the corresponding data into the manager's H later on --- libensemble/history.py | 22 +++++++++++----------- libensemble/manager.py | 24 +++++++++++++++++------- 2 files changed, 28 insertions(+), 18 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 22ef41b0b..5d01d91f7 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -118,7 +118,9 @@ def __init__( self.cache_dir = Path.home() / ".libE" self.cache_dir.mkdir(parents=True, exist_ok=True) - self.cache = open(self.cache_dir / "sims.pickle", "wb+") + self.cache = self.cache_dir / "cache.npy" + if not self.cache.exists(): + self.cache.touch() self.cache_set = False def _append_new_fields(self, H_f: npt.NDArray) -> None: @@ -132,22 +134,20 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: def _shelf_longrunning_sims(self, index): """Cache any f values that ran for more than a second.""" if 1: # self.H[index]['sim_ended_time'] - self.H[index]['sim_started_time'] > 1: - presumptive_keys_to_cache = [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] - self.new_dtype_cache_keys = [(name, self.H.dtype.fields[name][0]) for name in presumptive_keys_to_cache] + self.cache_keys = sorted([i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]]) + self.cache_dtype = sorted([(name, self.H.dtype.fields[name][0]) for name in self.cache_keys]) try: in_cache = np.load(self.cache, allow_pickle=True) except EOFError: - in_cache = np.zeros(1, dtype=self.new_dtype_cache_keys) - entry = self.H[index][presumptive_keys_to_cache] - in_cache = np.append(in_cache, entry) - np.save(self.cache, in_cache) + in_cache = np.zeros(1, dtype=self.cache_dtype) + entry = self.H[index][self.cache_keys] + if entry not in in_cache: + in_cache = np.append(in_cache, entry) + np.save(self.cache, in_cache, allow_pickle=True) self.cache_set = True def get_shelved_sims(self) -> npt.NDArray: - try: - in_cache = np.load(self.cache, allow_pickle=True) - except EOFError: - in_cache = np.zeros(1, dtype=self.new_dtype_cache_keys) + in_cache = np.load(self.cache, allow_pickle=True) return in_cache def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: diff --git a/libensemble/manager.py b/libensemble/manager.py index ec0201731..eb676589d 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -416,7 +416,7 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: if "sim_id" not in self.gen_specs["persis_in"]: self.gen_specs["persis_in"].append("sim_id") - def _send_work_order(self, Work: dict, w: int) -> None: + def _send_work_order(self, Work: dict, w: int) -> list: """Sends an allocation function order to a worker""" logger.debug(f"Manager sending work unit to worker {w}") @@ -437,15 +437,24 @@ def _send_work_order(self, Work: dict, w: int) -> None: for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) + # check if any of the generated points are already in the cache if Work["tag"] == EVAL_SIM_TAG and self.hist.cache_set: cached_H = self.hist.get_shelved_sims() - for entry in H_to_be_sent: - if np.allclose(entry[self.hist.new_dtype_cache_keys], cached_H, rtol=1e-8, atol=1e-8): - # probably figure out indexes for entries in H_to_be_sent that - # can simply be read back into History from cache? - pass + gen_keys = [j[0] for j in self.gen_specs["out"]] + cache_gen_keys = [i for i in self.hist.cache_keys if i in gen_keys] + discovered_cache_indexes = [] + for index, entry in enumerate(H_to_be_sent): + for field in cache_gen_keys: + if np.allclose(entry[field], cached_H[field], rtol=1e-8, atol=1e-8): + discovered_cache_indexes.append(index) + break + if len(discovered_cache_indexes) > 0: + for index in discovered_cache_indexes: + H_to_be_sent = np.delete(H_to_be_sent, index, axis=0) + return discovered_cache_indexes self.wcomms[w].send(0, H_to_be_sent) + return [] def _update_state_on_alloc(self, Work: dict, w: int): """Updates a workers' active/idle status following an allocation order""" @@ -720,7 +729,8 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: if self._sim_max_given(): break self._check_work_order(Work[w], w) - self._send_work_order(Work[w], w) + cache_indexes = self._send_work_order(Work[w], w) + print(cache_indexes) self._update_state_on_alloc(Work[w], w) assert self.term_test() or any( self.W["active"] != 0 From 8ad5f1beec160b32337bac31d5672ea57789e345 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 29 Aug 2025 12:59:31 -0500 Subject: [PATCH 04/38] little note...? --- libensemble/manager.py | 1 + 1 file changed, 1 insertion(+) diff --git a/libensemble/manager.py b/libensemble/manager.py index eb676589d..35fe70fa1 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -730,6 +730,7 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: break self._check_work_order(Work[w], w) cache_indexes = self._send_work_order(Work[w], w) + # JLN TODO: take these indexes, grab the data from cache, slot into history, then what...? print(cache_indexes) self._update_state_on_alloc(Work[w], w) assert self.term_test() or any( From 1ce85068e7b4a5b05b6525e0bca2969643791f85 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 29 Aug 2025 14:21:27 -0500 Subject: [PATCH 05/38] comments --- libensemble/history.py | 9 +++++++-- libensemble/manager.py | 25 ++++++++++++++++--------- 2 files changed, 23 insertions(+), 11 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 5d01d91f7..b5c3fd0a0 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -134,8 +134,13 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: def _shelf_longrunning_sims(self, index): """Cache any f values that ran for more than a second.""" if 1: # self.H[index]['sim_ended_time'] - self.H[index]['sim_started_time'] > 1: - self.cache_keys = sorted([i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]]) - self.cache_dtype = sorted([(name, self.H.dtype.fields[name][0]) for name in self.cache_keys]) + # ('f', 'x') and ('x', 'f') are not equivalent dtypes, unfortunately. So maybe sorted helps. + self.cache_keys = sorted( + [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] + ) # ('f', 'x') keys only + self.cache_dtype = sorted( + [(name, self.H.dtype.fields[name][0]) for name in self.cache_keys] + ) # only needed to init cache try: in_cache = np.load(self.cache, allow_pickle=True) except EOFError: diff --git a/libensemble/manager.py b/libensemble/manager.py index 35fe70fa1..4140cd854 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -431,6 +431,8 @@ def _send_work_order(self, Work: dict, w: int) -> list: work_rows = Work["libE_info"]["H_rows"] work_name = calc_type_strings[Work["tag"]] logger.debug(f"Manager sending {work_name} work to worker {w}. Rows {extract_H_ranges(Work) or None}") + + discovered_cache_indexes = [] if len(work_rows): new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) @@ -439,22 +441,27 @@ def _send_work_order(self, Work: dict, w: int) -> list: # check if any of the generated points are already in the cache if Work["tag"] == EVAL_SIM_TAG and self.hist.cache_set: - cached_H = self.hist.get_shelved_sims() - gen_keys = [j[0] for j in self.gen_specs["out"]] - cache_gen_keys = [i for i in self.hist.cache_keys if i in gen_keys] + cached_H = self.hist.get_shelved_sims() # get the cache + gen_keys = [j[0] for j in self.gen_specs["out"]] # get 'x' keys + cache_gen_keys = [i for i in self.hist.cache_keys if i in gen_keys] # get 'x' keys in cache discovered_cache_indexes = [] - for index, entry in enumerate(H_to_be_sent): + for index, entry in enumerate( + H_to_be_sent + ): # find indexes of H_to_be_sent where 'x' fields are in cache for field in cache_gen_keys: - if np.allclose(entry[field], cached_H[field], rtol=1e-8, atol=1e-8): + if np.allclose( + entry[field], cached_H[field], rtol=1e-8, atol=1e-8 + ): # iterate through cache entries too? discovered_cache_indexes.append(index) - break + break # but maybe the other 'x' fields are also in the cache...? if len(discovered_cache_indexes) > 0: for index in discovered_cache_indexes: - H_to_be_sent = np.delete(H_to_be_sent, index, axis=0) - return discovered_cache_indexes + H_to_be_sent = np.delete( + H_to_be_sent, index, axis=0 + ) # delete rows from H_to_be_sent. we already have f self.wcomms[w].send(0, H_to_be_sent) - return [] + return discovered_cache_indexes def _update_state_on_alloc(self, Work: dict, w: int): """Updates a workers' active/idle status following an allocation order""" From 1408c7698e34c54a16cc83e5bef10e0cf905149d Mon Sep 17 00:00:00 2001 From: jlnav Date: Tue, 8 Sep 2026 10:07:16 -0500 Subject: [PATCH 06/38] experimenting with having caching being a step of the alloc, once we've determined points_to_evaluate --- .../alloc_funcs/give_sim_work_first.py | 14 ++++++++ libensemble/manager.py | 34 ++++--------------- 2 files changed, 20 insertions(+), 28 deletions(-) diff --git a/libensemble/alloc_funcs/give_sim_work_first.py b/libensemble/alloc_funcs/give_sim_work_first.py index 96245f7a9..05e720c96 100644 --- a/libensemble/alloc_funcs/give_sim_work_first.py +++ b/libensemble/alloc_funcs/give_sim_work_first.py @@ -64,6 +64,20 @@ def give_sim_work_first( points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] + if len(libE_info["cache"]) and np.any(points_to_evaluate): + for H_index, H_entry in enumerate(H): + for cache_index, cache_entry in enumerate(libE_info["cache"]): + for field in [j[0] for j in gen_specs["out"]]: + if field in libE_info["cache"].dtype.names and np.allclose( + H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 + ): + H[H_index][field] = cache_entry[field] + libE_info["hist"].update_history_x_out(q_inds=np.array([H_index]), sim_worker=1) + libE_info["hist"].update_history_to_gen(q_inds=np.array([H_index])) + print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") + + points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] + 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) diff --git a/libensemble/manager.py b/libensemble/manager.py index 4140cd854..3ca0b8296 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -416,7 +416,7 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: if "sim_id" not in self.gen_specs["persis_in"]: self.gen_specs["persis_in"].append("sim_id") - def _send_work_order(self, Work: dict, w: int) -> list: + def _send_work_order(self, Work: dict, w: int) -> None: """Sends an allocation function order to a worker""" logger.debug(f"Manager sending work unit to worker {w}") @@ -431,37 +431,13 @@ def _send_work_order(self, Work: dict, w: int) -> list: work_rows = Work["libE_info"]["H_rows"] work_name = calc_type_strings[Work["tag"]] logger.debug(f"Manager sending {work_name} work to worker {w}. Rows {extract_H_ranges(Work) or None}") - - discovered_cache_indexes = [] if len(work_rows): new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) - # check if any of the generated points are already in the cache - if Work["tag"] == EVAL_SIM_TAG and self.hist.cache_set: - cached_H = self.hist.get_shelved_sims() # get the cache - gen_keys = [j[0] for j in self.gen_specs["out"]] # get 'x' keys - cache_gen_keys = [i for i in self.hist.cache_keys if i in gen_keys] # get 'x' keys in cache - discovered_cache_indexes = [] - for index, entry in enumerate( - H_to_be_sent - ): # find indexes of H_to_be_sent where 'x' fields are in cache - for field in cache_gen_keys: - if np.allclose( - entry[field], cached_H[field], rtol=1e-8, atol=1e-8 - ): # iterate through cache entries too? - discovered_cache_indexes.append(index) - break # but maybe the other 'x' fields are also in the cache...? - if len(discovered_cache_indexes) > 0: - for index in discovered_cache_indexes: - H_to_be_sent = np.delete( - H_to_be_sent, index, axis=0 - ) # delete rows from H_to_be_sent. we already have f - self.wcomms[w].send(0, H_to_be_sent) - return discovered_cache_indexes def _update_state_on_alloc(self, Work: dict, w: int): """Updates a workers' active/idle status following an allocation order""" @@ -668,6 +644,8 @@ def _sim_max_given(self) -> bool: def _get_alloc_libE_info(self) -> dict: """Selected statistics useful for alloc_f""" + cache = self.hist.get_shelved_sims() if self.hist.cache_set else [] + return { "any_idle_workers": any(self.W["active"] == 0), "exit_criteria": self.exit_criteria, @@ -682,6 +660,8 @@ def _get_alloc_libE_info(self) -> dict: "gen_num_procs": self.gen_num_procs, "gen_num_gpus": self.gen_num_gpus, "gen_on_worker": self.libE_specs.get("gen_on_worker", False), + "cache": cache, + "hist": self.hist, } def _alloc_work(self, H: npt.NDArray, persis_info: dict) -> dict: @@ -736,9 +716,7 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: if self._sim_max_given(): break self._check_work_order(Work[w], w) - cache_indexes = self._send_work_order(Work[w], w) - # JLN TODO: take these indexes, grab the data from cache, slot into history, then what...? - print(cache_indexes) + self._send_work_order(Work[w], w) self._update_state_on_alloc(Work[w], w) assert self.term_test() or any( self.W["active"] != 0 From 725bb1e4d0841fa8dd0abeb235204d26f767b94b Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 11 Sep 2025 16:41:24 -0500 Subject: [PATCH 07/38] set those H entries to sim_started? --- libensemble/alloc_funcs/give_sim_work_first.py | 1 + libensemble/history.py | 1 + 2 files changed, 2 insertions(+) diff --git a/libensemble/alloc_funcs/give_sim_work_first.py b/libensemble/alloc_funcs/give_sim_work_first.py index 05e720c96..62e4c9218 100644 --- a/libensemble/alloc_funcs/give_sim_work_first.py +++ b/libensemble/alloc_funcs/give_sim_work_first.py @@ -72,6 +72,7 @@ def give_sim_work_first( H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 ): H[H_index][field] = cache_entry[field] + H[H_index]["sim_started"] = True libE_info["hist"].update_history_x_out(q_inds=np.array([H_index]), sim_worker=1) libE_info["hist"].update_history_to_gen(q_inds=np.array([H_index])) print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") diff --git a/libensemble/history.py b/libensemble/history.py index b5c3fd0a0..dcff4845d 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -148,6 +148,7 @@ def _shelf_longrunning_sims(self, index): entry = self.H[index][self.cache_keys] if entry not in in_cache: in_cache = np.append(in_cache, entry) + in_cache = np.unique(in_cache, axis=0) np.save(self.cache, in_cache, allow_pickle=True) self.cache_set = True From f342e63d0b9efcd51b1943a57baddbae4118de87 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 12 Sep 2025 15:42:25 -0500 Subject: [PATCH 08/38] grab update-able indexes, then call update_history_x_out and update_history_f on/with those indexes and the associated cache values, pretending that the array of cache-values are a worker message --- .../alloc_funcs/give_sim_work_first.py | 39 ++++++++++++++++--- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/libensemble/alloc_funcs/give_sim_work_first.py b/libensemble/alloc_funcs/give_sim_work_first.py index 62e4c9218..44a846476 100644 --- a/libensemble/alloc_funcs/give_sim_work_first.py +++ b/libensemble/alloc_funcs/give_sim_work_first.py @@ -62,21 +62,50 @@ def give_sim_work_first( gen_count = support.count_gens() Work = {} + if not persis_info.get("updated_H_indices"): + persis_info["updated_H_indices"] = [] + points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] + gen_out_fields = [j[0] for j in gen_specs["out"]] + + indices_to_update = [] + cache_hit = False + if len(libE_info["cache"]) and np.any(points_to_evaluate): for H_index, H_entry in enumerate(H): + if H_index in persis_info["updated_H_indices"]: + continue for cache_index, cache_entry in enumerate(libE_info["cache"]): - for field in [j[0] for j in gen_specs["out"]]: + for field in gen_out_fields: if field in libE_info["cache"].dtype.names and np.allclose( H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 ): - H[H_index][field] = cache_entry[field] - H[H_index]["sim_started"] = True - libE_info["hist"].update_history_x_out(q_inds=np.array([H_index]), sim_worker=1) - libE_info["hist"].update_history_to_gen(q_inds=np.array([H_index])) + cache_hit = True + indices_to_update.append({"cache_index": cache_index, "H_index": H_index, "field": field}) + persis_info["updated_H_indices"].append(H_index) print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") + if cache_hit: + + q_inds = np.array([i["H_index"] for i in indices_to_update]) + libE_info["hist"].update_history_x_out(q_inds=q_inds, sim_worker=1) + + simulated_calc_out = np.zeros(len(q_inds), dtype=sim_specs["out"]) + for i, H_index in enumerate(q_inds): + simulated_calc_out[i] = libE_info["cache"][indices_to_update[i]["cache_index"]][ + indices_to_update[i]["field"] + ] + + simulated_D_recv = { + "calc_out": simulated_calc_out, + "libE_info": { + "H_rows": q_inds, + }, + } + + libE_info["hist"].update_history_f(simulated_D_recv, kill_canceled_sims=False) + points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] if np.any(points_to_evaluate): From 9fcad4c631e07bbd79a65360961884adcda50362 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 17 Sep 2025 15:47:51 -0500 Subject: [PATCH 09/38] moving cache logic into manager, into handle_msg_from_worker that overrides .recv, and in _send_work_order that forms a buffer from corresponding cache entries --- .../alloc_funcs/give_sim_work_first.py | 86 +++++++++---------- libensemble/manager.py | 67 +++++++++++++-- pyproject.toml | 11 +++ 3 files changed, 113 insertions(+), 51 deletions(-) diff --git a/libensemble/alloc_funcs/give_sim_work_first.py b/libensemble/alloc_funcs/give_sim_work_first.py index 44a846476..0aaf76acf 100644 --- a/libensemble/alloc_funcs/give_sim_work_first.py +++ b/libensemble/alloc_funcs/give_sim_work_first.py @@ -62,49 +62,49 @@ def give_sim_work_first( gen_count = support.count_gens() Work = {} - if not persis_info.get("updated_H_indices"): - persis_info["updated_H_indices"] = [] - - points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] - - gen_out_fields = [j[0] for j in gen_specs["out"]] - - indices_to_update = [] - cache_hit = False - - if len(libE_info["cache"]) and np.any(points_to_evaluate): - for H_index, H_entry in enumerate(H): - if H_index in persis_info["updated_H_indices"]: - continue - for cache_index, cache_entry in enumerate(libE_info["cache"]): - for field in gen_out_fields: - if field in libE_info["cache"].dtype.names and np.allclose( - H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 - ): - cache_hit = True - indices_to_update.append({"cache_index": cache_index, "H_index": H_index, "field": field}) - persis_info["updated_H_indices"].append(H_index) - print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") - - if cache_hit: - - q_inds = np.array([i["H_index"] for i in indices_to_update]) - libE_info["hist"].update_history_x_out(q_inds=q_inds, sim_worker=1) - - simulated_calc_out = np.zeros(len(q_inds), dtype=sim_specs["out"]) - for i, H_index in enumerate(q_inds): - simulated_calc_out[i] = libE_info["cache"][indices_to_update[i]["cache_index"]][ - indices_to_update[i]["field"] - ] - - simulated_D_recv = { - "calc_out": simulated_calc_out, - "libE_info": { - "H_rows": q_inds, - }, - } - - libE_info["hist"].update_history_f(simulated_D_recv, kill_canceled_sims=False) + # if not persis_info.get("updated_H_indices"): + # persis_info["updated_H_indices"] = [] + + # points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] + + # gen_out_fields = [j[0] for j in gen_specs["out"]] + + # indices_to_update = [] + # cache_hit = False + + # if len(libE_info["cache"]) and np.any(points_to_evaluate): + # for H_index, H_entry in enumerate(H): + # if H_index in persis_info["updated_H_indices"]: + # continue + # for cache_index, cache_entry in enumerate(libE_info["cache"]): + # for field in gen_out_fields: + # if field in libE_info["cache"].dtype.names and np.allclose( + # H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 + # ): + # cache_hit = True + # indices_to_update.append({"cache_index": cache_index, "H_index": H_index, "field": field}) + # persis_info["updated_H_indices"].append(H_index) + # print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") + + # if cache_hit: + + # q_inds = np.array([i["H_index"] for i in indices_to_update]) + # libE_info["hist"].update_history_x_out(q_inds=q_inds, sim_worker=1) + + # simulated_calc_out = np.zeros(len(q_inds), dtype=sim_specs["out"]) + # for i, H_index in enumerate(q_inds): + # simulated_calc_out[i] = libE_info["cache"][indices_to_update[i]["cache_index"]][ + # indices_to_update[i]["field"] + # ] + + # simulated_D_recv = { + # "calc_out": simulated_calc_out, + # "libE_info": { + # "H_rows": q_inds, + # }, + # } + + # libE_info["hist"].update_history_f(simulated_D_recv, kill_canceled_sims=False) points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] diff --git a/libensemble/manager.py b/libensemble/manager.py index 3ca0b8296..6c7d1d20f 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -217,6 +217,8 @@ def __init__( self.WorkerExc = False self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") + self.from_cache = [] + self.cache_hit = False dyn_keys = ("resource_sets", "num_procs", "num_gpus") dyn_keys_in_H = any(k in self.hist.H.dtype.names for k in dyn_keys) @@ -420,6 +422,32 @@ def _send_work_order(self, Work: dict, w: int) -> None: """Sends an allocation function order to a worker""" logger.debug(f"Manager sending work unit to worker {w}") + work_rows = Work["libE_info"]["H_rows"] + new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] + + if Work["tag"] == EVAL_SIM_TAG: + cache = self.hist.get_shelved_sims() + dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int)]).descr) + self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache + + for field in np.dtype(new_dtype).names: + if not len(self.from_cache): + break + if field in cache.dtype.names: + for row in work_rows: + for cache_row in cache: + if np.allclose( + cache_row[field], self.hist.H[field][row] + ): # we found outbound work in cache + self.cache_hit = True + from_cache_entry = np.empty( + 1, dtype=dtype_with_idx + ) # make an entry for this row, plus H_row + from_cache_entry["H_row"] = row + for remaining_field in cache.dtype.names: + from_cache_entry[remaining_field] = cache_row[remaining_field] + self.from_cache[row] = from_cache_entry + if self.resources: self._set_resources(Work, w) @@ -428,11 +456,19 @@ def _send_work_order(self, Work: dict, w: int) -> None: if Work["tag"] == EVAL_GEN_TAG: self.W[w]["gen_started_time"] = time.time() - work_rows = Work["libE_info"]["H_rows"] work_name = calc_type_strings[Work["tag"]] - logger.debug(f"Manager sending {work_name} work to worker {w}. Rows {extract_H_ranges(Work) or None}") + if self.cache_hit: + logger.debug( + f"Manager retrieved {work_name} work for worker {w} from cache. Rows {extract_H_ranges(Work) or None}" + ) + else: + logger.debug(f"Manager sending {work_name} work to worker {w}. Rows {extract_H_ranges(Work) or None}") + if len(work_rows): - new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] + + if all([i in self.from_cache["H_row"] for i in work_rows]): # if all rows in work_rows are found in cache + return + H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) @@ -468,6 +504,11 @@ def _receive_from_workers(self, persis_info: dict) -> dict: new_stuff = True while new_stuff: new_stuff = False + if self.cache_hit or len(self.from_cache): + self.cache_hit = False + new_stuff = True + self._handle_msg_from_worker(persis_info, 0, process_cache=True) + self.from_cache = [] for w in self.W["worker_id"]: if self.wcomms[w].mail_flag(): new_stuff = True @@ -533,11 +574,19 @@ def _update_state_on_worker_msg(self, persis_info: dict, D_recv: dict, w: int) - if D_recv.get("persis_info"): persis_info.setdefault(int(w), {}).update(D_recv["persis_info"]) - def _handle_msg_from_worker(self, persis_info: dict, w: int) -> None: + def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool = False) -> None: """Handles a message from worker w""" try: - msg = self.wcomms[w].recv() - tag, D_recv = msg + if process_cache: + D_recv = { + "calc_out": self.from_cache, # need cache entries without H_row + "libE_info": { + "H_rows": self.from_cache["H_row"], + }, + } + else: + msg = self.wcomms[w].recv() + tag, D_recv = msg except CommFinishedException: logger.debug(f"Finalizing message from Worker {w}") return @@ -644,8 +693,6 @@ def _sim_max_given(self) -> bool: def _get_alloc_libE_info(self) -> dict: """Selected statistics useful for alloc_f""" - cache = self.hist.get_shelved_sims() if self.hist.cache_set else [] - return { "any_idle_workers": any(self.W["active"] == 0), "exit_criteria": self.exit_criteria, @@ -659,9 +706,13 @@ def _get_alloc_libE_info(self) -> dict: "use_resource_sets": self.use_resource_sets, "gen_num_procs": self.gen_num_procs, "gen_num_gpus": self.gen_num_gpus, +<<<<<<< HEAD "gen_on_worker": self.libE_specs.get("gen_on_worker", False), "cache": cache, "hist": self.hist, +======= + "gen_on_manager": self.libE_specs.get("gen_on_manager", False), +>>>>>>> df29125a9 (moving cache logic into manager, into handle_msg_from_worker that overrides .recv, and in _send_work_order that forms a buffer from corresponding cache entries) } def _alloc_work(self, H: npt.NDArray, persis_info: dict) -> dict: diff --git a/pyproject.toml b/pyproject.toml index 257bcb15c..f7d1b9b3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -180,12 +180,23 @@ globus-compute-sdk = ">=4.10.2,<5" # Dependencies for libEnsemble [tool.pixi.dependencies] +<<<<<<< HEAD python = ">=3.11,<3.15" pip = ">=25.2,<26" setuptools = ">=80.8.0,<81" numpy = ">=2.2.6,<3" pydantic = ">=2.12.4,<3" gest-api = ">=0.1,<0.2" +======= +python = ">=3.10,<3.14" +pip = "*" +setuptools = "*" +numpy = "*" +pydantic = "*" +pyyaml = "*" +tomli = "*" +psutil = "*" +>>>>>>> df29125a9 (moving cache logic into manager, into handle_msg_from_worker that overrides .recv, and in _send_work_order that forms a buffer from corresponding cache entries) # macOS dependencies From 69918b311c9422b5d198121e6c211c4fab31ad12 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 18 Sep 2025 10:34:05 -0500 Subject: [PATCH 10/38] grow the manager's internal record of cache hits, instead of overwriting the same one --- libensemble/manager.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 6c7d1d20f..0f2c67c52 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -425,14 +425,15 @@ def _send_work_order(self, Work: dict, w: int) -> None: work_rows = Work["libE_info"]["H_rows"] new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] - if Work["tag"] == EVAL_SIM_TAG: + if Work["tag"] == EVAL_SIM_TAG and len(work_rows): cache = self.hist.get_shelved_sims() dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int)]).descr) - self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache + if not len(self.from_cache): + self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache + else: + self.from_cache = np.append(self.from_cache, np.zeros(len(work_rows), dtype=dtype_with_idx)) for field in np.dtype(new_dtype).names: - if not len(self.from_cache): - break if field in cache.dtype.names: for row in work_rows: for cache_row in cache: @@ -469,6 +470,9 @@ def _send_work_order(self, Work: dict, w: int) -> None: if all([i in self.from_cache["H_row"] for i in work_rows]): # if all rows in work_rows are found in cache return + if self.cache_hit: + work_rows = [row for row in work_rows if row not in self.from_cache["H_row"]] + H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) From 96399bc998c51b47d77bd09063b4f8eb6950a942 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 18 Sep 2025 16:24:28 -0500 Subject: [PATCH 11/38] save presumptive workerID for the worker that would've been given cached work. increment a cache_index for slotting into local buffer --- libensemble/manager.py | 40 +++++++++++++++++++++++++--------------- pyproject.toml | 11 ----------- 2 files changed, 25 insertions(+), 26 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 0f2c67c52..c5df4fad6 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -218,6 +218,7 @@ def __init__( self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") self.from_cache = [] + self.cache_index = 0 self.cache_hit = False dyn_keys = ("resource_sets", "num_procs", "num_gpus") @@ -425,9 +426,9 @@ def _send_work_order(self, Work: dict, w: int) -> None: work_rows = Work["libE_info"]["H_rows"] new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] - if Work["tag"] == EVAL_SIM_TAG and len(work_rows): + if Work["tag"] == EVAL_SIM_TAG and len(work_rows) and self.hist.cache_set: cache = self.hist.get_shelved_sims() - dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int)]).descr) + dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int), ("worker_id", int)]).descr) if not len(self.from_cache): self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache else: @@ -445,17 +446,18 @@ def _send_work_order(self, Work: dict, w: int) -> None: 1, dtype=dtype_with_idx ) # make an entry for this row, plus H_row from_cache_entry["H_row"] = row + from_cache_entry["worker_id"] = w for remaining_field in cache.dtype.names: from_cache_entry[remaining_field] = cache_row[remaining_field] - self.from_cache[row] = from_cache_entry + self.from_cache[self.cache_index] = from_cache_entry + self.cache_index += 1 if self.resources: self._set_resources(Work, w) - self.wcomms[w].send(Work["tag"], Work) - - if Work["tag"] == EVAL_GEN_TAG: + elif Work["tag"] == EVAL_GEN_TAG: self.W[w]["gen_started_time"] = time.time() + self.wcomms[w].send(Work["tag"], Work) work_name = calc_type_strings[Work["tag"]] if self.cache_hit: @@ -467,16 +469,18 @@ def _send_work_order(self, Work: dict, w: int) -> None: if len(work_rows): - if all([i in self.from_cache["H_row"] for i in work_rows]): # if all rows in work_rows are found in cache - return - if self.cache_hit: work_rows = [row for row in work_rows if row not in self.from_cache["H_row"]] + if all( + [i in self.from_cache["H_row"] for i in work_rows] + ): # if all rows in work_rows are found in cache + return H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) + self.wcomms[w].send(Work["tag"], Work) self.wcomms[w].send(0, H_to_be_sent) def _update_state_on_alloc(self, Work: dict, w: int): @@ -505,14 +509,17 @@ def _receive_from_workers(self, persis_info: dict) -> dict: looped back over. """ time.sleep(0.0001) # Critical for multiprocessing performance + + if self.cache_hit or len(self.from_cache): + self.cache_hit = False + for w in self.from_cache["worker_id"]: + self._handle_msg_from_worker(persis_info, w, process_cache=True) + self.from_cache = [] + self.cache_index = 0 + new_stuff = True while new_stuff: new_stuff = False - if self.cache_hit or len(self.from_cache): - self.cache_hit = False - new_stuff = True - self._handle_msg_from_worker(persis_info, 0, process_cache=True) - self.from_cache = [] for w in self.W["worker_id"]: if self.wcomms[w].mail_flag(): new_stuff = True @@ -583,10 +590,13 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool try: if process_cache: D_recv = { - "calc_out": self.from_cache, # need cache entries without H_row + "calc_out": self.from_cache[[name[0] for name in self.sim_specs["out"]]], "libE_info": { "H_rows": self.from_cache["H_row"], + "workerID": w, }, + "calc_status": 0, + "calc_type": 1, } else: msg = self.wcomms[w].recv() diff --git a/pyproject.toml b/pyproject.toml index f7d1b9b3c..257bcb15c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -180,23 +180,12 @@ globus-compute-sdk = ">=4.10.2,<5" # Dependencies for libEnsemble [tool.pixi.dependencies] -<<<<<<< HEAD python = ">=3.11,<3.15" pip = ">=25.2,<26" setuptools = ">=80.8.0,<81" numpy = ">=2.2.6,<3" pydantic = ">=2.12.4,<3" gest-api = ">=0.1,<0.2" -======= -python = ">=3.10,<3.14" -pip = "*" -setuptools = "*" -numpy = "*" -pydantic = "*" -pyyaml = "*" -tomli = "*" -psutil = "*" ->>>>>>> df29125a9 (moving cache logic into manager, into handle_msg_from_worker that overrides .recv, and in _send_work_order that forms a buffer from corresponding cache entries) # macOS dependencies From 2f61be720e60819925e11ba60f1b1cc8422d988c Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 19 Sep 2025 13:27:22 -0500 Subject: [PATCH 12/38] prevent redundant insertions into local cache retrieval. fix a bug involving gen hang. only if all sim work already in cache do we skip the send-stage. when processing cache entries, process them in the order in which they presumptively would've been received from that worker. additional logging. --- .../references/results_metadata.md | 1 - .pre-commit-config.yaml | 2 ++ libensemble/gen_classes/sampling.py | 4 +-- libensemble/manager.py | 36 ++++++++++--------- libensemble/tools/test_support.py | 6 ++-- pyproject.toml | 11 +++++- 6 files changed, 35 insertions(+), 25 deletions(-) diff --git a/.claude/skills/generate-scripts/references/results_metadata.md b/.claude/skills/generate-scripts/references/results_metadata.md index 97a71c14b..2a35a095e 100644 --- a/.claude/skills/generate-scripts/references/results_metadata.md +++ b/.claude/skills/generate-scripts/references/results_metadata.md @@ -39,4 +39,3 @@ If the minimum objective value is exactly 0.0, check whether those rows have `sim_ended == True`. Unevaluated rows often have fields initialized to zero. This is common for the last few rows when the simulation budget is exhausted — they were allocated by the generator but never evaluated. - diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 69c11918b..3df486faa 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -37,4 +37,6 @@ repos: rev: v1.19.1 hooks: - id: mypy + pass_filenames: false + args: [--package=libensemble.utils] exclude: ^docs/conf\.py$|libensemble/utils/(launcher|loc_stack|runners|pydantic|output_directory)\.py$|libensemble/tests/(regression_tests|functionality_tests|unit_tests|scaling_tests)/.* diff --git a/libensemble/gen_classes/sampling.py b/libensemble/gen_classes/sampling.py index 0b0662448..ff097105a 100644 --- a/libensemble/gen_classes/sampling.py +++ b/libensemble/gen_classes/sampling.py @@ -156,9 +156,7 @@ class UniformSampleWithVariableResources(LibensembleGenerator): path was tested with the default alloc. """ - def __init__( - self, vocs: VOCS, max_resource_sets: int, random_seed: int = 1, *args, **kwargs - ): + def __init__(self, vocs: VOCS, max_resource_sets: int, random_seed: int = 1, *args, **kwargs): super().__init__(vocs, *args, **kwargs) self.rng = np.random.default_rng(random_seed) self.max_rsets = max_resource_sets diff --git a/libensemble/manager.py b/libensemble/manager.py index c5df4fad6..b7083683c 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -438,13 +438,12 @@ def _send_work_order(self, Work: dict, w: int) -> None: if field in cache.dtype.names: for row in work_rows: for cache_row in cache: - if np.allclose( - cache_row[field], self.hist.H[field][row] - ): # we found outbound work in cache + if ( + np.allclose(cache_row[field], self.hist.H[field][row]) + and row not in self.from_cache["H_row"] + ): # we found outbound work in cache, that's not already been retrieved self.cache_hit = True - from_cache_entry = np.empty( - 1, dtype=dtype_with_idx - ) # make an entry for this row, plus H_row + from_cache_entry = np.empty(1, dtype=dtype_with_idx) from_cache_entry["H_row"] = row from_cache_entry["worker_id"] = w for remaining_field in cache.dtype.names: @@ -455,7 +454,7 @@ def _send_work_order(self, Work: dict, w: int) -> None: if self.resources: self._set_resources(Work, w) - elif Work["tag"] == EVAL_GEN_TAG: + if Work["tag"] == EVAL_GEN_TAG: self.W[w]["gen_started_time"] = time.time() self.wcomms[w].send(Work["tag"], Work) @@ -471,8 +470,8 @@ def _send_work_order(self, Work: dict, w: int) -> None: if self.cache_hit: work_rows = [row for row in work_rows if row not in self.from_cache["H_row"]] - if all( - [i in self.from_cache["H_row"] for i in work_rows] + if ( + all([i in self.from_cache["H_row"] for i in work_rows]) and Work["tag"] == EVAL_SIM_TAG ): # if all rows in work_rows are found in cache return @@ -510,7 +509,7 @@ def _receive_from_workers(self, persis_info: dict) -> dict: """ time.sleep(0.0001) # Critical for multiprocessing performance - if self.cache_hit or len(self.from_cache): + if self.cache_hit: self.cache_hit = False for w in self.from_cache["worker_id"]: self._handle_msg_from_worker(persis_info, w, process_cache=True) @@ -589,10 +588,11 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool """Handles a message from worker w""" try: if process_cache: + cache_entry_by_worker = self.from_cache["worker_id"] == w D_recv = { - "calc_out": self.from_cache[[name[0] for name in self.sim_specs["out"]]], + "calc_out": self.from_cache[cache_entry_by_worker][[name[0] for name in self.sim_specs["out"]]], "libE_info": { - "H_rows": self.from_cache["H_row"], + "H_rows": self.from_cache[cache_entry_by_worker]["H_row"], "workerID": w, }, "calc_status": 0, @@ -615,7 +615,10 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool logger.vdebug(f"Manager received a log message from worker {w}") # type: ignore[attr-defined] logging.getLogger(D_recv.name).handle(D_recv) else: - logger.debug(f"Manager received data message from worker {w}") + if process_cache: + logger.debug(f"Manager retrieved cached message redirected from worker {w}") + else: + logger.debug(f"Manager received data message from worker {w}") self._update_state_on_worker_msg(persis_info, D_recv, w) def _kill_cancelled_sims(self) -> None: @@ -707,6 +710,8 @@ def _sim_max_given(self) -> bool: def _get_alloc_libE_info(self) -> dict: """Selected statistics useful for alloc_f""" + cache = self.hist.get_shelved_sims() if self.hist.cache_set else [] + return { "any_idle_workers": any(self.W["active"] == 0), "exit_criteria": self.exit_criteria, @@ -720,13 +725,10 @@ def _get_alloc_libE_info(self) -> dict: "use_resource_sets": self.use_resource_sets, "gen_num_procs": self.gen_num_procs, "gen_num_gpus": self.gen_num_gpus, -<<<<<<< HEAD "gen_on_worker": self.libE_specs.get("gen_on_worker", False), + "gen_on_manager": self.libE_specs.get("gen_on_manager", False), "cache": cache, "hist": self.hist, -======= - "gen_on_manager": self.libE_specs.get("gen_on_manager", False), ->>>>>>> df29125a9 (moving cache logic into manager, into handle_msg_from_worker that overrides .recv, and in _send_work_order that forms a buffer from corresponding cache entries) } def _alloc_work(self, H: npt.NDArray, persis_info: dict) -> dict: diff --git a/libensemble/tools/test_support.py b/libensemble/tools/test_support.py index bad9c2a78..7e0727125 100644 --- a/libensemble/tools/test_support.py +++ b/libensemble/tools/test_support.py @@ -244,9 +244,9 @@ def check_gpu_setting(task, assert_setting=True, print_setting=False, resources= if assert_setting: if isinstance(expected, dict): for key, value in expected.items(): - assert key in gpu_setting, ( - f"Worker {task.workerID}: Expected env key '{key}' not found in GPU setting: {gpu_setting}" - ) + assert ( + key in gpu_setting + ), f"Worker {task.workerID}: Expected env key '{key}' not found in GPU setting: {gpu_setting}" assert gpu_setting[key] == value, ( f"Worker {task.workerID}: GPU setting key '{key}' has value '{gpu_setting[key]}', " f"expected '{value}'" diff --git a/pyproject.toml b/pyproject.toml index 257bcb15c..bef73a62e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -186,6 +186,9 @@ setuptools = ">=80.8.0,<81" numpy = ">=2.2.6,<3" pydantic = ">=2.12.4,<3" gest-api = ">=0.1,<0.2" +pyyaml = ">=6.0,<7" +tomli = ">=1.2.1,<3" +psutil = ">=5.9.4,<7" # macOS dependencies @@ -205,7 +208,13 @@ extra = [ "surmise>=0.3.0,<0.4", "optimas @ git+https://github.com/optimas-org/optimas", ] -dev = ["wat>=0.7.0,<0.8"] +dev = [ + "wat>=0.7.0,<0.8", + "pyenchant", + "enchant>=0.0.1,<0.0.2", + "flake8-modern-annotations>=1.6.0,<2", + "flake8-type-checking>=3.0.0,<4", +] docs = ["pyenchant", "enchant>=0.0.1,<0.0.2", "sphinx-lfs-content>=1.1.10,<2"] # Various config from here onward From 1f674e6ec07b4178a8161c705dc843888343ed01 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 19 Sep 2025 14:34:54 -0500 Subject: [PATCH 13/38] refactor, and remove first draft of code that was in alloc_f --- .../alloc_funcs/give_sim_work_first.py | 44 ---------- libensemble/manager.py | 88 +++++++++++++------ 2 files changed, 60 insertions(+), 72 deletions(-) diff --git a/libensemble/alloc_funcs/give_sim_work_first.py b/libensemble/alloc_funcs/give_sim_work_first.py index 0aaf76acf..96245f7a9 100644 --- a/libensemble/alloc_funcs/give_sim_work_first.py +++ b/libensemble/alloc_funcs/give_sim_work_first.py @@ -62,50 +62,6 @@ def give_sim_work_first( gen_count = support.count_gens() Work = {} - # if not persis_info.get("updated_H_indices"): - # persis_info["updated_H_indices"] = [] - - # points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] - - # gen_out_fields = [j[0] for j in gen_specs["out"]] - - # indices_to_update = [] - # cache_hit = False - - # if len(libE_info["cache"]) and np.any(points_to_evaluate): - # for H_index, H_entry in enumerate(H): - # if H_index in persis_info["updated_H_indices"]: - # continue - # for cache_index, cache_entry in enumerate(libE_info["cache"]): - # for field in gen_out_fields: - # if field in libE_info["cache"].dtype.names and np.allclose( - # H_entry[field], cache_entry[field], rtol=1e-8, atol=1e-8 - # ): - # cache_hit = True - # indices_to_update.append({"cache_index": cache_index, "H_index": H_index, "field": field}) - # persis_info["updated_H_indices"].append(H_index) - # print(f"Using cache entry {cache_index} for History index {H_index}. Field: {field}") - - # if cache_hit: - - # q_inds = np.array([i["H_index"] for i in indices_to_update]) - # libE_info["hist"].update_history_x_out(q_inds=q_inds, sim_worker=1) - - # simulated_calc_out = np.zeros(len(q_inds), dtype=sim_specs["out"]) - # for i, H_index in enumerate(q_inds): - # simulated_calc_out[i] = libE_info["cache"][indices_to_update[i]["cache_index"]][ - # indices_to_update[i]["field"] - # ] - - # simulated_D_recv = { - # "calc_out": simulated_calc_out, - # "libE_info": { - # "H_rows": q_inds, - # }, - # } - - # libE_info["hist"].update_history_f(simulated_D_recv, kill_canceled_sims=False) - points_to_evaluate = ~H["sim_started"] & ~H["cancel_requested"] if np.any(points_to_evaluate): diff --git a/libensemble/manager.py b/libensemble/manager.py index b7083683c..371df33e8 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -419,6 +419,44 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: if "sim_id" not in self.gen_specs["persis_in"]: self.gen_specs["persis_in"].append("sim_id") + def _refresh_from_cache( + self, cache: npt.NDArray, dtype_with_idx: np.dtype, cache_row: npt.NDArray, work_row: int, w: int + ) -> None: + """Add a cache entry, workerID, and H_row to the record array.""" + self.cache_hit = True + from_cache_entry = np.empty(1, dtype=dtype_with_idx) + from_cache_entry["H_row"] = work_row + from_cache_entry["worker_id"] = w + for remaining_field in cache.dtype.names: + from_cache_entry[remaining_field] = cache_row[remaining_field] + self.from_cache[self.cache_index] = from_cache_entry + self.cache_index += 1 + + def _cache_scan( + self, cache: npt.NDArray, Work: dict, w: int, dtype_with_idx: np.dtype, new_dtype: np.dtype + ) -> None: + """Check if any work rows are in the cache, and if so, call the above, _refresh_from_cache.""" + + for field in np.dtype(new_dtype).names: + if field in cache.dtype.names: + for work_row in Work["libE_info"]["H_rows"]: + for cache_row in cache: + if ( + np.allclose(cache_row[field], self.hist.H[field][work_row]) + and work_row not in self.from_cache["H_row"] + ): # we found outbound work in cache, that's not already been retrieved + self._refresh_from_cache(cache, dtype_with_idx, cache_row, work_row, w) + + def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: + """Retrieve saved cache from history, create local record-array of matching cache entries.""" + cache = self.hist.get_shelved_sims() + dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int), ("worker_id", int)]).descr) + if not len(self.from_cache): + self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache + else: + self.from_cache = np.append(self.from_cache, np.zeros(len(work_rows), dtype=dtype_with_idx)) + self._cache_scan(cache, Work, w, dtype_with_idx, new_dtype) + def _send_work_order(self, Work: dict, w: int) -> None: """Sends an allocation function order to a worker""" logger.debug(f"Manager sending work unit to worker {w}") @@ -427,29 +465,7 @@ def _send_work_order(self, Work: dict, w: int) -> None: new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] if Work["tag"] == EVAL_SIM_TAG and len(work_rows) and self.hist.cache_set: - cache = self.hist.get_shelved_sims() - dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int), ("worker_id", int)]).descr) - if not len(self.from_cache): - self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache - else: - self.from_cache = np.append(self.from_cache, np.zeros(len(work_rows), dtype=dtype_with_idx)) - - for field in np.dtype(new_dtype).names: - if field in cache.dtype.names: - for row in work_rows: - for cache_row in cache: - if ( - np.allclose(cache_row[field], self.hist.H[field][row]) - and row not in self.from_cache["H_row"] - ): # we found outbound work in cache, that's not already been retrieved - self.cache_hit = True - from_cache_entry = np.empty(1, dtype=dtype_with_idx) - from_cache_entry["H_row"] = row - from_cache_entry["worker_id"] = w - for remaining_field in cache.dtype.names: - from_cache_entry[remaining_field] = cache_row[remaining_field] - self.from_cache[self.cache_index] = from_cache_entry - self.cache_index += 1 + self._update_state_from_cache(Work, work_rows, w, new_dtype) if self.resources: self._set_resources(Work, w) @@ -473,6 +489,7 @@ def _send_work_order(self, Work: dict, w: int) -> None: if ( all([i in self.from_cache["H_row"] for i in work_rows]) and Work["tag"] == EVAL_SIM_TAG ): # if all rows in work_rows are found in cache + logger.debug("Manager skipping sending work to worker %s due to cache", w) return H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) @@ -501,8 +518,19 @@ def _update_state_on_alloc(self, Work: dict, w: int): # --- Handle incoming messages from workers - def _receive_from_workers(self, persis_info: dict) -> dict: - """Receives calculation output from workers. Loops over all + def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: + """ + Two stage process of handling either: + 1. Messages that could've been sent to a worker, but are already in the cache. + 2. Messages that have been sent by a worker. + + 1. + If the cache is not empty, the cache is scanned for messages that could've been sent. + Messages are processed as though they came from their corresponding worker. The local + record of the cache is then cleared to prevent duplicate processing. + + 2. + Receives calculation output from workers. Loops over all active workers and probes to see if worker is ready to communticate. If any output is received, all other workers are looped back over. @@ -585,7 +613,11 @@ def _update_state_on_worker_msg(self, persis_info: dict, D_recv: dict, w: int) - persis_info.setdefault(int(w), {}).update(D_recv["persis_info"]) def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool = False) -> None: - """Handles a message from worker w""" + """Handles a message from worker w. + + If processing from the cache, create a simulated worker message containing + the cache entry. + """ try: if process_cache: cache_entry_by_worker = self.from_cache["worker_id"] == w @@ -679,7 +711,7 @@ def _final_receive_and_kill(self, persis_info: dict) -> tuple[dict, int, int]: exit_flag = 0 while (any(self.W["active"]) or any(self.W["persis_state"])) and exit_flag == 0: - persis_info = self._receive_from_workers(persis_info) + persis_info = self._receive_from_workers_or_cache(persis_info) if self.term_test(logged=False) == 2: # Elapsed Wallclock has expired if not any(self.W["persis_state"]): @@ -774,7 +806,7 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: try: while not self.term_test(): self._kill_cancelled_sims() - persis_info = self._receive_from_workers(persis_info) + persis_info = self._receive_from_workers_or_cache(persis_info) Work, persis_info, flag = self._alloc_work(self.hist.trim_H(), persis_info) if flag: break From 28c80c41e6a39a418bb9e34f045f996cd53a8335 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 19 Sep 2025 14:38:19 -0500 Subject: [PATCH 14/38] for now, enable cache for sims that lasted longer than a second --- libensemble/history.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/history.py b/libensemble/history.py index dcff4845d..eb65f7697 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -133,7 +133,7 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: def _shelf_longrunning_sims(self, index): """Cache any f values that ran for more than a second.""" - if 1: # self.H[index]['sim_ended_time'] - self.H[index]['sim_started_time'] > 1: + if self.H[index]["sim_ended_time"] - self.H[index]["sim_started_time"] > 1: # ('f', 'x') and ('x', 'f') are not equivalent dtypes, unfortunately. So maybe sorted helps. self.cache_keys = sorted( [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] From ccfd5ea365c82588c0d8c24064b76b34862eec41 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 19 Sep 2025 14:45:23 -0500 Subject: [PATCH 15/38] experimenting with making disk cache name match calling script plus easily accessible args --- libensemble/history.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/libensemble/history.py b/libensemble/history.py index eb65f7697..ccc78403c 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -1,4 +1,5 @@ import logging +import sys import time from pathlib import Path @@ -118,7 +119,7 @@ def __init__( self.cache_dir = Path.home() / ".libE" self.cache_dir.mkdir(parents=True, exist_ok=True) - self.cache = self.cache_dir / "cache.npy" + self.cache = self.cache_dir / Path("".join(sys.argv) + ".npy") if not self.cache.exists(): self.cache.touch() self.cache_set = False From 9aae021dfa97ccb1574eee57a97790624fe82961 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 19 Sep 2025 15:39:30 -0500 Subject: [PATCH 16/38] fix redundant send of work if rows send to gen. tiny test fix --- libensemble/manager.py | 5 +++-- .../functionality_tests/test_executor_hworld_pass_fail.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 371df33e8..528ce11f2 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -489,14 +489,15 @@ def _send_work_order(self, Work: dict, w: int) -> None: if ( all([i in self.from_cache["H_row"] for i in work_rows]) and Work["tag"] == EVAL_SIM_TAG ): # if all rows in work_rows are found in cache - logger.debug("Manager skipping sending work to worker %s due to cache", w) + logger.debug("Manager skipping sending *all* work to worker %s due to cache", w) return H_to_be_sent = np.empty(len(work_rows), dtype=new_dtype) for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) - self.wcomms[w].send(Work["tag"], Work) + if Work["tag"] == EVAL_SIM_TAG: + self.wcomms[w].send(Work["tag"], Work) self.wcomms[w].send(0, H_to_be_sent) def _update_state_on_alloc(self, Work: dict, w: int): diff --git a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py index 83a9cbc9a..b7712ac29 100644 --- a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py +++ b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py @@ -121,7 +121,7 @@ # For debug print(f"Expecting: {calc_status_list}") - print("Received: {H['cstat']}\n") + print(f"Received: {H['cstat']}\n") assert np.array_equal(H["cstat"], calc_status_list), "Error - unexpected calc status. Received: " + str( H["cstat"] From d24fe866fa539d31f1c3f797a92832529fcc51f8 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 24 Sep 2025 12:33:27 -0500 Subject: [PATCH 17/38] add libE_specs.cache_long_sims, plus more/better docstrings --- libensemble/history.py | 7 +++-- libensemble/manager.py | 71 +++++++++++++++++++++++++++++++----------- libensemble/specs.py | 10 ++++++ 3 files changed, 68 insertions(+), 20 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index ccc78403c..747a8c9db 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -104,6 +104,7 @@ def __init__( self.index = len(H0) self.grow_count = 0 self.safe_mode = False + self.use_cache = False self.sim_started_count = np.sum(H["sim_started"]) self.sim_ended_count = np.sum(H["sim_ended"]) @@ -117,6 +118,7 @@ def __init__( self.last_started = -1 self.last_ended = -1 + def init_cache(self) -> None: self.cache_dir = Path.home() / ".libE" self.cache_dir.mkdir(parents=True, exist_ok=True) self.cache = self.cache_dir / Path("".join(sys.argv) + ".npy") @@ -149,7 +151,7 @@ def _shelf_longrunning_sims(self, index): entry = self.H[index][self.cache_keys] if entry not in in_cache: in_cache = np.append(in_cache, entry) - in_cache = np.unique(in_cache, axis=0) + in_cache = np.unique(in_cache, axis=0) # attempt to remove duplicates np.save(self.cache, in_cache, allow_pickle=True) self.cache_set = True @@ -193,7 +195,8 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: self.H["sim_ended"][ind] = True self.H["sim_ended_time"][ind] = time.time() self.sim_ended_count += 1 - self._shelf_longrunning_sims(ind) + if self.use_cache: + self._shelf_longrunning_sims(ind) if kill_canceled_sims: for j in range(self.last_ended + 1, np.max(new_inds) + 1): diff --git a/libensemble/manager.py b/libensemble/manager.py index 528ce11f2..b87c2a9ab 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -204,9 +204,11 @@ def __init__( timer.start() self.date_start = timer.date_start.replace(" ", "_") self.safe_mode = libE_specs.get("safe_mode") + self.use_cache = libE_specs.get("cache_long_sims") self.kill_canceled_sims = libE_specs.get("kill_canceled_sims") self.hist = hist self.hist.safe_mode = self.safe_mode + self.hist.use_cache = self.use_cache self.libE_specs = libE_specs self.alloc_specs = alloc_specs self.sim_specs = sim_specs @@ -217,8 +219,10 @@ def __init__( self.WorkerExc = False self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") - self.from_cache = [] - self.cache_index = 0 + if self.use_cache: + self.hist.init_cache() + self.from_cache = [] + self.cache_index = 0 self.cache_hit = False dyn_keys = ("resource_sets", "num_procs", "num_gpus") @@ -422,7 +426,11 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: def _refresh_from_cache( self, cache: npt.NDArray, dtype_with_idx: np.dtype, cache_row: npt.NDArray, work_row: int, w: int ) -> None: - """Add a cache entry, workerID, and H_row to the record array.""" + """Add a cache entry, workerID, and H_row to the local record array. + + Later on when we iterate over the cache for entries that could've been sent to a worker (but weren't), + we'll process that entry as though it came from this worker, with these H_rows. + """ self.cache_hit = True from_cache_entry = np.empty(1, dtype=dtype_with_idx) from_cache_entry["H_row"] = work_row @@ -435,7 +443,10 @@ def _refresh_from_cache( def _cache_scan( self, cache: npt.NDArray, Work: dict, w: int, dtype_with_idx: np.dtype, new_dtype: np.dtype ) -> None: - """Check if any work rows are in the cache, and if so, call the above, _refresh_from_cache.""" + """ + Check if any work rows are in the cache, and if so, call the above, _refresh_from_cache + to update the local `from_cache` record. + """ for field in np.dtype(new_dtype).names: if field in cache.dtype.names: @@ -444,17 +455,33 @@ def _cache_scan( if ( np.allclose(cache_row[field], self.hist.H[field][work_row]) and work_row not in self.from_cache["H_row"] - ): # we found outbound work in cache, that's not already been retrieved + ): # we found outbound work in cache, that's not already in the local record self._refresh_from_cache(cache, dtype_with_idx, cache_row, work_row, w) def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: - """Retrieve saved cache from history, create local record-array of matching cache entries.""" + """Retrieve saved cache from history, create local record-array of matching cache entries. + + The `from_cache` local record contains cache entries and the workerID and H_rows they are associated with, had + they been sent to a worker. + + Cache entries *must* be associated with the preempted outbound worker and H_rows because those values + are always associated with actual inbound results. Later on, when we iterate over the cache for entries that + could've been sent to a worker (but weren't), we'll process that entry as though it came from that worker, + with those H_rows. + """ + cache = self.hist.get_shelved_sims() + + # our local record resembles the cache, but additionally with the worker_id and H_row from the alloc_f dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int), ("worker_id", int)]).descr) + + # initialize or grow the local record, then call _cache_scan to fill it if not len(self.from_cache): - self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) # all work may be in cache + self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) else: self.from_cache = np.append(self.from_cache, np.zeros(len(work_rows), dtype=dtype_with_idx)) + + # populates the local record self._cache_scan(cache, Work, w, dtype_with_idx, new_dtype) def _send_work_order(self, Work: dict, w: int) -> None: @@ -464,7 +491,7 @@ def _send_work_order(self, Work: dict, w: int) -> None: work_rows = Work["libE_info"]["H_rows"] new_dtype = [(name, self.hist.H.dtype.fields[name][0]) for name in Work["H_fields"]] - if Work["tag"] == EVAL_SIM_TAG and len(work_rows) and self.hist.cache_set: + if self.use_cache and Work["tag"] == EVAL_SIM_TAG and len(work_rows) and self.hist.cache_set: self._update_state_from_cache(Work, work_rows, w, new_dtype) if self.resources: @@ -538,6 +565,7 @@ def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: """ time.sleep(0.0001) # Critical for multiprocessing performance + # Process messages from the cache if self.cache_hit: self.cache_hit = False for w in self.from_cache["worker_id"]: @@ -545,6 +573,7 @@ def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: self.from_cache = [] self.cache_index = 0 + # Process messages from workers new_stuff = True while new_stuff: new_stuff = False @@ -613,6 +642,21 @@ def _update_state_on_worker_msg(self, persis_info: dict, D_recv: dict, w: int) - if D_recv.get("persis_info"): persis_info.setdefault(int(w), {}).update(D_recv["persis_info"]) + def _create_simulated_D_recv(self, w: int) -> dict: + """Create a simulated worker message containing the cache entry instead of a message from a worker.""" + + cache_entry_by_worker = self.from_cache[self.from_cache["worker_id"] == w] + D_recv = { + "calc_out": cache_entry_by_worker[[name[0] for name in self.sim_specs["out"]]], + "libE_info": { + "H_rows": cache_entry_by_worker["H_row"], + "workerID": w, + }, + "calc_status": 0, + "calc_type": 1, + } + return D_recv + def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool = False) -> None: """Handles a message from worker w. @@ -621,16 +665,7 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool """ try: if process_cache: - cache_entry_by_worker = self.from_cache["worker_id"] == w - D_recv = { - "calc_out": self.from_cache[cache_entry_by_worker][[name[0] for name in self.sim_specs["out"]]], - "libE_info": { - "H_rows": self.from_cache[cache_entry_by_worker]["H_row"], - "workerID": w, - }, - "calc_status": 0, - "calc_type": 1, - } + D_recv = self._create_simulated_D_recv(w) else: msg = self.wcomms[w].recv() tag, D_recv = msg diff --git a/libensemble/specs.py b/libensemble/specs.py index c95bdd8f8..32611ab7f 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -579,6 +579,16 @@ class LibeSpecs(BaseModel): Forms the base of a generator directory. """ + cache_long_sims: bool | None = False + """ + Cache simulation results with runtimes >1s to disk. Subsequent runs of the same + base script with the same command-line arguments will access this cache. + + Upon the generator creating points already in the cache, those points will be skipped from + being sent for evaluation. Instead the corresponding cached results are retrieved and returned + to the generator. + """ + calc_dir_id_width: int | None = 4 """ The width of the numerical ID component of a calculation directory name. Leading From 4b57ff62200cb8fbc94e93aa7f2932010a94acaa Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 25 Sep 2025 15:40:37 -0500 Subject: [PATCH 18/38] manager builds libE_stats messages corresponding to cache retrievals. tiny refactors to worker.py to help reuse a method. add CACHE_RETRIEVE tag --- libensemble/manager.py | 31 ++++++++++++------- libensemble/message_numbers.py | 3 ++ .../test_executor_hworld_pass_fail.py | 2 ++ libensemble/worker.py | 7 +++-- 4 files changed, 29 insertions(+), 14 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index b87c2a9ab..1c4293e6b 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -20,8 +20,10 @@ from numpy.lib.recfunctions import repack_fields from libensemble.comms.comms import CommFinishedException, QCommThread +from libensemble.comms.logs import LogConfig from libensemble.executors.executor import Executor from libensemble.message_numbers import ( + CACHE_RETRIEVE, EVAL_GEN_TAG, EVAL_SIM_TAG, FINISHED_PERSISTENT_GEN_TAG, @@ -30,6 +32,7 @@ MAN_SIGNAL_KILL, PERSIS_STOP, STOP_TAG, + calc_status_strings, calc_type_strings, ) from libensemble.resources.resources import Resources @@ -38,7 +41,7 @@ from libensemble.utils.misc import _WorkerIndexer, extract_H_ranges from libensemble.utils.output_directory import EnsembleDirectory from libensemble.utils.timer import Timer -from libensemble.worker import WorkerErrMsg, worker_main +from libensemble.worker import Worker, WorkerErrMsg, worker_main logger = logging.getLogger(__name__) # For debug messages - uncomment @@ -448,15 +451,17 @@ def _cache_scan( to update the local `from_cache` record. """ - for field in np.dtype(new_dtype).names: - if field in cache.dtype.names: - for work_row in Work["libE_info"]["H_rows"]: - for cache_row in cache: - if ( - np.allclose(cache_row[field], self.hist.H[field][work_row]) - and work_row not in self.from_cache["H_row"] - ): # we found outbound work in cache, that's not already in the local record - self._refresh_from_cache(cache, dtype_with_idx, cache_row, work_row, w) + self.cache_timer = Timer() + with self.cache_timer: + for field in np.dtype(new_dtype).names: + if field in cache.dtype.names: + for work_row in Work["libE_info"]["H_rows"]: + for cache_row in cache: + if ( + np.allclose(cache_row[field], self.hist.H[field][work_row]) + and work_row not in self.from_cache["H_row"] + ): # we found outbound work in cache, that's not already in the local record + self._refresh_from_cache(cache, dtype_with_idx, cache_row, work_row, w) def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: """Retrieve saved cache from history, create local record-array of matching cache entries. @@ -652,7 +657,7 @@ def _create_simulated_D_recv(self, w: int) -> dict: "H_rows": cache_entry_by_worker["H_row"], "workerID": w, }, - "calc_status": 0, + "calc_status": CACHE_RETRIEVE, "calc_type": 1, } return D_recv @@ -666,6 +671,7 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool try: if process_cache: D_recv = self._create_simulated_D_recv(w) + enum_desc, calc_id = Worker._extract_debug_data(1, D_recv) else: msg = self.wcomms[w].recv() tag, D_recv = msg @@ -685,6 +691,9 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool else: if process_cache: logger.debug(f"Manager retrieved cached message redirected from worker {w}") + calc_msg = f"{enum_desc} {calc_id}: {"sim"} {self.cache_timer}" + calc_msg += f" Status: {calc_status_strings[CACHE_RETRIEVE]}" + logging.getLogger(LogConfig.config.stats_name).info(calc_msg) else: logger.debug(f"Manager received data message from worker {w}") self._update_state_on_worker_msg(persis_info, D_recv, w) diff --git a/libensemble/message_numbers.py b/libensemble/message_numbers.py index 0ecc7092e..46ff8c987 100644 --- a/libensemble/message_numbers.py +++ b/libensemble/message_numbers.py @@ -41,6 +41,8 @@ WORKER_DONE = 35 # Calculation was successful # last_calc_status_rst_tag CALC_EXCEPTION = 36 # Reserved: Automatically used if user_f raised an exception +CACHE_RETRIEVE = 40 # Manager retrieved sim from cache + MAN_KILL_SIGNALS = [MAN_SIGNAL_FINISH, MAN_SIGNAL_KILL] @@ -57,6 +59,7 @@ TASK_FAILED_TO_START: "Task Failed to start", WORKER_DONE: "Completed", CALC_EXCEPTION: "Exception occurred", + CACHE_RETRIEVE: "Retrieved from cache", None: "Unknown Status", } # last_calc_status_string_rst_tag diff --git a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py index b7712ac29..184cbeef5 100644 --- a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py +++ b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py @@ -62,6 +62,8 @@ if is_manager: print(f"\nCores req: {cores_all_tasks} Cores avail: {logical_cores}\n {mess_resources}\n") + libE_specs["cache_long_sims"] = True + sim_app = "./my_simtask.x" if not os.path.isfile(sim_app): build_simfunc() diff --git a/libensemble/worker.py b/libensemble/worker.py index c6ef5fcb3..0ecb91765 100644 --- a/libensemble/worker.py +++ b/libensemble/worker.py @@ -229,7 +229,8 @@ def _set_resources(workerID, comm: Comm, libE_specs) -> bool: logger.debug(f"No resources set on worker {workerID}") return False - def _extract_debug_data(self, calc_type, Work): + @staticmethod + def _extract_debug_data(calc_type, Work): if calc_type == EVAL_SIM_TAG: enum_desc = "sim_id" calc_id = extract_H_ranges(Work) @@ -260,7 +261,7 @@ def _handle_calc(self, Work: dict, calc_in: npt.NDArray) -> tuple[npt.NDArray | calc_type = Work["tag"] self.calc_iter[calc_type] += 1 - enum_desc, calc_id = self._extract_debug_data(calc_type, Work) + enum_desc, calc_id = Worker._extract_debug_data(calc_type, Work) timer = Timer() @@ -317,7 +318,7 @@ def _handle_calc(self, Work: dict, calc_in: npt.NDArray) -> tuple[npt.NDArray | logging.getLogger(LogConfig.config.stats_name).info(calc_msg) - def _get_calc_msg(self, enum_desc: str, calc_id: str, calc_type: str, timer: Timer, status: int | str) -> str: + def _get_calc_msg(self, enum_desc: str, calc_id: int | str, calc_type: str, timer: Timer, status: int | str) -> str: """Construct line for libE_stats.txt file""" calc_msg = f"{enum_desc} {calc_id}: {calc_type} {timer}" From f268eda744326cea8292a70545a952b8113bd12d Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 26 Sep 2025 15:06:37 -0500 Subject: [PATCH 19/38] user can specify database name; trying to figure out occasionally-malformed cache data / H_rows from the cache --- libensemble/history.py | 11 +++++++---- libensemble/manager.py | 7 ++++--- libensemble/specs.py | 8 ++++++++ .../test_executor_hworld_pass_fail.py | 1 + 4 files changed, 20 insertions(+), 7 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 747a8c9db..8c825e325 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -1,5 +1,4 @@ import logging -import sys import time from pathlib import Path @@ -118,12 +117,13 @@ def __init__( self.last_started = -1 self.last_ended = -1 - def init_cache(self) -> None: + def init_cache(self, cache_name: str) -> None: self.cache_dir = Path.home() / ".libE" self.cache_dir.mkdir(parents=True, exist_ok=True) - self.cache = self.cache_dir / Path("".join(sys.argv) + ".npy") + self.cache = self.cache_dir / Path(cache_name + ".npy") if not self.cache.exists(): self.cache.touch() + self.use_cache = True self.cache_set = False def _append_new_fields(self, H_f: npt.NDArray) -> None: @@ -156,7 +156,10 @@ def _shelf_longrunning_sims(self, index): self.cache_set = True def get_shelved_sims(self) -> npt.NDArray: - in_cache = np.load(self.cache, allow_pickle=True) + try: + in_cache = np.load(self.cache, allow_pickle=True) + except EOFError: + in_cache = np.zeros(1, dtype=self.cache_dtype) return in_cache def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: diff --git a/libensemble/manager.py b/libensemble/manager.py index 1c4293e6b..2396001e2 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -211,7 +211,6 @@ def __init__( self.kill_canceled_sims = libE_specs.get("kill_canceled_sims") self.hist = hist self.hist.safe_mode = self.safe_mode - self.hist.use_cache = self.use_cache self.libE_specs = libE_specs self.alloc_specs = alloc_specs self.sim_specs = sim_specs @@ -223,7 +222,7 @@ def __init__( self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") if self.use_cache: - self.hist.init_cache() + self.hist.init_cache(self.libE_specs.get("cache_name")) self.from_cache = [] self.cache_index = 0 self.cache_hit = False @@ -672,6 +671,8 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool if process_cache: D_recv = self._create_simulated_D_recv(w) enum_desc, calc_id = Worker._extract_debug_data(1, D_recv) + if calc_id.startswith("0_0"): + print("why do we have weird sim_id values?") else: msg = self.wcomms[w].recv() tag, D_recv = msg @@ -693,7 +694,7 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool logger.debug(f"Manager retrieved cached message redirected from worker {w}") calc_msg = f"{enum_desc} {calc_id}: {"sim"} {self.cache_timer}" calc_msg += f" Status: {calc_status_strings[CACHE_RETRIEVE]}" - logging.getLogger(LogConfig.config.stats_name).info(calc_msg) + logging.getLogger(LogConfig.config.stats_name).info(calc_msg) # libE_stats else: logger.debug(f"Manager received data message from worker {w}") self._update_state_on_worker_msg(persis_info, D_recv, w) diff --git a/libensemble/specs.py b/libensemble/specs.py index 32611ab7f..a0b08a877 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -1,4 +1,5 @@ import random +import sys import warnings from pathlib import Path @@ -587,6 +588,13 @@ class LibeSpecs(BaseModel): Upon the generator creating points already in the cache, those points will be skipped from being sent for evaluation. Instead the corresponding cached results are retrieved and returned to the generator. + + The cache is saved in $HOME/.libE, and by default is named after the joined command-line arguments. + """ + + cache_name: str | None = Path.home() / ".libE" / Path("".join(sys.argv) + ".npy") + """ + The name of the cache file. By default is the joined command-line arguments. """ calc_dir_id_width: int | None = 4 diff --git a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py index 184cbeef5..5a0da5e79 100644 --- a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py +++ b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py @@ -63,6 +63,7 @@ print(f"\nCores req: {cores_all_tasks} Cores avail: {logical_cores}\n {mess_resources}\n") libE_specs["cache_long_sims"] = True + libE_specs["cache_name"] = "asdf" sim_app = "./my_simtask.x" if not os.path.isfile(sim_app): From 37501d988f50190fa2b321ecd905b4ae85d8d590 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 08:28:17 -0500 Subject: [PATCH 20/38] fix syntax error uncaught by black and other tools? --- libensemble/manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 2396001e2..52941d47a 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -692,7 +692,7 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool else: if process_cache: logger.debug(f"Manager retrieved cached message redirected from worker {w}") - calc_msg = f"{enum_desc} {calc_id}: {"sim"} {self.cache_timer}" + calc_msg = f"""{enum_desc} {calc_id}: {"sim"} {self.cache_timer}""" calc_msg += f" Status: {calc_status_strings[CACHE_RETRIEVE]}" logging.getLogger(LogConfig.config.stats_name).info(calc_msg) # libE_stats else: From 77675f225b4df7929e3e51b073f37e062cda85c2 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 08:32:21 -0500 Subject: [PATCH 21/38] cache_name is only string. path not needed in specs.py --- libensemble/specs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/specs.py b/libensemble/specs.py index a0b08a877..ed41d0c88 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -592,7 +592,7 @@ class LibeSpecs(BaseModel): The cache is saved in $HOME/.libE, and by default is named after the joined command-line arguments. """ - cache_name: str | None = Path.home() / ".libE" / Path("".join(sys.argv) + ".npy") + cache_name: str | Path | None = Path.home() / ".libE" / Path("".join(sys.argv) + ".npy") """ The name of the cache file. By default is the joined command-line arguments. """ From fe3e03f38c38683092116132e1a9f2393b912f37 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 09:27:27 -0500 Subject: [PATCH 22/38] param fix --- libensemble/specs.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/libensemble/specs.py b/libensemble/specs.py index ed41d0c88..df59bbae3 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -592,9 +592,10 @@ class LibeSpecs(BaseModel): The cache is saved in $HOME/.libE, and by default is named after the joined command-line arguments. """ - cache_name: str | Path | None = Path.home() / ".libE" / Path("".join(sys.argv) + ".npy") + cache_name: str | None = "".join(sys.argv) """ The name of the cache file. By default is the joined command-line arguments. + Stored in $HOME/.libE, and by default is named after the joined command-line arguments. """ calc_dir_id_width: int | None = 4 From dc3202e27be5059ac3ef840a7e5a831002e59a23 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 09:43:12 -0500 Subject: [PATCH 23/38] still want to send Work on persis_stop if we're doing final_gen_send --- libensemble/manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 52941d47a..f96d3cc37 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -527,7 +527,7 @@ def _send_work_order(self, Work: dict, w: int) -> None: for i, row in enumerate(work_rows): H_to_be_sent[i] = repack_fields(self.hist.H[Work["H_fields"]][row]) - if Work["tag"] == EVAL_SIM_TAG: + if Work["tag"] in [EVAL_SIM_TAG, PERSIS_STOP]: # inclusion of PERSIS_STOP for final_gen_send self.wcomms[w].send(Work["tag"], Work) self.wcomms[w].send(0, H_to_be_sent) From 9def472414d61a6f99602f043cff623573c701f0 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 13:36:18 -0500 Subject: [PATCH 24/38] don't necessarily need cache collisions for these executor tests - since we're comparing to returned calc statuses --- .../tests/functionality_tests/test_executor_hworld_pass_fail.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py index 5a0da5e79..8b2ed8b61 100644 --- a/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py +++ b/libensemble/tests/functionality_tests/test_executor_hworld_pass_fail.py @@ -63,7 +63,7 @@ print(f"\nCores req: {cores_all_tasks} Cores avail: {logical_cores}\n {mess_resources}\n") libE_specs["cache_long_sims"] = True - libE_specs["cache_name"] = "asdf" + libE_specs["cache_name"] = "executor_hworld_" + str(nworkers) + "_" + libE_specs.get("comms") sim_app = "./my_simtask.x" if not os.path.isfile(sim_app): From 5cca0a478099b7d680b0c37f4a8e216322a7db7d Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 13:58:33 -0500 Subject: [PATCH 25/38] fix iterating over blank template cache entries as though they're valid data --- libensemble/manager.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index f96d3cc37..682dd6938 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -573,7 +573,8 @@ def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: if self.cache_hit: self.cache_hit = False for w in self.from_cache["worker_id"]: - self._handle_msg_from_worker(persis_info, w, process_cache=True) + if w > 0: # actual cache entry - not blank. assuming w0 gets no sim work + self._handle_msg_from_worker(persis_info, w, process_cache=True) self.from_cache = [] self.cache_index = 0 @@ -671,8 +672,6 @@ def _handle_msg_from_worker(self, persis_info: dict, w: int, process_cache: bool if process_cache: D_recv = self._create_simulated_D_recv(w) enum_desc, calc_id = Worker._extract_debug_data(1, D_recv) - if calc_id.startswith("0_0"): - print("why do we have weird sim_id values?") else: msg = self.wcomms[w].recv() tag, D_recv = msg From e41f006f9f67f7eca25c930e9cd055cd94656293 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 14:00:25 -0500 Subject: [PATCH 26/38] add functionality test for cache_sims --- .../functionality_tests/test_cache_sims.py | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 libensemble/tests/functionality_tests/test_cache_sims.py diff --git a/libensemble/tests/functionality_tests/test_cache_sims.py b/libensemble/tests/functionality_tests/test_cache_sims.py new file mode 100644 index 000000000..885a6556e --- /dev/null +++ b/libensemble/tests/functionality_tests/test_cache_sims.py @@ -0,0 +1,70 @@ +""" +Runs libEnsemble with Latin hypercube sampling on a simple 1D problem + +Execute via one of the following commands (e.g. 3 workers): + mpiexec -np 4 python test_1d_sampling.py + python test_1d_sampling.py --nworkers 3 + python test_1d_sampling.py --nworkers 3 --comms tcp + +The number of concurrent evaluations of the objective function will be 4-1=3. +""" + +# Do not change these lines - they are parsed by run-tests.sh +# TESTSUITE_COMMS: mpi local +# TESTSUITE_NPROCS: 2 4 + +import time + +import numpy as np + +from libensemble.gen_funcs.sampling import latin_hypercube_sample as gen_f + +# Import libEnsemble items for this test +from libensemble.libE import libE +from libensemble.tools import add_unique_random_streams, parse_args, save_libE_output + + +def sim_f(In): + Out = np.zeros(1, dtype=[("f", float)]) + time.sleep(1.1) + Out["f"] = np.linalg.norm(In) + return Out + + +if __name__ == "__main__": + nworkers, is_manager, libE_specs, _ = parse_args() + libE_specs["cache_long_sims"] = True + + sim_specs = { + "sim_f": sim_f, + "in": ["x"], + "out": [("f", float)], + } + + gen_specs = { + "gen_f": gen_f, + "out": [("x", float, (1,))], + "user": { + "gen_batch_size": 10, + "lb": np.array([-3]), + "ub": np.array([3]), + }, + } + + persis_info = add_unique_random_streams({}, nworkers + 1, seed=1234) + + exit_criteria = {"sim_max": 11} + + H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, persis_info, libE_specs=libE_specs) + + if is_manager: + assert len(H) >= 11 + print("\nlibEnsemble with random sampling has generated enough points") + save_libE_output(H, persis_info, __file__, nworkers) + + persis_info = add_unique_random_streams({}, nworkers + 1, seed=1234) + H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, persis_info, libE_specs=libE_specs) + + if is_manager: + # better way of seeing "long" sims not actually taking so long (because of cache?) + assert any(H["sim_ended_time"] - H["sim_started_time"] < 1.1) From f88a25d307bff9f61ed9fe92d9bc81633d832127 Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 22 Oct 2025 14:36:20 -0500 Subject: [PATCH 27/38] non-existing cache already dealt with earlier? --- libensemble/history.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 8c825e325..936f67ae7 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -156,11 +156,7 @@ def _shelf_longrunning_sims(self, index): self.cache_set = True def get_shelved_sims(self) -> npt.NDArray: - try: - in_cache = np.load(self.cache, allow_pickle=True) - except EOFError: - in_cache = np.zeros(1, dtype=self.cache_dtype) - return in_cache + return np.load(self.cache, allow_pickle=True) def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: """ From 6117b948a6b1cd0bfa31f3131d66cba332591f2a Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 23 Oct 2025 09:04:03 -0500 Subject: [PATCH 28/38] add the new libe specs options to libe_specs.rst --- .../libE_specs/libE_specs_general.rst | 14 ++++++++++++++ libensemble/specs.py | 4 ++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/docs/data_structures/libE_specs/libE_specs_general.rst b/docs/data_structures/libE_specs/libE_specs_general.rst index f7f07f75f..88e099036 100644 --- a/docs/data_structures/libE_specs/libE_specs_general.rst +++ b/docs/data_structures/libE_specs/libE_specs_general.rst @@ -38,3 +38,17 @@ General **gen_workers** [list of ints]: List of workers that should run only generators. All other workers will run only simulator functions. + +**cache_long_sims** [bool] = ``False``: + Cache simulation results with runtimes >1s to disk. Subsequent runs of the same + base script with the same command-line arguments will access this cache. + + Upon the generator creating points already in the cache, those points will be skipped from + being sent for evaluation. Instead the corresponding cached results are retrieved and returned + to the generator. + + The cache is saved in ``$HOME/.libE``, and by default is named after the joined command-line arguments. + +**cache_name** [str] = ``"".join(sys.argv)``: + The name of the cache file. Stored in ``$HOME/.libE``, and by default is named after the + joined command-line arguments. diff --git a/libensemble/specs.py b/libensemble/specs.py index df59bbae3..7fda8dd95 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -594,8 +594,8 @@ class LibeSpecs(BaseModel): cache_name: str | None = "".join(sys.argv) """ - The name of the cache file. By default is the joined command-line arguments. - Stored in $HOME/.libE, and by default is named after the joined command-line arguments. + The name of the cache file. Stored in $HOME/.libE, and by default is named after the + joined command-line arguments. """ calc_dir_id_width: int | None = 4 From 7abf447bc89018e5a7492e01e3a2068d392bddc8 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 9 Jan 2026 14:13:32 -0600 Subject: [PATCH 29/38] check that *all* fields in an outbound row match a cache row before copying the cache values into the manager's local record --- libensemble/manager.py | 37 ++++++++++++++++++------------------- 1 file changed, 18 insertions(+), 19 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index 682dd6938..e52831a53 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -425,22 +425,22 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: if "sim_id" not in self.gen_specs["persis_in"]: self.gen_specs["persis_in"].append("sim_id") - def _refresh_from_cache( - self, cache: npt.NDArray, dtype_with_idx: np.dtype, cache_row: npt.NDArray, work_row: int, w: int - ) -> None: - """Add a cache entry, workerID, and H_row to the local record array. + def _check_cache_matches(self, cache_row: npt.NDArray, work_row: int, new_dtype: np.dtype) -> bool: + """Checks if a cache row matches the work row for all outbound fields""" + return all(np.allclose(cache_row[f], self.hist.H[f][work_row]) for f in np.dtype(new_dtype).names) - Later on when we iterate over the cache for entries that could've been sent to a worker (but weren't), - we'll process that entry as though it came from this worker, with these H_rows. - """ - self.cache_hit = True + def _update_local_entry_from_cache( + self, cache_row: npt.NDArray, work_row: int, new_dtype: np.dtype, w: int, dtype_with_idx: np.dtype + ) -> None: + """Updates the local `from_cache` record with the cache row""" from_cache_entry = np.empty(1, dtype=dtype_with_idx) from_cache_entry["H_row"] = work_row from_cache_entry["worker_id"] = w - for remaining_field in cache.dtype.names: - from_cache_entry[remaining_field] = cache_row[remaining_field] + for field in np.dtype(new_dtype).names: + from_cache_entry[field] = cache_row[field] self.from_cache[self.cache_index] = from_cache_entry self.cache_index += 1 + self.cache_hit = True def _cache_scan( self, cache: npt.NDArray, Work: dict, w: int, dtype_with_idx: np.dtype, new_dtype: np.dtype @@ -452,15 +452,14 @@ def _cache_scan( self.cache_timer = Timer() with self.cache_timer: - for field in np.dtype(new_dtype).names: - if field in cache.dtype.names: - for work_row in Work["libE_info"]["H_rows"]: - for cache_row in cache: - if ( - np.allclose(cache_row[field], self.hist.H[field][work_row]) - and work_row not in self.from_cache["H_row"] - ): # we found outbound work in cache, that's not already in the local record - self._refresh_from_cache(cache, dtype_with_idx, cache_row, work_row, w) + for work_row in Work["libE_info"]["H_rows"]: + for cache_row in cache: + if ( + self._check_cache_matches(cache_row, work_row, new_dtype) + and work_row not in self.from_cache["H_row"] + ): # we found outbound work in cache, that's not already in the local record + self._update_local_entry_from_cache(cache_row, work_row, new_dtype, w, dtype_with_idx) + break def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: """Retrieve saved cache from history, create local record-array of matching cache entries. From 42da8c2ba84ce8e9b26841834d419f7203ed67a0 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 9 Jan 2026 14:27:29 -0600 Subject: [PATCH 30/38] don't need cache_index since we're always updating the last row of the local record. most clarifying comments --- libensemble/manager.py | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/libensemble/manager.py b/libensemble/manager.py index e52831a53..6f6a97f1d 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -224,7 +224,6 @@ def __init__( if self.use_cache: self.hist.init_cache(self.libE_specs.get("cache_name")) self.from_cache = [] - self.cache_index = 0 self.cache_hit = False dyn_keys = ("resource_sets", "num_procs", "num_gpus") @@ -426,7 +425,10 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: self.gen_specs["persis_in"].append("sim_id") def _check_cache_matches(self, cache_row: npt.NDArray, work_row: int, new_dtype: np.dtype) -> bool: - """Checks if a cache row matches the work row for all outbound fields""" + """Checks if a cache row matches the work row for all *outbound* *sim* fields + + See _send_work_order for the source of new_dtype - this is the dtype of the outbound sim fields + """ return all(np.allclose(cache_row[f], self.hist.H[f][work_row]) for f in np.dtype(new_dtype).names) def _update_local_entry_from_cache( @@ -434,12 +436,11 @@ def _update_local_entry_from_cache( ) -> None: """Updates the local `from_cache` record with the cache row""" from_cache_entry = np.empty(1, dtype=dtype_with_idx) - from_cache_entry["H_row"] = work_row - from_cache_entry["worker_id"] = w - for field in np.dtype(new_dtype).names: + from_cache_entry["H_row"] = work_row # log this for later checking if outbound rows are already cached + from_cache_entry["worker_id"] = w # used to simulate the worker sending back work that actually came from cache + for field in np.dtype(new_dtype).names: # we now only do this since all outbound fields were close from_cache_entry[field] = cache_row[field] - self.from_cache[self.cache_index] = from_cache_entry - self.cache_index += 1 + self.from_cache[-1] = from_cache_entry # the local record was already appended self.cache_hit = True def _cache_scan( @@ -448,21 +449,24 @@ def _cache_scan( """ Check if any work rows are in the cache, and if so, call the above, _refresh_from_cache to update the local `from_cache` record. + + Each H_row in the work order is checked against the cache, field-wise for closeness. + If a match is found, the local `from_cache` record is updated with the cache row. """ self.cache_timer = Timer() with self.cache_timer: - for work_row in Work["libE_info"]["H_rows"]: + for work_row in Work["libE_info"]["H_rows"]: # used to compare H entries against the cache for cache_row in cache: if ( self._check_cache_matches(cache_row, work_row, new_dtype) and work_row not in self.from_cache["H_row"] ): # we found outbound work in cache, that's not already in the local record self._update_local_entry_from_cache(cache_row, work_row, new_dtype, w, dtype_with_idx) - break + break # we only need to update the local record once def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: - """Retrieve saved cache from history, create local record-array of matching cache entries. + """Retrieve saved cache from history, create local record-array qof matching cache entries. The `from_cache` local record contains cache entries and the workerID and H_rows they are associated with, had they been sent to a worker. @@ -575,7 +579,6 @@ def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: if w > 0: # actual cache entry - not blank. assuming w0 gets no sim work self._handle_msg_from_worker(persis_info, w, process_cache=True) self.from_cache = [] - self.cache_index = 0 # Process messages from workers new_stuff = True From 0ca88ed74e279d98c2d7837e5b931ed1372260f3 Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 1 May 2026 09:45:54 -0500 Subject: [PATCH 31/38] cache-name joined with _. cache lives in memory for duration of run instead of saving/loading each sim. move cache location to HOME/.cache/libensemble. Fix init issue. Fix redundant uniqueness checking --- .../libE_specs/libE_specs_general.rst | 9 ++++-- libensemble/history.py | 29 ++++++++++++------- libensemble/manager.py | 9 +++--- libensemble/specs.py | 11 +++++-- 4 files changed, 37 insertions(+), 21 deletions(-) diff --git a/docs/data_structures/libE_specs/libE_specs_general.rst b/docs/data_structures/libE_specs/libE_specs_general.rst index 88e099036..3c69073ae 100644 --- a/docs/data_structures/libE_specs/libE_specs_general.rst +++ b/docs/data_structures/libE_specs/libE_specs_general.rst @@ -47,8 +47,11 @@ General being sent for evaluation. Instead the corresponding cached results are retrieved and returned to the generator. - The cache is saved in ``$HOME/.libE``, and by default is named after the joined command-line arguments. + The cache is saved in ``cache_dir``, and by default is named after the joined command-line arguments. -**cache_name** [str] = ``"".join(sys.argv)``: - The name of the cache file. Stored in ``$HOME/.libE``, and by default is named after the +**cache_dir** [str] = ``"~/.cache/libensemble"``: + The directory to store the cache file. Defaults to ``~/.cache/libensemble``. + +**cache_name** [str] = ``"_".join(sys.argv)``: + The name of the cache file. Stored in ``cache_dir``, and by default is named after the joined command-line arguments. diff --git a/libensemble/history.py b/libensemble/history.py index 936f67ae7..889177502 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -117,14 +117,18 @@ def __init__( self.last_started = -1 self.last_ended = -1 - def init_cache(self, cache_name: str) -> None: - self.cache_dir = Path.home() / ".libE" + def init_cache(self, cache_name: str, cache_dir: str | Path) -> None: + self.cache_dir = Path(cache_dir).expanduser() self.cache_dir.mkdir(parents=True, exist_ok=True) self.cache = self.cache_dir / Path(cache_name + ".npy") if not self.cache.exists(): self.cache.touch() self.use_cache = True self.cache_set = False + try: + self.in_cache = np.load(self.cache, allow_pickle=True) + except EOFError: + self.in_cache = None def _append_new_fields(self, H_f: npt.NDArray) -> None: dtype_new = np.dtype(list(set(self.H.dtype.descr + np.lib.recfunctions.repack_fields(H_f).dtype.descr))) @@ -144,19 +148,22 @@ def _shelf_longrunning_sims(self, index): self.cache_dtype = sorted( [(name, self.H.dtype.fields[name][0]) for name in self.cache_keys] ) # only needed to init cache - try: - in_cache = np.load(self.cache, allow_pickle=True) - except EOFError: - in_cache = np.zeros(1, dtype=self.cache_dtype) + entry = self.H[index][self.cache_keys] - if entry not in in_cache: - in_cache = np.append(in_cache, entry) - in_cache = np.unique(in_cache, axis=0) # attempt to remove duplicates - np.save(self.cache, in_cache, allow_pickle=True) + + if self.in_cache is None: + self.in_cache = np.array([entry], dtype=self.cache_dtype) + else: + self.in_cache = np.append(self.in_cache, entry) + self.in_cache = np.unique(self.in_cache, axis=0) # attempt to remove duplicates self.cache_set = True + def save_cache(self) -> None: + if self.use_cache and self.cache_set and self.in_cache is not None: + np.save(self.cache, self.in_cache, allow_pickle=True) + def get_shelved_sims(self) -> npt.NDArray: - return np.load(self.cache, allow_pickle=True) + return self.in_cache if self.in_cache is not None else np.load(self.cache, allow_pickle=True) def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: """ diff --git a/libensemble/manager.py b/libensemble/manager.py index 6f6a97f1d..e69a40596 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -222,8 +222,8 @@ def __init__( self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") if self.use_cache: - self.hist.init_cache(self.libE_specs.get("cache_name")) - self.from_cache = [] + self.hist.init_cache(self.libE_specs.get("cache_name"), self.libE_specs.get("cache_dir")) + self.from_cache: Any = None self.cache_hit = False dyn_keys = ("resource_sets", "num_procs", "num_gpus") @@ -483,7 +483,7 @@ def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, n dtype_with_idx = np.dtype(cache.dtype.descr + np.dtype([("H_row", int), ("worker_id", int)]).descr) # initialize or grow the local record, then call _cache_scan to fill it - if not len(self.from_cache): + if self.from_cache is None: self.from_cache = np.zeros(len(work_rows), dtype=dtype_with_idx) else: self.from_cache = np.append(self.from_cache, np.zeros(len(work_rows), dtype=dtype_with_idx)) @@ -578,7 +578,7 @@ def _receive_from_workers_or_cache(self, persis_info: dict) -> dict: for w in self.from_cache["worker_id"]: if w > 0: # actual cache entry - not blank. assuming w0 gets no sim work self._handle_msg_from_worker(persis_info, w, process_cache=True) - self.from_cache = [] + self.from_cache = None # Process messages from workers new_stuff = True @@ -874,6 +874,7 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: logger.error(traceback.format_exc()) raise LoggedException(e.args) from None finally: + self.hist.save_cache() # Return persis_info, exit_flag, elapsed time result = self._final_receive_and_kill(persis_info) self.wcomms = [] diff --git a/libensemble/specs.py b/libensemble/specs.py index 7fda8dd95..fad0235dc 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -589,12 +589,17 @@ class LibeSpecs(BaseModel): being sent for evaluation. Instead the corresponding cached results are retrieved and returned to the generator. - The cache is saved in $HOME/.libE, and by default is named after the joined command-line arguments. + The cache is saved in cache_dir, and by default is named after the joined command-line arguments. """ - cache_name: str | None = "".join(sys.argv) + cache_dir: str | Path | None = str(Path.home() / ".cache" / "libensemble") """ - The name of the cache file. Stored in $HOME/.libE, and by default is named after the + The directory to store the cache file. Defaults to `~/.cache/libensemble`. + """ + + cache_name: str | None = "_".join(sys.argv) + """ + The name of the cache file. Stored in cache_dir, and by default is named after the joined command-line arguments. """ From 1af71dbdfe4dc46362b66d28dcdbd549d5f46369 Mon Sep 17 00:00:00 2001 From: jlnav Date: Mon, 4 May 2026 13:30:05 -0500 Subject: [PATCH 32/38] update functionality test for develop --- .../tests/functionality_tests/test_cache_sims.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/libensemble/tests/functionality_tests/test_cache_sims.py b/libensemble/tests/functionality_tests/test_cache_sims.py index 885a6556e..4f3ca993e 100644 --- a/libensemble/tests/functionality_tests/test_cache_sims.py +++ b/libensemble/tests/functionality_tests/test_cache_sims.py @@ -17,11 +17,12 @@ import numpy as np +from libensemble.alloc_funcs.give_sim_work_first import give_sim_work_first from libensemble.gen_funcs.sampling import latin_hypercube_sample as gen_f # Import libEnsemble items for this test from libensemble.libE import libE -from libensemble.tools import add_unique_random_streams, parse_args, save_libE_output +from libensemble.tools import parse_args, save_libE_output def sim_f(In): @@ -44,26 +45,25 @@ def sim_f(In): gen_specs = { "gen_f": gen_f, "out": [("x", float, (1,))], + "batch_size": 10, "user": { - "gen_batch_size": 10, "lb": np.array([-3]), "ub": np.array([3]), }, } - persis_info = add_unique_random_streams({}, nworkers + 1, seed=1234) + alloc_specs = {"alloc_f": give_sim_work_first} exit_criteria = {"sim_max": 11} - H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, persis_info, libE_specs=libE_specs) + H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, alloc_specs=alloc_specs, libE_specs=libE_specs) if is_manager: assert len(H) >= 11 print("\nlibEnsemble with random sampling has generated enough points") save_libE_output(H, persis_info, __file__, nworkers) - persis_info = add_unique_random_streams({}, nworkers + 1, seed=1234) - H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, persis_info, libE_specs=libE_specs) + H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, alloc_specs=alloc_specs, libE_specs=libE_specs) if is_manager: # better way of seeing "long" sims not actually taking so long (because of cache?) From 7e0643258c9e4393b4f6dde527535d56d26bc76e Mon Sep 17 00:00:00 2001 From: jlnav Date: Fri, 8 May 2026 08:13:28 -0500 Subject: [PATCH 33/38] most importantly, specify seed to gen --- libensemble/tests/functionality_tests/test_cache_sims.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/libensemble/tests/functionality_tests/test_cache_sims.py b/libensemble/tests/functionality_tests/test_cache_sims.py index 4f3ca993e..d93907203 100644 --- a/libensemble/tests/functionality_tests/test_cache_sims.py +++ b/libensemble/tests/functionality_tests/test_cache_sims.py @@ -35,6 +35,7 @@ def sim_f(In): if __name__ == "__main__": nworkers, is_manager, libE_specs, _ = parse_args() libE_specs["cache_long_sims"] = True + libE_specs["cache_dir"] = "." sim_specs = { "sim_f": sim_f, @@ -49,6 +50,7 @@ def sim_f(In): "user": { "lb": np.array([-3]), "ub": np.array([3]), + "gen_seed": 42, }, } @@ -67,4 +69,5 @@ def sim_f(In): if is_manager: # better way of seeing "long" sims not actually taking so long (because of cache?) - assert any(H["sim_ended_time"] - H["sim_started_time"] < 1.1) + durations = H["sim_ended_time"] - H["sim_started_time"] + assert any((durations < 1.1) & (durations != -np.inf)) From fe51a5c3ecbb0eb8e13698ddd0c281f859860758 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 21 May 2026 10:38:48 -0500 Subject: [PATCH 34/38] cache is computed based on a hash of all the specs, all the callables, and H0. cache rebuilt upon changes to those objects --- libensemble/history.py | 37 +++++++++++- libensemble/libE.py | 17 +++++- libensemble/manager.py | 8 ++- libensemble/specs.py | 15 ++--- libensemble/utils/misc.py | 116 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 13 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 889177502..b82e3f90f 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -1,3 +1,4 @@ +import json import logging import time from pathlib import Path @@ -117,14 +118,41 @@ def __init__( self.last_started = -1 self.last_ended = -1 - def init_cache(self, cache_name: str, cache_dir: str | Path) -> None: + def init_cache( + self, + cache_name: str, + cache_dir: str | Path, + spec_hash: str | None = None, + ) -> None: self.cache_dir = Path(cache_dir).expanduser() self.cache_dir.mkdir(parents=True, exist_ok=True) self.cache = self.cache_dir / Path(cache_name + ".npy") - if not self.cache.exists(): - self.cache.touch() + self.cache_meta = self.cache_dir / Path(cache_name + ".meta.json") + self.spec_hash = spec_hash self.use_cache = True self.cache_set = False + + # Validate any existing cache against the configuration hash. + cache_valid = False + if self.cache.exists(): + if self.cache_meta.exists(): + try: + with open(self.cache_meta) as f: + meta = json.load(f) + if meta.get("spec_hash") == spec_hash: + cache_valid = True + except (json.JSONDecodeError, KeyError): + pass + if not cache_valid: + logger.debug( + "Cache hash mismatch or missing metadata — starting fresh: %s", + self.cache.name, + ) + self.cache.unlink(missing_ok=True) + + if not self.cache.exists(): + self.cache.touch() + try: self.in_cache = np.load(self.cache, allow_pickle=True) except EOFError: @@ -161,6 +189,9 @@ def _shelf_longrunning_sims(self, index): def save_cache(self) -> None: if self.use_cache and self.cache_set and self.in_cache is not None: np.save(self.cache, self.in_cache, allow_pickle=True) + if self.spec_hash: + with open(self.cache_meta, "w") as f: + json.dump({"spec_hash": self.spec_hash}, f) def get_shelved_sims(self) -> npt.NDArray: return self.in_cache if self.in_cache is not None else np.load(self.cache, allow_pickle=True) diff --git a/libensemble/libE.py b/libensemble/libE.py index 091a79e7f..b7a0daccd 100644 --- a/libensemble/libE.py +++ b/libensemble/libE.py @@ -137,7 +137,7 @@ from libensemble.tools.alloc_support import AllocSupport from libensemble.tools.tools import _USER_SIM_ID_WARNING from libensemble.utils import launcher -from libensemble.utils.misc import specs_dump +from libensemble.utils.misc import compute_config_hash, specs_dump from libensemble.utils.timer import Timer from libensemble.version import __version__ from libensemble.worker import worker_main @@ -235,12 +235,27 @@ def libE( exit_criteria=exit_criteria, ) + # Compute a deterministic hash of the full configuration for cache integrity. + spec_hash = compute_config_hash( + sim_specs=ensemble.sim_specs, + gen_specs=ensemble.gen_specs, + alloc_specs=ensemble.alloc_specs, + libE_specs=ensemble.libE_specs, + exit_criteria=ensemble.exit_criteria, + H0=H0, + ) + (sim_specs, gen_specs, alloc_specs, libE_specs) = [ specs_dump(spec, by_alias=True) for spec in [ensemble.sim_specs, ensemble.gen_specs, ensemble.alloc_specs, ensemble.libE_specs] ] exit_criteria = specs_dump(ensemble.exit_criteria, by_alias=True, exclude_none=True) + # Inject spec hash and auto-generate cache name when not explicitly provided. + libE_specs["_spec_hash"] = spec_hash + if libE_specs.get("cache_long_sims") and not libE_specs.get("cache_name"): + libE_specs["cache_name"] = f".libe_cache_{spec_hash[:16]}" + # Restore objects that don't survive serialization via model_dump if hasattr(ensemble.sim_specs, "simulator") and ensemble.sim_specs.simulator is not None: sim_specs["simulator"] = ensemble.sim_specs.simulator diff --git a/libensemble/manager.py b/libensemble/manager.py index e69a40596..27aac2d14 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -222,7 +222,11 @@ def __init__( self.persis_pending: list[int] = [] self.live_data = libE_specs.get("live_data") if self.use_cache: - self.hist.init_cache(self.libE_specs.get("cache_name"), self.libE_specs.get("cache_dir")) + self.hist.init_cache( + self.libE_specs.get("cache_name"), + self.libE_specs.get("cache_dir"), + spec_hash=self.libE_specs.get("_spec_hash"), + ) self.from_cache: Any = None self.cache_hit = False @@ -252,7 +256,7 @@ def __init__( self.wcomms = [local_worker_comm] + self.wcomms self.W = _WorkerIndexer(self.W, 1 - gen_on_worker) # if gen on worker, then no additional worker - self.wcomms = _WorkerIndexer(self.wcomms, 1 - gen_on_worker) + self.wcomms = _WorkerIndexer(self.wcomms, 1 - gen_on_worker) # type: ignore[assignment] temp_EnsembleDirectory = EnsembleDirectory(libE_specs=libE_specs) self.resources = Resources.resources diff --git a/libensemble/specs.py b/libensemble/specs.py index fad0235dc..5476fafbc 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -1,5 +1,4 @@ import random -import sys import warnings from pathlib import Path @@ -582,14 +581,15 @@ class LibeSpecs(BaseModel): cache_long_sims: bool | None = False """ - Cache simulation results with runtimes >1s to disk. Subsequent runs of the same - base script with the same command-line arguments will access this cache. + Cache simulation results with runtimes >1s to disk. Subsequent runs with an + identical configuration (specs, callables, H0) will access this cache. Upon the generator creating points already in the cache, those points will be skipped from being sent for evaluation. Instead the corresponding cached results are retrieved and returned to the generator. - The cache is saved in cache_dir, and by default is named after the joined command-line arguments. + The cache is saved in ``cache_dir``. When ``cache_name`` is ``None``, the filename is + automatically derived from a SHA-256 hash of the full ensemble configuration. """ cache_dir: str | Path | None = str(Path.home() / ".cache" / "libensemble") @@ -597,10 +597,11 @@ class LibeSpecs(BaseModel): The directory to store the cache file. Defaults to `~/.cache/libensemble`. """ - cache_name: str | None = "_".join(sys.argv) + cache_name: str | None = None """ - The name of the cache file. Stored in cache_dir, and by default is named after the - joined command-line arguments. + The name of the cache file. Stored in cache_dir. + When ``None`` and ``cache_long_sims`` is ``True``, a name is automatically + derived from a SHA-256 hash of the full ensemble configuration. """ calc_dir_id_width: int | None = 4 diff --git a/libensemble/utils/misc.py b/libensemble/utils/misc.py index da709f1c0..73b618b0c 100644 --- a/libensemble/utils/misc.py +++ b/libensemble/utils/misc.py @@ -2,6 +2,9 @@ Misc internal functions """ +import hashlib +import inspect +import json from itertools import chain, groupby from operator import itemgetter @@ -9,6 +12,119 @@ import numpy.typing as npt +def _get_callable_source(obj) -> str: + """Get source code for a function or callable object. + + Tries ``inspect.getsource`` on the object directly, then on its class. + Falls back to ``name.module`` when source is unavailable. + """ + if obj is None: + return "" + for target in (obj, type(obj)): + try: + return inspect.getsource(target) + except (TypeError, OSError): + continue + name = getattr(obj, "__name__", type(obj).__name__) + module = getattr(obj, "__module__", type(obj).__module__) + return f"{module}.{name}" + + +def compute_config_hash( + sim_specs, + gen_specs, + alloc_specs=None, + libE_specs=None, + exit_criteria=None, + H0=None, +) -> str: + """Compute a deterministic SHA-256 hash of the full ensemble configuration. + + All Pydantic spec models are dumped to stable dictionaries. Callables + (``sim_f``, ``gen_f``, ``simulator``, ``generator``, ``alloc_f``) are + replaced with their source code (or ``module.name`` fallback) so that + code changes invalidate the cache. ``H0`` data is also included. + + Parameters + ---------- + sim_specs : SimSpecs + gen_specs : GenSpecs | None + alloc_specs : AllocSpecs + libE_specs : LibeSpecs + exit_criteria : ExitCriteria + H0 : numpy.ndarray | None + + Returns + ------- + str + 64-character hex digest. + """ + spec_dicts: dict = {} + + def _dump(spec, **kwargs): + return spec.model_dump(**kwargs) + + def _dump_or_empty(spec, **kwargs): + return _dump(spec, **kwargs) if spec is not None else {} + + spec_dicts["sim"] = _dump(sim_specs, by_alias=True, exclude_none=True, exclude_defaults=True) + spec_dicts["gen"] = _dump_or_empty(gen_specs, by_alias=True, exclude_none=True, exclude_defaults=True) + spec_dicts["alloc"] = _dump_or_empty(alloc_specs, by_alias=True, exclude_none=True, exclude_defaults=True) + spec_dicts["exit"] = _dump_or_empty(exit_criteria, by_alias=True, exclude_none=True) + + libE_dict = _dump_or_empty(libE_specs, by_alias=True, exclude_none=True, exclude_defaults=True) + for key in ("cache_long_sims", "cache_dir", "cache_name"): + libE_dict.pop(key, None) + spec_dicts["libE"] = libE_dict + + # Hash callable sources, then strip raw objects so memory addresses + # don't leak into the JSON serialization. + _strip_raw_objects(spec_dicts) + _add_callable_sources(spec_dicts, sim_specs, gen_specs, alloc_specs) + + # Hash H0 data + if H0 is not None and len(H0): + spec_dicts["H0_hash"] = hashlib.sha256(H0.tobytes()).hexdigest() + + serialized = json.dumps(spec_dicts, sort_keys=True, default=str) + return hashlib.sha256(serialized.encode()).hexdigest() + + +_RAW_OBJECT_FIELDS = { + "sim": {"sim_f", "simulator", "vocs"}, + "gen": {"gen_f", "generator", "vocs"}, + "alloc": {"alloc_f"}, +} + + +def _strip_raw_objects(spec_dicts): + """Remove object-typed fields from dumped spec dicts. + + These are handled separately via source extraction to avoid + non-deterministic memory-address-based serialization. + """ + for key, fields in _RAW_OBJECT_FIELDS.items(): + if key in spec_dicts: + for field in fields: + spec_dicts[key].pop(field, None) + + +def _add_callable_sources(spec_dicts, sim_specs, gen_specs, alloc_specs): + """Extract source for callables and store them in spec_dicts.""" + for spec_name, spec, field in [ + ("sim", sim_specs, "sim_f"), + ("sim", sim_specs, "simulator"), + ("gen", gen_specs, "gen_f"), + ("gen", gen_specs, "generator"), + ("alloc", alloc_specs, "alloc_f"), + ]: + if spec is None: + continue + obj = getattr(spec, field, None) + if obj is not None: + spec_dicts[f"{spec_name}_source_{field}"] = _get_callable_source(obj) + + def extract_H_ranges(Work: dict) -> str: """Convert received H_rows into ranges for labeling""" work_H_rows = Work["libE_info"]["H_rows"] From 0ed23d2a49853d5c2e07af7b814932ea8fccc6e6 Mon Sep 17 00:00:00 2001 From: jlnav Date: Thu, 21 May 2026 13:12:47 -0500 Subject: [PATCH 35/38] bump pixi versions in ci yml files, formatting? --- .github/workflows/extra.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/extra.yml b/.github/workflows/extra.yml index 07d47d812..9f07e4d8f 100644 --- a/.github/workflows/extra.yml +++ b/.github/workflows/extra.yml @@ -76,6 +76,7 @@ jobs: rm ./libensemble/tests/unit_tests/test_ufunc_runners.py # needs globus-compute rm ./libensemble/tests/regression_tests/test_gpCAM.py # needs gpcam, which doesn't build on 3.13 rm ./libensemble/tests/regression_tests/test_asktell_gpCAM.py # needs gpcam, which doesn't build on 3.13 + rm ./libensemble/tests/regression_tests/test_persistent_gp_multitask_ax.py # needs ax-platform, which doesn't yet support 3.14 rm ./libensemble/tests/regression_tests/test_optimas_ax_mf.py # needs ax-platform, which doesn't yet support 3.14 rm ./libensemble/tests/regression_tests/test_optimas_ax_sf.py # needs ax-platform, which doesn't yet support 3.14 From 8b2da0c72229aa47a6bc598d54b42ff65feb5246 Mon Sep 17 00:00:00 2001 From: jlnav Date: Tue, 26 May 2026 15:41:27 -0500 Subject: [PATCH 36/38] various speed optimizations, especially around the caching sections --- .../alloc_funcs/start_only_persistent.py | 11 +- libensemble/history.py | 158 +++++++++++++----- libensemble/manager.py | 48 ++++-- 3 files changed, 156 insertions(+), 61 deletions(-) diff --git a/libensemble/alloc_funcs/start_only_persistent.py b/libensemble/alloc_funcs/start_only_persistent.py index 6b02b4c60..e97383f0a 100644 --- a/libensemble/alloc_funcs/start_only_persistent.py +++ b/libensemble/alloc_funcs/start_only_persistent.py @@ -68,17 +68,20 @@ def only_persistent_gens(W, H, sim_specs, gen_specs, alloc_specs, persis_info, l gen_count = support.count_persis_gens() Work = {} - # Asynchronous return to generator - async_return = user.get("async_return", False) and sum(H["sim_ended"]) >= initial_batch_size + # Asynchronous return to generator. + # Use the manager-maintained counter instead of re-scanning the full H array. + async_return = user.get("async_return", False) and libE_info["sim_ended_count"] >= initial_batch_size if gen_count < persis_info.get("num_gens_started", 0): # When a persistent worker is done, trigger a shutdown (returning exit condition of 1) return Work, persis_info, 1 - # Give evaluated results back to a running persistent gen + # Give evaluated results back to a running persistent gen. + # Compute the sim_ended & ~gen_informed mask once; AND with per-worker gen_inds inside the loop. + pending_sim = H["sim_ended"] & ~H["gen_informed"] for wid in support.avail_worker_ids(persistent=EVAL_GEN_TAG, active_recv=active_recv_gen): gen_inds = H["gen_worker"] == wid - returned_but_not_given = np.logical_and.reduce((H["sim_ended"], ~H["gen_informed"], gen_inds)) + returned_but_not_given = pending_sim & gen_inds if np.any(returned_but_not_given): if async_return or support.all_sim_ended(H, gen_inds): point_ids = np.where(returned_but_not_given)[0] diff --git a/libensemble/history.py b/libensemble/history.py index b82e3f90f..361df498e 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -132,6 +132,16 @@ def init_cache( self.use_cache = True self.cache_set = False + # Precompute the sorted user-field names and their dtypes once, so + # _shelf_longrunning_sims doesn't recompute them on every sim return. + libE_field_names = {k[0] for k in libE_fields} + self.cache_keys = sorted([n for n in self.H.dtype.names if n not in libE_field_names]) + self.cache_dtype = np.dtype(sorted([(n, self.H.dtype.fields[n][0]) for n in self.cache_keys])) + + # Buffer for new entries collected during this run; deduplicated via bytes key. + self._cache_buffer: list = [] + self._cache_seen: set = set() + # Validate any existing cache against the configuration hash. cache_valid = False if self.cache.exists(): @@ -158,6 +168,14 @@ def init_cache( except EOFError: self.in_cache = None + # Pre-populate the seen-set from any on-disk entries so we don't re-add them. + # Also mark cache_set=True immediately when there is existing data — the manager + # uses this flag to decide whether to scan the cache when dispatching sim work. + if self.in_cache is not None and len(self.in_cache) > 0: + for row in self.in_cache: + self._cache_seen.add(row.tobytes()) + self.cache_set = True + def _append_new_fields(self, H_f: npt.NDArray) -> None: dtype_new = np.dtype(list(set(self.H.dtype.descr + np.lib.recfunctions.repack_fields(H_f).dtype.descr))) H_new = np.zeros(len(self.H), dtype=dtype_new) @@ -167,34 +185,80 @@ def _append_new_fields(self, H_f: npt.NDArray) -> None: self.H = H_new def _shelf_longrunning_sims(self, index): - """Cache any f values that ran for more than a second.""" - if self.H[index]["sim_ended_time"] - self.H[index]["sim_started_time"] > 1: - # ('f', 'x') and ('x', 'f') are not equivalent dtypes, unfortunately. So maybe sorted helps. - self.cache_keys = sorted( - [i for i in self.H.dtype.names if i not in [k[0] for k in libE_fields]] - ) # ('f', 'x') keys only - self.cache_dtype = sorted( - [(name, self.H.dtype.fields[name][0]) for name in self.cache_keys] - ) # only needed to init cache - - entry = self.H[index][self.cache_keys] - - if self.in_cache is None: - self.in_cache = np.array([entry], dtype=self.cache_dtype) - else: - self.in_cache = np.append(self.in_cache, entry) - self.in_cache = np.unique(self.in_cache, axis=0) # attempt to remove duplicates - self.cache_set = True + """Cache any f values that ran for more than a second. + + Uses a bytes-keyed set for O(1) deduplication instead of np.unique on + every insertion, and accumulates new entries in a plain Python list that + is only materialised into a structured array at save_cache() time. + """ + if self.H[index]["sim_ended_time"] - self.H[index]["sim_started_time"] <= 1: + return + entry = np.array([self.H[index][self.cache_keys]], dtype=self.cache_dtype) + key = entry[0].tobytes() + if key in self._cache_seen: + return + self._cache_seen.add(key) + self._cache_buffer.append(entry) + self.cache_set = True + + def _materialize_cache(self) -> npt.NDArray | None: + """Combine the on-disk cache with any buffered new entries into one array.""" + parts = [] + if self.in_cache is not None: + parts.append(self.in_cache) + if self._cache_buffer: + parts.append(np.concatenate(self._cache_buffer)) + if not parts: + return None + return np.concatenate(parts) if len(parts) > 1 else parts[0] def save_cache(self) -> None: - if self.use_cache and self.cache_set and self.in_cache is not None: - np.save(self.cache, self.in_cache, allow_pickle=True) - if self.spec_hash: - with open(self.cache_meta, "w") as f: - json.dump({"spec_hash": self.spec_hash}, f) + if self.use_cache and self.cache_set: + combined = self._materialize_cache() + if combined is not None: + np.save(self.cache, combined, allow_pickle=True) + if self.spec_hash: + with open(self.cache_meta, "w") as f: + json.dump({"spec_hash": self.spec_hash}, f) def get_shelved_sims(self) -> npt.NDArray: - return self.in_cache if self.in_cache is not None else np.load(self.cache, allow_pickle=True) + combined = self._materialize_cache() + return combined if combined is not None else np.load(self.cache, allow_pickle=True) + + @staticmethod + def _classify_fields(fields, returned_H, H): + """Partition returned fields into three buckets for update_history_f. + + Returns + ------- + scalar_fields : list[str] + Fields whose per-row value is a scalar or object (can be assigned + with a single fancy-indexed write across all rows). + uniform_fields : list[str] + Fixed-shape array fields whose shape exactly matches H's storage + shape (can also be assigned in one fancy-indexed write). + ragged_fields : list[str] + Fixed-shape array fields that are *smaller* than H's storage shape + (need per-row slice assignment). + """ + scalar_fields = [] + uniform_fields = [] + ragged_fields = [] + for field in fields: + if field in protected_libE_fields: + continue + dt = returned_H.dtype[field] + if dt.shape == () or dt.hasobject: + scalar_fields.append(field) + else: + # Compare element shape: returned vs H's allocated shape + h_shape = H.dtype[field].shape + r_shape = dt.shape + if r_shape == h_shape: + uniform_fields.append(field) + else: + ragged_fields.append(field) + return scalar_fields, uniform_fields, ragged_fields def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: """ @@ -208,31 +272,41 @@ def update_history_f(self, D: dict, kill_canceled_sims: bool = False) -> None: if returned_H is not None and any([field not in self.H.dtype.names for field in returned_H.dtype.names]): self._append_new_fields(returned_H) - for j, ind in enumerate(new_inds): + if self.safe_mode: for field in fields: - if field in protected_libE_fields: - if self.safe_mode: - assert False, "The field '" + field + "' is protected" - continue + assert field not in protected_libE_fields, "The field '" + field + "' is protected" - if np.isscalar(returned_H[field][j]) or returned_H.dtype[field].hasobject: - self.H[field][ind] = returned_H[field][j] - else: - # len or np.size + new_inds = np.asarray(new_inds) + + if fields and returned_H is not None: + scalar_fields, uniform_fields, ragged_fields = self._classify_fields(fields, returned_H, self.H) + + # Vectorized assignment for scalar and object fields (one op per field) + for field in scalar_fields: + self.H[field][new_inds] = returned_H[field] + + # Vectorized assignment for fixed-shape array fields that exactly match H's shape + for field in uniform_fields: + self.H[field][new_inds] = returned_H[field] + + # Per-row loop only for ragged (partial-fill) array fields + for j, ind in enumerate(new_inds): + for field in ragged_fields: H0_size = len(returned_H[field][j]) assert H0_size <= len(self.H[field][ind]), ( "History update Error: Too many values received for " + field ) assert H0_size, "History update Error: No values in this field " + field - if H0_size == len(self.H[field][ind]): - self.H[field][ind] = returned_H[field][j] # ref - else: - self.H[field][ind][:H0_size] = returned_H[field][j] # Slice View - - self.H["sim_ended"][ind] = True - self.H["sim_ended_time"][ind] = time.time() - self.sim_ended_count += 1 - if self.use_cache: + self.H[field][ind][:H0_size] = returned_H[field][j] + + # Batch-update bookkeeping fields for all returned rows at once + t = time.time() + self.H["sim_ended"][new_inds] = True + self.H["sim_ended_time"][new_inds] = t + self.sim_ended_count += len(new_inds) + + if self.use_cache: + for ind in new_inds: self._shelf_longrunning_sims(ind) if kill_canceled_sims: diff --git a/libensemble/manager.py b/libensemble/manager.py index 27aac2d14..53a136fe5 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -428,12 +428,28 @@ def _ensure_sim_id_in_persis_in(self, D: npt.NDArray) -> None: if "sim_id" not in self.gen_specs["persis_in"]: self.gen_specs["persis_in"].append("sim_id") - def _check_cache_matches(self, cache_row: npt.NDArray, work_row: int, new_dtype: np.dtype) -> bool: - """Checks if a cache row matches the work row for all *outbound* *sim* fields + def _find_cache_match(self, work_row: int, cache: npt.NDArray, new_dtype: np.dtype) -> int: + """Return the index of the first cache row that matches the work row for all outbound sim fields. - See _send_work_order for the source of new_dtype - this is the dtype of the outbound sim fields + Vectorizes the comparison across the full cache axis per field, short-circuiting as soon + as any field eliminates all remaining candidates. Returns -1 when no match is found. """ - return all(np.allclose(cache_row[f], self.hist.H[f][work_row]) for f in np.dtype(new_dtype).names) + mask = np.ones(len(cache), dtype=bool) + for f in np.dtype(new_dtype).names: + cf = cache[f] + hf = self.hist.H[f][work_row] + try: + if cf.ndim == 1: + mask &= np.isclose(cf, hf) + else: + # multi-dim field: reduce over all axes except the cache axis + mask &= np.all(np.isclose(cf, hf), axis=tuple(range(1, cf.ndim))) + except (TypeError, ValueError): + # object or non-numeric dtype: fall back to element-wise equality + mask &= np.array([c == hf for c in cf]) + if not mask.any(): + return -1 + return int(np.argmax(mask)) if mask.any() else -1 def _update_local_entry_from_cache( self, cache_row: npt.NDArray, work_row: int, new_dtype: np.dtype, w: int, dtype_with_idx: np.dtype @@ -451,23 +467,25 @@ def _cache_scan( self, cache: npt.NDArray, Work: dict, w: int, dtype_with_idx: np.dtype, new_dtype: np.dtype ) -> None: """ - Check if any work rows are in the cache, and if so, call the above, _refresh_from_cache + Check if any work rows are in the cache, and if so, call _update_local_entry_from_cache to update the local `from_cache` record. - Each H_row in the work order is checked against the cache, field-wise for closeness. - If a match is found, the local `from_cache` record is updated with the cache row. + Each H_row in the work order is compared against the full cache array in a vectorized + manner (one NumPy operation per field across all cache rows) rather than iterating over + cache rows one at a time. A set tracks already-matched rows for O(1) membership tests. """ - self.cache_timer = Timer() + # worker_id == 0 means an uninitialised (blank) slot; filter those out. + # H_row 0 is a valid row index so we cannot use >= 0 as the sentinel. + seen_rows: set[int] = set(self.from_cache["H_row"][self.from_cache["worker_id"] > 0]) with self.cache_timer: for work_row in Work["libE_info"]["H_rows"]: # used to compare H entries against the cache - for cache_row in cache: - if ( - self._check_cache_matches(cache_row, work_row, new_dtype) - and work_row not in self.from_cache["H_row"] - ): # we found outbound work in cache, that's not already in the local record - self._update_local_entry_from_cache(cache_row, work_row, new_dtype, w, dtype_with_idx) - break # we only need to update the local record once + if work_row in seen_rows: + continue + match_idx = self._find_cache_match(work_row, cache, new_dtype) + if match_idx >= 0: + self._update_local_entry_from_cache(cache[match_idx], work_row, new_dtype, w, dtype_with_idx) + seen_rows.add(work_row) def _update_state_from_cache(self, Work: dict, work_rows: npt.NDArray, w: int, new_dtype: np.dtype) -> None: """Retrieve saved cache from history, create local record-array qof matching cache entries. From e88228ee1eb463b7d23e423a26893c9466e6eccd Mon Sep 17 00:00:00 2001 From: jlnav Date: Wed, 27 May 2026 12:43:14 -0500 Subject: [PATCH 37/38] adjust tolerance again because of flakiness of ci... --- libensemble/tests/regression_tests/test_aposmm_nlopt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libensemble/tests/regression_tests/test_aposmm_nlopt.py b/libensemble/tests/regression_tests/test_aposmm_nlopt.py index 40ebcc497..97958728d 100644 --- a/libensemble/tests/regression_tests/test_aposmm_nlopt.py +++ b/libensemble/tests/regression_tests/test_aposmm_nlopt.py @@ -95,7 +95,7 @@ def six_hump_camel_func(x): print("[Manager]:", H[np.where(H["local_min"])]["x"]) print("[Manager]: Time taken =", time() - start_time, flush=True) - tol = 1e-5 + tol = 1e-4 for m in minima: # The minima are known on this test problem. # We use their values to test APOSMM has identified all minima From 68dbe87a37e3df30d933c74f4ad32d23a90407aa Mon Sep 17 00:00:00 2001 From: jlnav Date: Tue, 8 Sep 2026 12:39:56 -0500 Subject: [PATCH 38/38] small deps and attribute bugfixes --- libensemble/history.py | 1 + libensemble/tests/unit_tests/test_history.py | 1 + pixi.lock | 4 ++-- pyproject.toml | 3 --- 4 files changed, 4 insertions(+), 5 deletions(-) diff --git a/libensemble/history.py b/libensemble/history.py index 361df498e..00834ce84 100644 --- a/libensemble/history.py +++ b/libensemble/history.py @@ -105,6 +105,7 @@ def __init__( self.grow_count = 0 self.safe_mode = False self.use_cache = False + self.cache_set = False self.sim_started_count = np.sum(H["sim_started"]) self.sim_ended_count = np.sum(H["sim_ended"]) diff --git a/libensemble/tests/unit_tests/test_history.py b/libensemble/tests/unit_tests/test_history.py index b6bba140e..375a15c54 100644 --- a/libensemble/tests/unit_tests/test_history.py +++ b/libensemble/tests/unit_tests/test_history.py @@ -92,6 +92,7 @@ def test_hist_init_1(): assert hist.index == 0 assert hist.sim_ended_count == 0 assert hist.gen_informed_count == 0 + assert hist.cache_set is False def test_hist_init_1A_H0(): diff --git a/pixi.lock b/pixi.lock index 4ffdd63bc..51dcebc00 100644 --- a/pixi.lock +++ b/pixi.lock @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:17f90f7745889e8d6d423a8aa93c74ef9787036f95086ce5ec38b983de3d36f4 -size 1245374 +oid sha256:3154573cf509ae2670a2fd34cabc674d10bf052d91636e64432ae194b947a01c +size 1248259 diff --git a/pyproject.toml b/pyproject.toml index bef73a62e..7f38c2e78 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -186,9 +186,6 @@ setuptools = ">=80.8.0,<81" numpy = ">=2.2.6,<3" pydantic = ">=2.12.4,<3" gest-api = ">=0.1,<0.2" -pyyaml = ">=6.0,<7" -tomli = ">=1.2.1,<3" -psutil = ">=5.9.4,<7" # macOS dependencies