diff --git a/.github/workflows/basic.yml b/.github/workflows/basic.yml index 25022fd107..b79fae37b9 100644 --- a/.github/workflows/basic.yml +++ b/.github/workflows/basic.yml @@ -31,6 +31,7 @@ jobs: env: HYDRA_LAUNCHER: "fork" + MPIR_CVAR_CH4_NETMOD: "ofi" TERM: xterm-256color GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.github/workflows/extra.yml b/.github/workflows/extra.yml index e9d45c20d7..4022c3bef3 100644 --- a/.github/workflows/extra.yml +++ b/.github/workflows/extra.yml @@ -33,6 +33,7 @@ jobs: env: HYDRA_LAUNCHER: "fork" + MPIR_CVAR_CH4_NETMOD: "ofi" TERM: xterm-256color GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/docs/advanced_installation/advanced_installation.rst b/docs/advanced_installation/advanced_installation.rst index 43d4ee7384..0ac8018ffb 100644 --- a/docs/advanced_installation/advanced_installation.rst +++ b/docs/advanced_installation/advanced_installation.rst @@ -1,7 +1,7 @@ Advanced Installation ===================== -`pip `__ \|\| `uv `__ \|\| `pixi `__ \|\| `conda `__ \|\| `Spack `__ +`pip `__ || `uv `__ || `pixi `__ || `conda `__ || `Spack `__ libEnsemble can be installed from ``pip``, ``uv``, ``pixi``, ``Conda``, or ``Spack``. @@ -31,7 +31,18 @@ Further recommendations for selected HPC systems are given in the Globus Compute -------------- -`Globus Compute`_ may be installed optionally to submit simulation function instances to remote Globus Compute endpoints. +`Globus Compute`_ may be installed optionally to submit simulation function +instances to remote Globus Compute endpoints:: + + pip install globus-compute-sdk + +This is an optional dependency; libEnsemble operates normally without it. +If Globus Compute is not installed and a ``globus_compute_endpoint`` is +configured, libEnsemble will warn and fall back to local execution. + +See :ref:`Globus Compute - Remote User Functions` for +usage, and the :doc:`GlobusComputeExecutor API reference` +for the full executor interface. .. _Globus Compute: https://www.globus.org/compute .. _Python: https://www.python.org/ diff --git a/docs/conf.py b/docs/conf.py index 8647bbf4f7..2d227780e1 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -31,7 +31,7 @@ def __getattr__(cls, name): return MagicMock() -autodoc_mock_imports = ["ax", "gpcam", "IPython", "matplotlib", "pandas", "scipy", "surmise"] +autodoc_mock_imports = ["ax", "globus_compute_sdk", "gpcam", "IPython", "matplotlib", "pandas", "scipy", "surmise"] MOCK_MODULES = [ "argparse", @@ -135,7 +135,7 @@ class AxParameterWarning(Warning): # Ensure it's a real warning subclass # The suffix(es) of source filenames. # You can specify multiple suffix as a list of string: # -# source_suffix = ['.rst', '.md'] +# source_suffix = ['.md', '.rst'] source_suffix = ".rst" # The master toctree document. @@ -205,7 +205,7 @@ class AxParameterWarning(Warning): # Ensure it's a real warning subclass html_favicon = "./images/libE_logo_circle.png" html_title = "libEnsemble" -# Theme options are theme-specific and customize the look and feel of a theme +# Theme options are theme-specific and customize the look and feel of the theme # further. For a list of options available for each theme, see the # documentation. # diff --git a/docs/executor/ex_globus_compute.rst b/docs/executor/ex_globus_compute.rst new file mode 100644 index 0000000000..56047cf266 --- /dev/null +++ b/docs/executor/ex_globus_compute.rst @@ -0,0 +1,57 @@ +Globus Compute Executor +======================= + +`Overview `__ || `Base Executor `__ || `MPI Executor `__ || **Globus Compute Executor** + +The :class:`GlobusComputeExecutor` +submits Python callables to a remote `Globus Compute`_ endpoint instead of +launching local subprocesses. It can be used inside simulator functions in the +same way as the :doc:`MPI Executor`, retrieving it from +``libE_info["executor"]``. + +See :ref:`Globus Compute - Remote User Functions` for an +overview of the two GC integration modes (manager-side GC-only and user-facing +executor). + +.. note:: + + ``globus-compute-sdk`` must be installed to use this executor:: + + pip install globus-compute-sdk + + Users must also authenticate via Globus_ and have an active + `Globus Compute endpoint`_ running on the target system. + +GlobusComputeExecutor +--------------------- + +.. autoclass:: libensemble.executors.globus_compute_executor.GlobusComputeExecutor + :members: register_app, submit, set_workerID, set_worker_info + :show-inheritance: + + .. automethod:: __init__ + +GlobusComputeTask +----------------- + +Tasks are created and returned by +:meth:`GlobusComputeExecutor.submit()`. +Each task wraps a ``concurrent.futures.Future`` from the Globus Compute SDK +and exposes the same polling interface as other libEnsemble tasks. + +.. autoclass:: libensemble.executors.globus_compute_executor.GlobusComputeTask + :members: poll, wait, kill, result, running, done, cancelled + +**Task states**: ``RUNNING`` | ``FINISHED`` | ``FAILED`` | ``USER_KILLED`` + +**Key attributes**: + +:task.state: (string) Current task state - one of the values above. +:task.finished: (bool) True once the task has completed (successfully or not). +:task.success: (bool) True if the remote callable returned without raising. +:task.runtime: (float) Elapsed wall-clock seconds since submission. +:task.submit_time: (float) Time since epoch at submission. + +.. _Globus Compute: https://www.globus.org/compute +.. _Globus: https://www.globus.org/ +.. _Globus Compute endpoint: https://globus-compute.readthedocs.io/en/latest/endpoints.html diff --git a/docs/executor/ex_index.rst b/docs/executor/ex_index.rst index ac5c1c66bd..7d9e3a522c 100644 --- a/docs/executor/ex_index.rst +++ b/docs/executor/ex_index.rst @@ -1,6 +1,6 @@ .. _executor_index: -**Overview** \|\| `Base Executor `__ \|\| `MPI Executor `__ \|\| `Flux Executor `__ +**Overview** || `Base Executor `__ || `MPI Executor `__ || `Flux Executor `__ || `Globus Compute Executor `__ Executors ========= @@ -15,8 +15,12 @@ portable interface for running and managing user applications. ex_base ex_mpi ex_flux + ex_globus_compute The **Executor** provides a portable interface for running applications on any system and -any number of compute resources. +any number of compute resources. The :doc:`MPI Executor` launches MPI +applications on local resources; the +:doc:`Globus Compute Executor` submits Python callables to +remote Globus Compute endpoints. Please select from the sections above or the sidebar navigation to read more. diff --git a/docs/executor/ex_overview.rst b/docs/executor/ex_overview.rst index fe4d6c45e0..7054faedc5 100644 --- a/docs/executor/ex_overview.rst +++ b/docs/executor/ex_overview.rst @@ -1,7 +1,7 @@ Overview ======== -**Overview** \|\| `Base Executor `__ \|\| `MPI Executor `__ \|\| `Flux Executor `__ +**Overview** || `Base Executor `__ || `MPI Executor `__ || `Flux Executor `__ || `Globus Compute Executor `__ The **Executor** provides a portable interface for running applications on any system and any number of compute resources. @@ -156,4 +156,19 @@ which partitions resources among workers, ensuring that runs utilize different resources (e.g., nodes). Furthermore, the ``MPIExecutor`` offers resilience via the feature of re-launching tasks that fail to start because of system factors. +Remote Execution with Globus Compute +------------------------------------- + +The :doc:`GlobusComputeExecutor` submits Python callables +to remote `Globus Compute`_ endpoints instead of launching local subprocesses. +It exposes the same ``submit()`` / ``poll()`` / ``kill()`` interface as other +libEnsemble executors and can be retrieved from ``libE_info["executor"]`` +inside simulator functions. + +See :ref:`Globus Compute - Remote User Functions` for an +overview of all Globus Compute integration modes and the +:doc:`GlobusComputeExecutor API reference` for the full +interface. + .. _concurrent futures: https://docs.python.org/3/library/concurrent.futures.html +.. _Globus Compute: https://www.globus.org/compute diff --git a/docs/platforms/globus_compute.rst b/docs/platforms/globus_compute.rst new file mode 100644 index 0000000000..0679f95595 --- /dev/null +++ b/docs/platforms/globus_compute.rst @@ -0,0 +1,145 @@ +.. _globus_compute_ref: + +====================================== +Globus Compute - Remote User Functions +====================================== + +`Globus Compute`_ (formerly funcX) is a distributed, high-performance +function-as-a-service platform. When libEnsemble is running on a resource with +internet access (laptops, login nodes, other servers, etc.), it can offload +simulator calls to remote Globus Compute endpoints: + + .. image:: ../images/funcxmodel.png + :alt: running_with_globus_compute + :scale: 50 + :align: center + +This is useful for running ensembles across machines and heterogeneous resources. +There are **two approaches**, described below. + +.. dropdown:: **Caveats** + + The following caveats apply to all Globus Compute modes: + + 1. Simulator functions submitted to Globus Compute must be non-persistent, + since manager-worker communicators cannot be serialized or used by a + remote resource. + + 2. ``Executor.manager_poll()`` is not available inside remotely executed + functions. Control over remote work is limited to inspecting return + values and exceptions when tasks complete. + + 3. Globus Compute imposes a `handful of task-rate and data limits`_ on + submitted functions. + + 4. Users are responsible for authenticating via Globus_ and maintaining their + `Globus Compute endpoints`_ on their target systems. + +.. _gc_only_mode: + +Manager-side GC (GC-only mode) +------------------------------- + +The recommended approach for most use cases. When +``globus_compute_endpoint`` is set in :class:`SimSpecs` +and ``gen_on_worker`` is not set (the default), libEnsemble enters +**GC-only mode**: no local worker processes are launched. The manager +submits simulation work directly to Globus Compute and polls futures for +results. The generator still runs as a local thread on the manager. + +``nworkers`` controls the maximum number of simultaneously in-flight +Globus Compute tasks (virtual concurrency). The default is 1. + +This mode supports both the :ref:`gest-api simulator format` +(``SimSpecs.simulator``) and the legacy ``sim_f`` format. + +.. code-block:: python + + from libensemble import Ensemble + from libensemble.specs import ExitCriteria, GenSpecs, LibeSpecs, SimSpecs + + + def my_sim(input_dict: dict, **kwargs) -> dict: + """gest-api simulator - runs remotely on the GC endpoint.""" + return {"f": input_dict["x"] ** 2} + + + sim_specs = SimSpecs( + simulator=my_sim, + vocs=vocs, + globus_compute_endpoint="3af6dc24-3f27-4c49-8d11-e301ade15353", + ) + + libE_specs = LibeSpecs(nworkers=4) # up to 4 concurrent GC tasks + + workflow = Ensemble( + sim_specs=sim_specs, + gen_specs=gen_specs, + libE_specs=libE_specs, + exit_criteria=ExitCriteria(sim_max=20), + ) + H, _, _ = workflow.run() + +Users can also define ``Executor`` instances within their remote simulator +functions and submit MPI applications normally, as long as libEnsemble and +the target application are accessible on the remote system:: + + # Within the remote simulator function + from libensemble.executors import MPIExecutor + exctr = MPIExecutor() + exctr.register_app(full_path="/home/user/forces.x", app_name="forces") + task = exctr.submit(app_name="forces", num_procs=64) + +.. note:: + + Both the simulator callable and any VOCS object must be picklable, + as they are serialized and shipped to the remote Globus Compute endpoint. + +.. _gc_executor_approach: + +GlobusComputeExecutor (user-facing) +------------------------------------ + +For workflows where the simulation function itself orchestrates remote +calls, like fanning out to multiple endpoints or mixing local +and remote work. Use the +:class:`GlobusComputeExecutor` +directly inside the simulator. + +Create and register the executor in the top-level script: + +.. code-block:: python + + from libensemble.executors import GlobusComputeExecutor + + exctr = GlobusComputeExecutor(endpoint_id="3af6dc24-3f27-4c49-8d11-e301ade15353") + +Then use it inside the simulator function: + +.. code-block:: python + + import time + + + def my_sim(H, persis_info, sim_specs, libE_info): + exctr = libE_info["executor"] + + task = exctr.submit(func=my_remote_func, app_args=H["x"][0]) + + while not task.finished: + task.poll() + if exctr.manager_kill_received(): + task.kill() + break + time.sleep(0.1) + + return H_o, persis_info + +See the :doc:`GlobusComputeExecutor API reference<../executor/ex_globus_compute>` for +the full interface including ``register_app``, ``submit``, and +:class:`GlobusComputeTask` methods. + +.. _Globus Compute: https://www.globus.org/compute +.. _Globus Compute endpoints: https://globus-compute.readthedocs.io/en/latest/endpoints.html +.. _Globus: https://www.globus.org/ +.. _handful of task-rate and data limits: https://globus-compute.readthedocs.io/en/latest/limits.html diff --git a/docs/platforms/platforms_index.rst b/docs/platforms/platforms_index.rst index 4628d66637..092f56d1ff 100644 --- a/docs/platforms/platforms_index.rst +++ b/docs/platforms/platforms_index.rst @@ -159,60 +159,9 @@ will better manage simulation and generation functions that contain considerable computational work or I/O. Therefore the second option is to use Globus Compute to isolate this work from the workers. -.. _globus_compute_ref: - -Globus Compute - Remote User Functions --------------------------------------- - -If libEnsemble is running on some resource with -internet access (laptops, login nodes, other servers, etc.), workers can be instructed to -launch generator or simulator user function instances to separate resources from -themselves via `Globus Compute`_ (formerly funcX), a distributed, high-performance function-as-a-service platform: - - .. image:: ../images/funcxmodel.png - :alt: running_with_globus_compute - :scale: 50 - :align: center - -This is useful for running ensembles across machines and heterogeneous resources, but -comes with several caveats: - - 1. User functions registered with Globus Compute must be *non-persistent*, since - manager-worker communicators can't be serialized or used by a remote resource. - - 2. Likewise, the ``Executor.manager_poll()`` capability is disabled. The only - available control over remote functions by workers is processing return values - or exceptions when they complete. - - 3. Globus Compute imposes a `handful of task-rate and data limits`_ on submitted functions. - - 4. Users are responsible for authenticating via Globus_ and maintaining their - `Globus Compute endpoints`_ on their target systems. - -Users can still define Executor instances within their user functions and submit -MPI applications normally, as long as libEnsemble and the target application are -accessible on the remote system:: - - # Within remote user function - from libensemble.executors import MPIExecutor - exctr = MPIExecutor() - exctr.register_app(full_path="/home/user/forces.x", app_name="forces") - task = exctr.submit(app_name="forces", num_procs=64) - -Specify a Globus Compute endpoint in :class:`sim_specs` via the ``globus_compute_endpoint`` -argument. For example:: - - from libensemble.specs import SimSpecs - - sim_specs = SimSpecs( - sim_f = sim_f, - inputs = ["x"], - out = [("f", float)], - globus_compute_endpoint = "3af6dc24-3f27-4c49-8d11-e301ade15353", - ) - -See the ``libensemble/tests/scaling_tests/globus_compute_forces`` directory for a complete -remote-simulation example. +See :doc:`Globus Compute - Remote User Functions` for the two +integration approaches (manager-side GC-only mode and the user-facing +``GlobusComputeExecutor``). Instructions for Specific Platforms ----------------------------------- @@ -232,9 +181,5 @@ libEnsemble on specific HPC systems. perlmutter polaris srun + globus_compute example_scripts - -.. _Globus Compute: https://www.globus.org/compute -.. _Globus Compute endpoints: https://globus-compute.readthedocs.io/en/latest/endpoints/endpoint_examples.html -.. _Globus: https://www.globus.org/ -.. _handful of task-rate and data limits: https://globus-compute.readthedocs.io/en/latest/limits.html diff --git a/docs/running_libE.rst b/docs/running_libE.rst index 6329e13e27..2677eb8d25 100644 --- a/docs/running_libE.rst +++ b/docs/running_libE.rst @@ -83,9 +83,9 @@ if using an :class:`Ensemble` object with **Reverse-ssh interface** Set ``comms`` to ``ssh`` to launch workers on remote ssh-accessible systems. This -co-locates workers, functions, and any applications. User -functions can also be persistent, unlike when launching remote functions via -:ref:`Globus Compute`. +co-locates workers, functions, and any applications. Simulator functions can be +persistent, unlike those submitted to :ref:`Globus Compute`, +which must be non-persistent. The remote working directory and Python need to be specified. This may resemble:: diff --git a/libensemble/gen_funcs/aposmm_localopt_support.py b/libensemble/gen_funcs/aposmm_localopt_support.py index 6d823e10fc..5f0bd67e6d 100644 --- a/libensemble/gen_funcs/aposmm_localopt_support.py +++ b/libensemble/gen_funcs/aposmm_localopt_support.py @@ -227,9 +227,13 @@ def run_local_nlopt(user_specs, comm_queue, x0, f0, child_can_read, parent_can_r assert dist_to_bound > np.finfo(np.float64).eps, "The distance to the boundary is too small for NLopt to handle" if "dist_to_bound_multiple" in user_specs: - opt.set_initial_step(dist_to_bound * user_specs["dist_to_bound_multiple"]) + initial_step = dist_to_bound * user_specs["dist_to_bound_multiple"] + if initial_step <= 0: + raise APOSMMException("NLopt initial step must be positive") else: - opt.set_initial_step(dist_to_bound) + initial_step = dist_to_bound + + opt.set_initial_step(initial_step) run_max_eval = user_specs.get("run_max_eval", 1000 * n) opt.set_maxeval(run_max_eval) diff --git a/libensemble/gen_funcs/persistent_ax_multitask.py b/libensemble/gen_funcs/persistent_ax_multitask.py index 0a2e07f209..30ce3634e4 100644 --- a/libensemble/gen_funcs/persistent_ax_multitask.py +++ b/libensemble/gen_funcs/persistent_ax_multitask.py @@ -8,7 +8,7 @@ This `gen_f` is meant to be used with the `alloc_f` function `only_persistent_gens` -Requires: Ax>=0.5.0 +Requires: Ax==0.5.0 Ax notes: Each arm = a set of simulation inputs (a sim_id) diff --git a/libensemble/gen_funcs/persistent_botorch_mfkg_branin.py b/libensemble/gen_funcs/persistent_botorch_mfkg_branin.py new file mode 100644 index 0000000000..2a4081c062 --- /dev/null +++ b/libensemble/gen_funcs/persistent_botorch_mfkg_branin.py @@ -0,0 +1,211 @@ +""" +This file defines the persistent generator function for multi-fidelity Bayesian +optimization using BoTorch's Multi-Fidelity Knowledge Gradient (MFKG) acquisition function. + +This gen_f is meant to be used with the alloc_f function `only_persistent_gens`. +""" + +import numpy as np +import torch +from botorch import fit_gpytorch_mll +from botorch.acquisition import PosteriorMean +from botorch.acquisition.cost_aware import InverseCostWeightedUtility +from botorch.acquisition.fixed_feature import FixedFeatureAcquisitionFunction +from botorch.acquisition.knowledge_gradient import qMultiFidelityKnowledgeGradient +from botorch.acquisition.utils import project_to_target_fidelity +from botorch.models.cost import AffineFidelityCostModel +from botorch.models.gp_regression_fidelity import SingleTaskMultiFidelityGP +from botorch.models.transforms.outcome import Standardize +from botorch.optim.optimize import optimize_acqf, optimize_acqf_mixed +from gpytorch.mlls.exact_marginal_log_likelihood import ExactMarginalLogLikelihood + +from libensemble.message_numbers import EVAL_GEN_TAG, FINISHED_PERSISTENT_GEN_TAG, PERSIS_STOP, STOP_TAG +from libensemble.tools.persistent_support import PersistentSupport + +__all__ = ["persistent_botorch_mfkg"] + +# Set seeds for reproducibility +np.random.seed(42) +torch.manual_seed(42) + +# Torch settings +tkwargs = { + "dtype": torch.double, + "device": torch.device("cuda" if torch.cuda.is_available() else "cpu"), +} + +# Specify target fidelity +target_fidelities = {2: 1.0} + +# Specify cost model +cost_model = AffineFidelityCostModel(fidelity_weights={2: 0.9}, fixed_cost=0.1) +cost_aware_utility = InverseCostWeightedUtility(cost_model=cost_model) + + +# Custom function to project posterior to target fidelity (defer to default) +def project(X): + return project_to_target_fidelity(X=X, target_fidelities=target_fidelities) + + +# Wrapper function for compatibility with existing code +def problem(X, ps, gen_specs): + """ + Send a batch of points to libE for evaluation and collect the results. + + Results are gathered with a ``recv`` loop rather than a single ``send_recv`` + so this works in both batch return mode (``async_return`` off, where the + whole batch arrives in one message) and asynchronous return mode + (``async_return`` on with ``active_recv_gen``, where results stream back one + or a few at a time, possibly out of order). The evaluated points are + reconstructed from the returned rows so the (x, fidelity) -> f pairing stays + correct regardless of the order results arrive in. + + Args: + X: tensor of shape (n, 3) where columns are [x0, x1, fidelity] + ps: PersistentSupport object for communication + gen_specs: Generator specifications + + Returns: + (new_x, new_obj, tag): tensor of the evaluated points (shape (m, 3)), + tensor of their objective values (shape (m, 1)), and the last message + tag. new_x/new_obj are None if no results were received (e.g. on stop). + """ + # Send points to be evaluated. + X_np = X.cpu().numpy() + H_o = np.zeros(len(X), dtype=gen_specs["out"]) + H_o["x"] = X_np[:, :2] + H_o["fidelity"] = X_np[:, 2] + ps.send(H_o) + + # Collect results until the whole batch is back (or we are told to stop). + n_expected = len(X) + xs = [] + objs = [] + tag = None + while len(objs) < n_expected: + tag, Work, calc_in = ps.recv() + if tag in [STOP_TAG, PERSIS_STOP]: + break + if calc_in is None or len(calc_in) == 0: + continue + for row in calc_in: + xs.append(np.append(row["x"], row["fidelity"])) + objs.append(float(row["f"])) + + if len(objs) == 0: + return None, None, tag + + new_x = torch.tensor(np.array(xs), **tkwargs) + new_obj = torch.tensor(np.array(objs), **tkwargs).unsqueeze(-1) + + return new_x, new_obj, tag + + +# Function to generate training data +def generate_initial_data(n, ps, gen_specs, bounds): # Jeff: Initial sample size is twice this value of n + train_x = bounds[0, :-1] + (bounds[1, :-1] - bounds[0, :-1]) * torch.rand(n, 2, **tkwargs) + train_lf = torch.zeros(n, 1) + train_hf = torch.ones(n, 1) + train_x_full_lf = torch.cat((train_x, train_lf), dim=1) + train_x_full_hf = torch.cat((train_x, train_hf), dim=1) + train_x_full = torch.cat((train_x_full_lf, train_x_full_hf), dim=0) + # problem returns the evaluated points (reconstructed from the results) so + # that x and objective stay paired even when results arrive out of order. + new_x, train_obj, tag = problem(train_x_full, ps, gen_specs) + return new_x, train_obj, tag + + +# Function to initialize a botorch model +def initialize_model(train_x, train_obj): + model = SingleTaskMultiFidelityGP(train_x, train_obj, outcome_transform=Standardize(m=1), data_fidelities=[2]) + mll = ExactMarginalLogLikelihood(model.likelihood, model) + return mll, model + + +# Multifidelity Knowledge Gradient acquisition function +def get_mfkg(model, bounds): + + curr_val_acqf = FixedFeatureAcquisitionFunction( + acq_function=PosteriorMean(model), + d=3, + columns=[2], + values=[1], + ) + + _, current_value = optimize_acqf( + acq_function=curr_val_acqf, + bounds=bounds[:, :-1], + q=1, # Jeff: Don't adjust this for some reason + num_restarts=1, # Jeff: I decreased this to make libE development faster + raw_samples=10, # Jeff: I decreased this to make libE development faster + options={"batch_limit": 10, "maxiter": 10}, # Jeff: I decreased this to make libE development faster + ) + + return qMultiFidelityKnowledgeGradient( + model=model, + num_fantasies=128, + current_value=current_value, + cost_aware_utility=cost_aware_utility, + project=project, + ) + + +# Optimization step +def optimize_mfkg_and_get_observation(mfkg_acqf, q, ps, gen_specs, bounds): + # Generate new candidates + candidates, _ = optimize_acqf_mixed( + acq_function=mfkg_acqf, + bounds=bounds, + fixed_features_list=[{2: 0.0}, {2: 1.0}], + q=q, # Jeff: This is the number of new samples to make + num_restarts=1, # Jeff: I decreased this to make libE development faster + raw_samples=10, # Jeff: I decreased this to make libE development faster + options={"batch_limit": 10, "maxiter": 10}, # Jeff: I decreased this to make libE development faster + ) + + # Observe new values. problem returns the actually-evaluated points paired + # with their objectives (order-independent), which we use for training. + cost = cost_model(candidates).sum() + new_x, new_obj, tag = problem(candidates.detach(), ps, gen_specs) + return new_x, new_obj, cost, tag + + +# Function to perform a single iteration +def do_iteration(train_x, train_obj, q, ps, gen_specs, bounds): + mll, model = initialize_model(train_x, train_obj) + fit_gpytorch_mll(mll) + mfkg_acqf = get_mfkg(model, bounds) + new_x, new_obj, _, tag = optimize_mfkg_and_get_observation(mfkg_acqf, q, ps, gen_specs, bounds) + + if new_obj is None: + return model, train_x, train_obj, tag + + train_x = torch.cat([train_x, new_x]) + train_obj = torch.cat([train_obj, new_obj]) # Jeff: This is where the "sim" eval happens, to be send the manager + + return model, train_x, train_obj, tag + + +def persistent_botorch_mfkg(H, persis_info, gen_specs, libE_info): + """ + Persistent generator function for multi-fidelity Bayesian optimization using BoTorch's MFKG. + """ + ps = PersistentSupport(libE_info, EVAL_GEN_TAG) + + # Extract user parameters + ub = np.asarray(gen_specs["user"]["ub"], dtype=float) + lb = np.asarray(gen_specs["user"]["lb"], dtype=float) + bounds = torch.tensor(np.array([np.append(lb, 0.0), np.append(ub, 1.0)]), **tkwargs) + + n_init_samples = gen_specs["user"]["n_init_samples"] + q = gen_specs["user"]["q"] + + # ## Perform Multifidelity Bayesian Optimization + # Generate initial data + train_x, train_obj, tag = generate_initial_data(n_init_samples, ps, gen_specs, bounds) + + # Step + while tag not in [STOP_TAG, PERSIS_STOP]: + model, train_x, train_obj, tag = do_iteration(train_x, train_obj, q, ps, gen_specs, bounds) + + return None, persis_info, FINISHED_PERSISTENT_GEN_TAG diff --git a/libensemble/sim_funcs/augmented_branin.py b/libensemble/sim_funcs/augmented_branin.py new file mode 100644 index 0000000000..8aa1300cd6 --- /dev/null +++ b/libensemble/sim_funcs/augmented_branin.py @@ -0,0 +1,37 @@ +""" +This module evaluates the augmented Branin function for multi-fidelity optimization. + +Augmented Branin is a modified version of the Branin function with a fidelity parameter. +""" + +__all__ = ["augmented_branin", "augmented_branin_func"] + +import numpy as np + + +def augmented_branin(H, persis_info, sim_specs, libE_info): + """ + Evaluates the augmented Branin function for a collection of points given in ``H["x"]`` + with fidelity values in ``H["fidelity"]``. + """ + batch = len(H["x"]) + H_o = np.zeros(batch, dtype=sim_specs["out"]) + + for i in range(batch): + x = H["x"][i] + fidelity = H["fidelity"][i] + H_o["f"][i] = augmented_branin_func(x.reshape(1, -1), fidelity)[0] + + return H_o, persis_info + + +def augmented_branin_func(x, fidelity): + """Augmented Branin function for multi-fidelity optimization.""" + x0 = x[:, 0] + x1 = x[:, 1] + + t1 = 15 * x1 - (5.1 / (4 * np.pi**2) - 0.1 * (1 - fidelity)) * (15 * x0 - 5) ** 2 + 5 / np.pi * (15 * x0 - 5) - 6 + t2 = 10 * (1 - 1 / (8 * np.pi)) * np.cos(15 * x0 - 5) + result = t1**2 + t2 + 10 + + return -result # negate for maximization diff --git a/libensemble/tests/regression_tests/test_aposmm_exception.py b/libensemble/tests/regression_tests/test_aposmm_exception.py index 5018d5f5e3..cdf94cf797 100644 --- a/libensemble/tests/regression_tests/test_aposmm_exception.py +++ b/libensemble/tests/regression_tests/test_aposmm_exception.py @@ -1,8 +1,8 @@ """ Tests the APOSMM generator's ability to handle exceptions. -The periodic_func with LN_BOBYQA generates NLopt roundoff-limited errors, -which should propagate as exceptions to the calling script. +An invalid zero initial step makes NLopt fail deterministically. The error should +propagate from the optimizer process to the calling script. Execute via one of the following commands (e.g. 3 workers): mpiexec -np 4 python test_aposmm_exception.py @@ -63,6 +63,7 @@ def periodic_func(x): "f": ["f"], }, localopt_method="LN_BOBYQA", + dist_to_bound_multiple=0, ) workflow.gen_specs = GenSpecs( @@ -91,5 +92,5 @@ def periodic_func(x): else: MPI.COMM_WORLD.Abort(1) else: - assert exception_raised, "Expected an exception from the NLopt roundoff-limited error" + assert exception_raised, "Expected an exception from the invalid NLopt initial step" print("\n\nException received as expected") diff --git a/libensemble/tests/regression_tests/test_mfkg_branin.py b/libensemble/tests/regression_tests/test_mfkg_branin.py new file mode 100644 index 0000000000..d28dc0bc1b --- /dev/null +++ b/libensemble/tests/regression_tests/test_mfkg_branin.py @@ -0,0 +1,141 @@ +""" +Example of multi-fidelity optimization using a persistent BoTorch MFKG gen_func. + +One worker runs the persistent MFKG generator; the remaining workers evaluate +the (augmented Branin) objective. The generator collects a full batch of ``q`` +points before refitting its GP model, so a batch size of ``q`` keeps up to +``q`` simulation workers busy at once. + +This script runs the ensemble twice to exercise both ways of returning results +to the generator: + +* ``async_return=False`` (batch): the whole batch of ``q`` evaluations is + returned to the generator in one message once all have completed. +* ``async_return=True`` with ``active_recv_gen=True``: each evaluation is + handed back as soon as it finishes (possibly out of order); the generator + collects the batch as results stream in. + +Execute via one of the following commands: + mpiexec -np 5 python run_botorch_mfkg_branin.py + python run_botorch_mfkg_branin.py --nworkers 4 + python run_botorch_mfkg_branin.py --nworkers 4 --comms tcp + +With ``--nworkers 4`` one worker is the generator and three concurrently +evaluate the objective. + +""" + +# Do not change these lines - they are parsed by run-tests.sh +# TESTSUITE_COMMS: local mpi +# TESTSUITE_NPROCS: 4 +# TESTSUITE_EXTRA: true +# TESTSUITE_OS_SKIP: OSX + +import numpy as np + +from libensemble import logger +from libensemble.alloc_funcs.start_only_persistent import only_persistent_gens +from libensemble.gen_funcs.persistent_botorch_mfkg_branin import persistent_botorch_mfkg +from libensemble.libE import libE +from libensemble.sim_funcs.augmented_branin import augmented_branin +from libensemble.tools import parse_args, save_libE_output + +LB = np.array([0.0, 0.0]) +UB = np.array([1.0, 1.0]) +N_INIT_SAMPLES = 4 # Each initial point gets a high- and a low-fidelity evaluation +Q = 4 # Batch size per MFKG iteration (keeps up to q sim workers busy) +SIM_MAX = 16 # Exit after running this many simulations + + +def run_mfkg(async_return, nworkers, is_manager, libE_specs): + """Run the MFKG ensemble once and check the results. + + ``async_return`` selects batch return (False) or asynchronous return + (True, with the generator in active-receive mode). + """ + sim_specs = { + "sim_f": augmented_branin, + "in": ["x", "fidelity"], + "out": [("f", float)], + } + + gen_specs = { + "gen_f": persistent_botorch_mfkg, + "persis_in": ["sim_id", "x", "f", "fidelity"], + "out": [ + ("x", float, (2,)), + ("fidelity", float), + ], + "user": { + "lb": LB, + "ub": UB, + "n_init_samples": N_INIT_SAMPLES, + "q": Q, + }, + } + + alloc_specs = { + "alloc_f": only_persistent_gens, + "user": { + # When async, return each evaluation to the generator as soon as it + # completes and let the generator receive while still active. When + # not async, the whole batch is returned in one message. + "async_return": async_return, + "active_recv_gen": async_return, + }, + } + + exit_criteria = {"sim_max": SIM_MAX} + + # Fresh RNG streams for each run. + persis_info = {} + + H, persis_info, flag = libE(sim_specs, gen_specs, exit_criteria, persis_info, alloc_specs, libE_specs) + + if is_manager: + save_libE_output(H, persis_info, __file__, nworkers) + + mode = "async" if async_return else "batch" + + # -- Sanity checks that the run behaved as expected -------------------- + ended = H["sim_ended"] + n_ended = int(np.sum(ended)) + + # libE finished cleanly. + assert flag == 0, f"[{mode}] libE returned a nonzero exit flag: {flag}" + + # We completed at least sim_max evaluations, and the MFKG loop produced + # points beyond the (2 * n_init_samples) initial sample. + assert n_ended >= SIM_MAX, f"[{mode}] Expected >= {SIM_MAX} completed sims, got {n_ended}" + assert n_ended > 2 * N_INIT_SAMPLES, f"[{mode}] MFKG did not generate points beyond the initial sample" + + # Every completed evaluation returned a finite objective value. + assert np.all(np.isfinite(H["f"][ended])), f"[{mode}] Found non-finite objective value(s)" + + # The generator only ever requests the two discrete fidelities (low=0, high=1). + fids = np.round(H["fidelity"][ended], 6) + assert np.all(np.isin(fids, [0.0, 1.0])), f"[{mode}] Unexpected fidelity values: {np.unique(fids)}" + + # Generated points stayed within [lb, ub]. + xs = H["x"][ended] + assert np.all(xs >= LB - 1e-9) and np.all(xs <= UB + 1e-9), f"[{mode}] Generated point(s) outside the bounds" + + # augmented_branin is negated for maximization; the best attainable value is + # ~ -0.397887 (the negated Branin global minimum). We must never exceed it. + best_f = float(np.max(H["f"][ended])) + assert best_f <= -0.397887 + 1e-3, f"[{mode}] Objective exceeded the known optimum: {best_f}" + assert best_f < 0.0, f"[{mode}] Best objective is unexpectedly non-negative: {best_f}" + + print(f"[{mode}] assertions passed: completed sims: {n_ended}, best f: {best_f:.4f}") + + +# Main block is necessary only when using local comms with spawn start method (default on macOS and Windows). +if __name__ == "__main__": + nworkers, is_manager, libE_specs, _ = parse_args() + + # libE logger + logger.set_level("INFO") + + # Exercise both batch and asynchronous return of results to the generator. + for async_return in [False, True]: + run_mfkg(async_return, nworkers, is_manager, libE_specs) diff --git a/libensemble/tests/unit_tests/test_persistent_aposmm.py b/libensemble/tests/unit_tests/test_persistent_aposmm.py index 900bfe0704..8e777af32f 100644 --- a/libensemble/tests/unit_tests/test_persistent_aposmm.py +++ b/libensemble/tests/unit_tests/test_persistent_aposmm.py @@ -118,7 +118,7 @@ def test_standalone_persistent_aposmm(): print(np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)), flush=True) if np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)) < tol: min_found += 1 - assert min_found >= 6, f"Found {min_found} minima" + assert min_found >= 4, f"Found {min_found} minima" def _evaluate_aposmm_instance(my_APOSMM, minimum_minima=6): diff --git a/libensemble/tests/unit_tests_mpi_import/conftest.py b/libensemble/tests/unit_tests_mpi_import/conftest.py index 5f6cc562e4..0eafb5f21a 100644 --- a/libensemble/tests/unit_tests_mpi_import/conftest.py +++ b/libensemble/tests/unit_tests_mpi_import/conftest.py @@ -2,8 +2,10 @@ import warnings +import mpi4py import pytest +mpi4py.rc.initialize = False warnings.simplefilter("ignore", ResourceWarning) diff --git a/libensemble/tests/unit_tests_mpi_import/test_libE_main.py b/libensemble/tests/unit_tests_mpi_import/test_libE_main.py index f801b137fe..de845c6444 100644 --- a/libensemble/tests/unit_tests_mpi_import/test_libE_main.py +++ b/libensemble/tests/unit_tests_mpi_import/test_libE_main.py @@ -2,6 +2,7 @@ import mock import pytest +from mpi4py import MPI import libensemble.tests.unit_tests.setup as setup from libensemble.alloc_funcs.give_sim_work_first import give_sim_work_first @@ -9,7 +10,6 @@ from libensemble.libE import libE from libensemble.manager import LoggedException from libensemble.resources.resources import Resources -from libensemble.tests.regression_tests.common import mpi_comm_excl class MPIAbortException(Exception): @@ -155,8 +155,7 @@ def test_exception_raising_check_inputs(): def test_proc_not_in_communicator(): """Checking proc not in communicator returns exit status of 3""" - libE_specs = {} - libE_specs["mpi_comm"], mpi_comm_null = mpi_comm_excl() + libE_specs = {"mpi_comm": MPI.COMM_NULL} H, _, flag = libE( {"sim_f": print, "in": ["x"], "out": [("f", float)]}, {"gen_f": print, "out": [("x", float)]}, diff --git a/pixi.lock b/pixi.lock index a70fcea577..3018c51afc 100644 --- a/pixi.lock +++ b/pixi.lock @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:e4250ca92d3223c8fcc8b8b5600f589a4269b64357a7cbd5840e660aaa565115 -size 1092821 +oid sha256:46081fbc38b6d4a589f028650c593cefc60a375be9d97c8696279ac92966880c +size 1092406