diff --git a/.github/workflows/basic.yml b/.github/workflows/basic.yml index 366e4eaa3..c88962417 100644 --- a/.github/workflows/basic.yml +++ b/.github/workflows/basic.yml @@ -80,7 +80,7 @@ jobs: -A "-W error" --no-coverage --fail-fast --durations 25 --timings-json timings-${{ matrix.comms }}.json - name: Upload timings - if: always() + if: always() && env.ACT != 'true' uses: actions/upload-artifact@v7 with: name: core-${{ matrix.comms }}-timings diff --git a/.github/workflows/extra.yml b/.github/workflows/extra.yml index 156514b50..16e898086 100644 --- a/.github/workflows/extra.yml +++ b/.github/workflows/extra.yml @@ -44,7 +44,7 @@ jobs: --tier core --no-coverage --fail-fast --durations 25 --timings-json timings-core-${{ matrix.comms }}.json - name: Upload timings - if: always() + if: always() && env.ACT != 'true' uses: actions/upload-artifact@v7 with: name: complete-core-${{ matrix.comms }}-timings @@ -113,7 +113,7 @@ jobs: include-hidden-files: true if-no-files-found: warn - name: Upload timings - if: always() + if: always() && env.ACT != 'true' uses: actions/upload-artifact@v7 with: name: coverage-timings-${{ matrix.shard }} @@ -200,7 +200,7 @@ jobs: --shard ${{ matrix.shard }}/2 --timings-json timings-external-${{ matrix.shard }}.json - name: Upload timings - if: always() + if: always() && env.ACT != 'true' uses: actions/upload-artifact@v7 with: name: external-${{ matrix.shard }}-timings @@ -227,7 +227,7 @@ jobs: --tier slow --no-coverage --fail-fast --durations 25 --timings-json timings-slow.json - name: Upload timings - if: always() + if: always() && env.ACT != 'true' uses: actions/upload-artifact@v7 with: name: slow-test-timings diff --git a/AGENTS.md b/AGENTS.md index 254b38234..0053819e9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -86,6 +86,7 @@ for local development running individual tests is sufficient. - Individual unit tests can be run with ``pixi run -e dev pytest path/to/test_file``. - A libEnsemble run typically outputs an ``ensemble.log`` and ``libE_stats.txt`` file in the working directory. Check these files for tracebacks or run statistics. - An "ensemble" or "workflow" directory may also be created, often containing per-simulation output directories +- Do not add tests solely to cover code that proves unreachable after analysis. If a branch/guard/except cannot be hit in production (e.g., defensive `None` checks before `set_rset_team()` or `try/except` around inputs already validated to `int|float|None`), prefer removing the dead code over manufacturing synthetic coverage (e.g., `__index__`/`__float__` tricks or `math.ceil` monkeypatches). Justify removal in the PR/commit. Modernizing Scripts for libEnsemble 2.0 --------------------------------------- diff --git a/docs/conf.py b/docs/conf.py index e03aa9d34..e319db525 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -128,6 +128,7 @@ class AxParameterWarning(Warning): # Ensure it's a real warning subclass r"https://www\.sfu\.ca/~ssurjano/camel6\.html", r"https://github\.com/Libensemble/libensemble/blob/.*", r"https://img\.shields\.io/conda/v/conda-forge/libensemble", + r"https://github\.com/xopt-org/Xopt", ] autodoc_pydantic_model_show_json = False diff --git a/docs/data_structures/libE_specs/libE_specs_resources.rst b/docs/data_structures/libE_specs/libE_specs_resources.rst index b7016a906..cfaa9b48c 100644 --- a/docs/data_structures/libE_specs/libE_specs_resources.rst +++ b/docs/data_structures/libE_specs/libE_specs_resources.rst @@ -18,17 +18,23 @@ Resources **num_resource_sets** [int]: The total number of resource sets into which resources will be divided. By default resources will be divided by workers (excluding - ``zero_resource_workers``). + ``zero_resource_workers``), or, if ``gpu_sharing_factor > 1``, into one + resource set per GPU share. + +**gpu_sharing_factor** [int] = ``1``: + Number of tasks that may share each GPU. For example, ``2`` lets two tasks + requesting ``num_gpus=0.5`` run on the same GPU. Tasks sharing a GPU share + its memory. See :ref:`GPU sharing`. **gen_num_procs** [int] = ``0``: The default number of processors (MPI ranks) required by generators. Unless overridden by equivalent ``persis_info`` settings, generators will be allocated this many processors for applications launched via the MPIExecutor. -**gen_num_gpus** [int] = ``0``: +**gen_num_gpus** [int|float] = ``0``: The default number of GPUs required by generators. Unless overridden by the equivalent ``persis_info`` settings, generators will be allocated this - many GPUs. + many GPUs. May be fractional when ``gpu_sharing_factor > 1``. **gpus_per_group** [int]: Number of GPUs for each group in the scheduler. This can be used when diff --git a/docs/resource_manager/overview.rst b/docs/resource_manager/overview.rst index f980eca3b..656393c25 100644 --- a/docs/resource_manager/overview.rst +++ b/docs/resource_manager/overview.rst @@ -186,6 +186,13 @@ will not be scheduled unless ``match_slots`` is set to *False*: This is only recommended if not enumerating resources to slot IDs (e.g., via ``CUDA_VISIBLE_DEVICES``). + **gpu_packing** [str]: + Where to place tasks that use part of a GPU when + :ref:`GPU sharing` is enabled. ``"pack"`` fills partly used GPUs + and nodes first, keeping whole GPUs free for larger tasks. ``"spread"`` + prefers the least used GPUs and nodes, giving tasks more room. + Default: ``"pack"`` + Note that if six resource sets were requested, then they would be split three per node, even if ``split2fit`` is *False*, as this could otherwise never be scheduled. @@ -207,6 +214,138 @@ a number of resource sets. Note that persistent workers maintain their resources until they come out of a persistent state. +.. _gpu-sharing: + +Sharing GPUs between tasks +-------------------------- + +Some simulations need only part of a GPU. ``gpu_sharing_factor`` lets several such +tasks run on the same GPU at once, so a node's GPUs can serve more concurrent +simulations. + +Set the number of tasks that may share each GPU, and have the generator request +fractional GPUs through the ``num_gpus`` field: + +.. code-block:: python + :emphasize-lines: 2, 5 + + ensemble.libE_specs = LibeSpecs( + gpu_sharing_factor=2, # Up to two tasks per GPU + ) + + ensemble.gen_specs = GenSpecs(outputs=[("x", float, 2), ("num_gpus", float)], ...) + +with the generator setting, for example: + +.. code-block:: python + + H_o["num_gpus"] = [0.5, 1.0, 1.5, 2.0] + +GPUs are assigned and ``CUDA_VISIBLE_DEVICES`` (or equivalent) is set automatically, +as with whole-GPU requests. There is no need to set ``num_resource_sets``. + +What tasks can request +^^^^^^^^^^^^^^^^^^^^^^ + +The sharing factor sets the step size of GPU requests. Requests are rounded up to +the nearest step. + +.. list-table:: + :header-rows: 1 + + * - ``gpu_sharing_factor`` + - Smallest request + - Example requests + * - 1 (default) + - 1 GPU + - 1, 2, 4 + * - 2 + - 0.5 GPU + - 0.5, 1, 1.5, 2 + * - 4 + - 0.25 GPU + - 0.25, 0.5, 0.75, 1.25 + +Requests can mix whole and partial GPUs. Each task gets its whole GPUs to itself. +The fractional part of a request always comes from one GPU, never split across +several. For example, with ``gpu_sharing_factor=4``, a task requesting ``0.75`` GPUs +gets three quarters of one GPU, not shares of two or three different GPUs. + +With ``gpu_sharing_factor=2``, a task requesting ``1.5`` GPUs gets one GPU to itself +and half of a second GPU, which it may share with one other task. It sees two GPUs. + +Generators can also use a fraction of a GPU, via ``gen_num_gpus``: + +.. code-block:: python + + ensemble.libE_specs = LibeSpecs(gpu_sharing_factor=4, gen_num_gpus=0.25) + +Example: a node with four GPUs +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +With ``gpu_sharing_factor=2`` on a node with four GPUs, any of these sets of tasks +can run at once (using the default ``"pack"`` placement, below): + +* Eight tasks requesting ``0.5`` GPUs (two per GPU). +* Four tasks requesting ``0.5`` GPUs and two requesting ``1`` GPU. +* Two tasks requesting ``1.5`` GPUs and two requesting ``0.5`` GPUs. + +Run with enough workers to cover the concurrency you want. For eight concurrent +half-GPU tasks per node, use eight workers per node. + +Choosing a sharing factor +^^^^^^^^^^^^^^^^^^^^^^^^^ + +Tasks sharing a GPU are not isolated: they share its memory and compute. Choose a +factor so that this many tasks fit in one GPU's memory at once. For example, with +40 GB GPUs and tasks needing about 8 GB, a factor of up to ``4`` is safe. + +If your platform provides tools such as NVIDIA MPS for better GPU sharing, they can +be enabled as usual alongside ``gpu_sharing_factor``. + +Each task's share of CPU cores also gets smaller: libEnsemble divides each node's +cores between all GPU shares. For example, on a node with 32 cores and four GPUs, +a factor of ``2`` gives four cores per share. + +Placing partial-GPU tasks +^^^^^^^^^^^^^^^^^^^^^^^^^ + +By default, tasks using part of a GPU are packed onto GPUs that are already partly +used, keeping whole GPUs free for larger tasks. To instead spread such tasks over +the least used GPUs and nodes: + +.. code-block:: python + + ensemble.libE_specs = LibeSpecs( + gpu_sharing_factor=2, + scheduler_opts={"gpu_packing": "spread"}, + ) + +Use ``"pack"`` (the default) when tasks of different sizes are mixed. With +``"spread"``, four half-GPU tasks on a four-GPU node take one share of every GPU, +so a following task requesting a whole GPU must wait. Use ``"spread"`` when tasks +are all the same size, or when they compete for memory or bandwidth and should be +kept apart. + +Resource sets and errors +^^^^^^^^^^^^^^^^^^^^^^^^ + +When ``gpu_sharing_factor`` is greater than 1 and ``num_resource_sets`` is not set, +libEnsemble uses one resource set per GPU share and logs a message such as:: + + gpu_sharing_factor=2: num_resource_sets defaulting to 16 (2 node(s) x 4 GPUs x 2 shares) + +If you do set ``num_resource_sets``, it must provide at least one resource set per +GPU share on each node (for example, at least ``16`` in the case above). Fewer +raises an error. More is allowed, with the extra resource sets having no GPUs. + +An error is also raised if the factor would leave a resource set with no CPU cores. + +libEnsemble warns at startup if ``gpu_sharing_factor`` is set but tasks do not +request GPUs, or if no GPUs are detected. + +For an example, see `test_GPU_sharing_variable_resources.py`_. + Example scenarios ----------------- @@ -270,6 +409,7 @@ Also, this can be set on the command line as a convenience. python run_ensemble.py --nworkers 5 --nresource_sets 8 .. _persistent_sampling_var_resources.py: https://github.com/Libensemble/libensemble/blob/develop/libensemble/gen_funcs/persistent_sampling_var_resources.py +.. _test_GPU_sharing_variable_resources.py: https://github.com/Libensemble/libensemble/blob/develop/libensemble/tests/regression_tests/test_GPU_sharing_variable_resources.py .. _test_GPU_variable_resources.py: https://github.com/Libensemble/libensemble/blob/develop/libensemble/tests/regression_tests/test_GPU_variable_resources.py .. _test_persistent_sampling_CUDA_variable_resources.py: https://github.com/Libensemble/libensemble/blob/develop/libensemble/tests/functionality_tests/test_persistent_sampling_CUDA_variable_resources.py .. _var_resources.py: https://github.com/Libensemble/libensemble/blob/develop/libensemble/sim_funcs/var_resources.py diff --git a/libensemble/executors/executor.py b/libensemble/executors/executor.py index 369308ada..9aa4437a4 100644 --- a/libensemble/executors/executor.py +++ b/libensemble/executors/executor.py @@ -79,18 +79,18 @@ class Application: def __init__( self, - full_path: str, + full_path: str | os.PathLike[str], name: str | None = None, calc_type: str | None = "sim", desc: str | None = None, pyobj: Any | None = None, - precedent: str = "", + precedent: str | os.PathLike[str] = "", ) -> None: """Instantiates a new Application instance.""" - self.full_path = full_path + self.full_path = os.fspath(full_path) self.calc_type = calc_type - self.calc_dir, self.exe = os.path.split(full_path) - self.precedent = precedent + self.calc_dir, self.exe = os.path.split(self.full_path) + self.precedent = os.fspath(precedent) if precedent else "" if self.exe.endswith(".py"): if not precedent: diff --git a/libensemble/executors/mpi_runner.py b/libensemble/executors/mpi_runner.py index d44152aff..e527609bd 100644 --- a/libensemble/executors/mpi_runner.py +++ b/libensemble/executors/mpi_runner.py @@ -1,5 +1,6 @@ import argparse import logging +import math from libensemble.executors.executor import jassert from libensemble.resources import mpi_resources @@ -124,8 +125,8 @@ def _set_gpu_cli_option(self, wresources, extra_args, gpu_setting_name, gpu_valu def _set_gpu_env_var(self, wresources, task, gpus_per_node, gpus_env): """Add GPU environment variable setting to the tasks environment""" jassert(wresources.matching_slots, f"Cannot assign CPUs/GPUs to non-matching slots per node {wresources.slots}") - slot_list = wresources.get_slots_as_string(multiplier=wresources.gpus_per_rset_per_node, limit=gpus_per_node) - task._add_to_env(gpus_env, slot_list) + gpu_list = wresources.get_gpu_ids_as_string(limit=gpus_per_node) + task._add_to_env(gpus_env, gpu_list) def _local_runner_set_gpus(self, task, wresources, extra_args, gpus_per_node, ppn): """Set default GPU setting for MPI runner""" @@ -174,7 +175,7 @@ def _assign_gpus(self, task, resources, nprocs, nnodes, ppn, ngpus, extra_args, # gpus per node for this worker. if wresources.doihave_gpus(): - gpus_avail_per_node = wresources.slot_count * wresources.gpus_per_rset_per_node + gpus_avail_per_node = len(set(wresources.gpu_ids_on_node)) * wresources.gpus_per_rset_per_node else: gpus_avail_per_node = 0 @@ -188,7 +189,7 @@ def _assign_gpus(self, task, resources, nprocs, nnodes, ppn, ngpus, extra_args, nnodes = wresources.local_node_count if ngpus is not None: - gpus_req_per_node = ngpus // nnodes + gpus_req_per_node = ngpus // nnodes if isinstance(ngpus, int) else math.ceil(ngpus / nnodes) if gpus_req_per_node > gpus_avail_per_node: logger.info(f"Asked for more GPUs per node than available - max is {gpus_avail_per_node}") gpus_per_node = min(gpus_req_per_node, gpus_avail_per_node) @@ -246,8 +247,8 @@ def _get_min_nodes(self, nprocs, ppn, nnodes, ngpus, resources): if nprocs: proc_min_nodes = (nprocs + procs_on_node - 1) // procs_on_node if ngpus: - gpus_on_node = wresources.slot_count * wresources.gpus_per_rset_per_node - gpu_min_nodes = (ngpus + gpus_on_node - 1) // gpus_on_node + gpus_on_node = len(set(wresources.gpu_ids_on_node)) * wresources.gpus_per_rset_per_node + gpu_min_nodes = math.ceil(ngpus / gpus_on_node) if gpus_on_node else 1 min_nodes = max(proc_min_nodes, gpu_min_nodes) nnodes = min(min_nodes, total_nodes) @@ -261,6 +262,9 @@ def _adjust_procs(self, nprocs, ppn, nnodes, ngpus, resources): def adjust_resource(n_units, units_attr, units_name): if n_units is not None and nnodes: + # Fractional GPU requests are rounded up per node in _assign_gpus + if isinstance(n_units, float) and not n_units.is_integer(): + return n_units mod_n_units = n_units % nnodes if mod_n_units != 0: try_n_units = n_units + (nnodes - mod_n_units) diff --git a/libensemble/manager.py b/libensemble/manager.py index dc42975fe..ccd1a8a7d 100644 --- a/libensemble/manager.py +++ b/libensemble/manager.py @@ -279,6 +279,9 @@ def __init__( if self.resources is not None: gresource = self.resources.glob_resources self.scheduler_opts = gresource.update_scheduler_opts(self.scheduler_opts) + if self.resources.resource_manager is not None and self.resources.resource_manager.rsets_sizing_msg: + logger.info(self.resources.resource_manager.rsets_sizing_msg) + self._check_gpu_sharing(gresource) for wrk in self.W: if wrk["worker_id"] in self.libE_specs.get("gen_workers", []): @@ -296,6 +299,20 @@ def __init__( e, ) + def _check_gpu_sharing(self, gresource) -> None: + """Warn about gpu_sharing_factor configurations that will not share GPUs as intended""" + sf = gresource.gpu_sharing_factor + if sf <= 1: + return + if not gresource.gpus_avail_per_node: + logger.manager_warning(f"gpu_sharing_factor={sf} is set, but no GPUs were detected. It will be ignored.") + elif "num_gpus" not in self.hist.H.dtype.names and not self.gen_num_gpus: + logger.manager_warning( + f"gpu_sharing_factor={sf} is set, but tasks do not request GPUs. For best GPU utilization, " + "tasks should request GPUs via a 'num_gpus' field in gen_specs['out'] (or 'gen_num_gpus' " + "for generators). Without GPU requests, tasks are not placed by GPU share and GPUs may be underused." + ) + # --- Termination logic routines def term_test_wallclock(self, max_elapsed: int) -> bool: @@ -1116,15 +1133,11 @@ def run(self, persis_info: dict) -> tuple[dict, int, int]: if self._service_idle_start is None: self._service_idle_start = time.time() elif time.time() - self._service_idle_start > idle_timeout: - logger.info( - f"service_mode idle for {idle_timeout}s with no work, exiting" - ) + logger.info(f"service_mode idle for {idle_timeout}s with no work, exiting") break time.sleep(0.1) continue - raise AssertionError( - "alloc_f did not return any work, although all workers are idle." - ) + raise AssertionError("alloc_f did not return any work, although all workers are idle.") # work was dispatched OR workers active OR terminating — reset idle clock self._service_idle_start = None except WorkerException as e: diff --git a/libensemble/resources/resources.py b/libensemble/resources/resources.py index 9b68e43f0..7bd7e008e 100644 --- a/libensemble/resources/resources.py +++ b/libensemble/resources/resources.py @@ -164,6 +164,7 @@ def __init__(self, libE_specs: dict, platform_info: dict = {}, top_level_dir: st self.num_resource_sets = libE_specs.get("num_resource_sets", None) self.enforce_worker_core_bounds = libE_specs.get("enforce_worker_core_bounds", False) self.gpus_per_group = libE_specs.get("gpus_per_group") + self.gpu_sharing_factor = libE_specs.get("gpu_sharing_factor", 1) resource_info = libE_specs.get("resource_info", {}) # resource_info overrides platform diff --git a/libensemble/resources/rset_resources.py b/libensemble/resources/rset_resources.py index e43a51b88..82cd32e52 100644 --- a/libensemble/resources/rset_resources.py +++ b/libensemble/resources/rset_resources.py @@ -13,6 +13,10 @@ # logger.setLevel(logging.DEBUG) +class RSetResourcesException(Exception): + """Resource sets cannot be created for the given configuration.""" + + class RSetResources: """A class that creates a fixed mapping of resource sets to the available resources. @@ -33,6 +37,8 @@ class RSetResources: ("group", int), # Group ID this resource set belongs to ("slot", int), # Slot ID this resource set belongs to ("gpus", bool), # Does this resource set have GPUs + ("gpu_id", int), # Physical GPU ID for GPU resource sets + ("gpu_share", int), # Logical share within a physical GPU # ('pool', int), # Pool ID (eg. separate gen/sim resources) - not yet used. ] @@ -42,6 +48,8 @@ def __init__(self, num_workers: int, resources: GlobalResources): Determines the compute resources available for each resource set. Unless resource sets is set explicitly, the number of resource sets is the number of workers. + If ``gpu_sharing_factor > 1`` and GPUs are available, the default is instead one resource set + per GPU share: ``nodes * gpus_per_node * gpu_sharing_factor``. Parameters ---------- @@ -55,26 +63,65 @@ def __init__(self, num_workers: int, resources: GlobalResources): """ self.num_workers = num_workers self.num_workers_2assign2 = RSetResources.get_workers2assign2(self.num_workers, resources) - self.total_num_rsets = resources.num_resource_sets or self.num_workers_2assign2 self.num_nodes = len(resources.global_nodelist) + self.gpu_sharing_factor = resources.gpu_sharing_factor + gpus_avail_per_node = resources.gpus_avail_per_node + num_gpu_shares_per_node = gpus_avail_per_node * self.gpu_sharing_factor + self.rsets_sizing_msg = None + self.rsets_sized_by_gpu_sharing = ( + not resources.num_resource_sets and self.gpu_sharing_factor > 1 and gpus_avail_per_node > 0 + ) + if self.rsets_sized_by_gpu_sharing: + self.total_num_rsets = self.num_nodes * num_gpu_shares_per_node + self.rsets_sizing_msg = ( + f"gpu_sharing_factor={self.gpu_sharing_factor}: num_resource_sets defaulting to " + f"{self.total_num_rsets} ({self.num_nodes} node(s) x {gpus_avail_per_node} GPUs " + f"x {self.gpu_sharing_factor} shares)" + ) + else: + self.total_num_rsets = resources.num_resource_sets or self.num_workers_2assign2 self.split_list, self.local_rsets_list = RSetResources.get_partitioned_nodelist(self.total_num_rsets, resources) self.nodes_in_rset = len(self.split_list[0]) - gpus_avail_per_node = resources.gpus_avail_per_node self.rsets_per_node = RSetResources.get_rsets_on_a_node(self.total_num_rsets, resources) - self.gpu_rsets_per_node = min(gpus_avail_per_node, self.rsets_per_node) + if self.gpu_sharing_factor > 1 and gpus_avail_per_node: + RSetResources.check_gpu_sharing_rsets( + self.total_num_rsets, self.num_nodes, self.rsets_per_node, gpus_avail_per_node, self.gpu_sharing_factor + ) + self.gpu_rsets_per_node = min(gpus_avail_per_node * self.gpu_sharing_factor, self.rsets_per_node) self.nongpu_rsets_per_node = self.rsets_per_node - self.gpu_rsets_per_node self.all_rsets = np.zeros(self.total_num_rsets, dtype=RSetResources.rset_dtype) self.all_rsets["group"], self.all_rsets["slot"], self.all_rsets["gpus"] = RSetResources.get_group_list( - self.split_list, gpus_avail_per_node, resources.gpus_per_group + self.split_list, gpus_avail_per_node, resources.gpus_per_group, self.gpu_sharing_factor ) + if gpus_avail_per_node: + gpu_positions = RSetResources.get_gpu_positions(self.split_list) + self.all_rsets["gpu_id"] = np.where( + self.all_rsets["gpus"], + (gpu_positions // self.gpu_sharing_factor) % gpus_avail_per_node, + -1, + ) + self.all_rsets["gpu_share"] = np.where(self.all_rsets["gpus"], gpu_positions % self.gpu_sharing_factor, -1) self.total_num_gpu_rsets = np.count_nonzero(self.all_rsets["gpus"]) self.total_num_nongpu_rsets = np.count_nonzero(~self.all_rsets["gpus"]) - self.gpus_per_rset_per_node = gpus_avail_per_node // self.gpu_rsets_per_node if self.gpu_rsets_per_node else 0 + if self.gpu_rsets_per_node: + self.gpus_per_rset_per_node = ( + 1 + if self.gpu_sharing_factor > 1 + else gpus_avail_per_node // min(gpus_avail_per_node, self.rsets_per_node) + ) + else: + self.gpus_per_rset_per_node = 0 self.cores_per_rset_per_node = resources.physical_cores_avail_per_node // self.rsets_per_node + if self.cores_per_rset_per_node == 0 and self.gpu_sharing_factor > 1: + raise RSetResourcesException( + f"gpu_sharing_factor={self.gpu_sharing_factor} gives {self.rsets_per_node} resource sets per node, " + f"but there are only {resources.physical_cores_avail_per_node} CPU cores per node. " + "Reduce gpu_sharing_factor so that each resource set has at least one core." + ) # Oversubsribe if self.cores_per_rset_per_node == 0: @@ -89,25 +136,50 @@ def __init__(self, num_workers: int, resources: GlobalResources): self.procs_per_rset = self.procs_per_rset_per_node * self.nodes_in_rset @staticmethod - def get_group_list(split_list, gpus_per_node=0, gpus_per_group=None): - """Return lists of group ids and slot IDs by resource set""" + def check_gpu_sharing_rsets(total_num_rsets, num_nodes, rsets_per_node, gpus_per_node, gpu_sharing_factor): + """Check that resource sets cover every GPU share on each node""" + shares_per_node = gpus_per_node * gpu_sharing_factor + if rsets_per_node < shares_per_node: + raise RSetResourcesException( + f"gpu_sharing_factor={gpu_sharing_factor} with {gpus_per_node} GPUs per node needs " + f"{shares_per_node} resource sets per node ({shares_per_node * num_nodes} total on {num_nodes} " + f"node(s)), but num_resource_sets={total_num_rsets} gives only {rsets_per_node} per node. " + f"Set num_resource_sets to at least {shares_per_node * num_nodes}, or leave it unset." + ) + + @staticmethod + def get_group_list(split_list, gpus_per_node=0, gpus_per_group=None, gpu_sharing_factor=1): + """Return group, slot, and GPU lists by resource set""" group = 1 slot = 0 group_list, slot_list, gpu_list = [], [], [] node = split_list[0] for i in range(len(split_list)): - # still break on new node if gpus_per_group is set - if split_list[i] != node or slot == gpus_per_group: + # break on new node or when group size reached (account for sharing) + group_threshold = gpus_per_group * gpu_sharing_factor if gpus_per_group is not None else None + if split_list[i] != node or slot == group_threshold: node = split_list[i] group += 1 slot = 0 group_list.append(group) slot_list.append(slot) - gpu_list.append(slot < gpus_per_node) + is_gpu = slot < gpus_per_node * gpu_sharing_factor + gpu_list.append(is_gpu) slot += 1 return group_list, slot_list, gpu_list + @staticmethod + def get_gpu_positions(split_list): + """Return each resource set's node-local position, independent of scheduler groups.""" + positions = [] + node_counts = {} + for nodes in split_list: + node = nodes[0] + positions.append(node_counts.get(node, 0)) + node_counts[node] = positions[-1] + 1 + return np.asarray(positions) + @staticmethod def best_split(a, n): """Creates the most even split of list a into n parts and return list of lists""" diff --git a/libensemble/resources/scheduler.py b/libensemble/resources/scheduler.py index 2e75e3c84..c0b63824e 100644 --- a/libensemble/resources/scheduler.py +++ b/libensemble/resources/scheduler.py @@ -57,6 +57,13 @@ def __init__(self, user_resources=None, sched_opts={}): Useful if setting an environment variable such as ``CUDA_VISIBLE_DEVICES`` to specific slots, which should match over multiple nodes. Default: True + + "gpu_packing" [String]: + Where to place tasks that use part of a GPU when ``gpu_sharing_factor > 1``. + "pack" fills partly used GPUs and nodes first, keeping whole GPUs free for + larger tasks. "spread" prefers the least used GPUs and nodes, giving tasks + more room. + Default: "pack" """ self.resources = user_resources or Resources.resources.resource_manager @@ -71,6 +78,9 @@ def __init__(self, user_resources=None, sched_opts={}): # Process scheduler options self.split2fit = sched_opts.get("split2fit", True) self.match_slots = sched_opts.get("match_slots", True) + self.gpu_packing = sched_opts.get("gpu_packing", "pack") + if self.gpu_packing not in ("pack", "spread"): + raise ValueError(f"gpu_packing must be 'pack' or 'spread', got {self.gpu_packing!r}") self.last_use_gpus = None def assign_resources(self, rsets_req, use_gpus=None, user_params=[]): @@ -93,6 +103,10 @@ def assign_resources(self, rsets_req, use_gpus=None, user_params=[]): self.check_total_rsets(rsets_req, use_gpus) self.log_msg = None # Log resource messages only when find resources + + if use_gpus and self.resources.gpu_sharing_factor > 1: + return self._assign_gpu_sharing_resources(rsets_req, user_params) + num_groups = self.resources.num_groups max_grpsize = self.resources.rsets_per_node # assumes even @@ -180,6 +194,104 @@ def assign_resources(self, rsets_req, use_gpus=None, user_params=[]): return rset_team + def _gpu_bins(self, groups, valid_rsets_by_group): + """Return GPU bins of shares whose slots are free in all ``groups``. + + Requiring the same slots in every group gives matching slots, and therefore the + same physical GPU IDs, on each node. The GPU index within a group is + ``slot // gpu_sharing_factor``. + + Returns ``(bins, slot_maps)``. ``bins`` maps a GPU index to a list of free slots + on that GPU. ``slot_maps`` contains a ``slot -> rset`` dictionary for each group. + """ + rsets = self.resources.rsets + sf = self.resources.gpu_sharing_factor + slot_maps = [{int(rsets[r]["slot"]): r for r in valid_rsets_by_group[g]} for g in groups] + bins = {} + for slot in sorted(set.intersection(*(set(m) for m in slot_maps))): + bins.setdefault(slot // sf, []).append(slot) + return bins, slot_maps + + def _pick_from_gpu_bins(self, bins, need): + """Pick ``need`` shares from ``bins`` with physical-exclusive semantics. + + Whole GPUs (``need // gpu_sharing_factor``) must be fully free. The remainder + must come from a single physical GPU so every share is visible to the task. + Returns a list of selected slots, or None if not possible. + """ + sf = self.resources.gpu_sharing_factor + whole, rem = divmod(need, sf) + fully_free = sorted(gid for gid, slots in bins.items() if len(slots) == sf) + if len(fully_free) < whole: + return None + chosen = fully_free[:whole] + selected = [slot for gid in chosen for slot in bins[gid]] + if rem: + candidates = [(len(p), gid) for gid, p in bins.items() if gid not in chosen and len(p) >= rem] + if not candidates: + return None + sign = 1 if self.gpu_packing == "pack" else -1 + _, gid = min(candidates, key=lambda c: (sign * c[0], c[1])) + selected.extend(bins[gid][:rem]) + return selected + + def _pick_team_for_groups(self, groups, need, valid_rsets_by_group): + """Return an rset team giving ``need`` shares in each group, or None""" + group_sets = [groups] if self.match_slots else [[g] for g in groups] + team = [] + for gset in group_sets: + bins, slot_maps = self._gpu_bins(gset, valid_rsets_by_group) + slots = self._pick_from_gpu_bins(bins, need) + if slots is None: + return None + team.extend(m[slot] for m in slot_maps for slot in slots) + return sorted(team) + + def _gpu_share_splits(self, rsets_req, num_groups_req, rsets_per_group, num_groups, user_params): + """Yield candidate (rsets_req, num_groups, rsets_per_group) splits in order of preference""" + yield rsets_req, num_groups_req, rsets_per_group + if self.split2fit: + for ngroups in range(num_groups_req + 1, num_groups + 1): + if rsets_req % ngroups == 0 and self.check_params(user_params, ngroups): + yield rsets_req, ngroups, rsets_req // ngroups + + def _assign_gpu_sharing_resources(self, rsets_req, user_params): + """Physical-exclusive GPU-share-aware assignment with packing""" + avail_rsets_by_group = self.get_avail_rsets_by_group() + valid_rsets_by_group = self.filter_for_rset_type(avail_rsets_by_group, True) + rsets_req, num_groups_req, rsets_per_group = self.calc_req_split( + rsets_req, self.resources.gpu_rsets_per_node, self.resources.num_groups, extend=True + ) + sign = 1 if self.gpu_packing == "pack" else -1 + groups_by_pref = sorted(valid_rsets_by_group, key=lambda g: (sign * len(valid_rsets_by_group[g]), g)) + + rset_team = None + for rsets_req, ngroups, need in self._gpu_share_splits( + rsets_req, num_groups_req, rsets_per_group, self.resources.num_groups, user_params + ): + groups = [] + for g in groups_by_pref: + if self._pick_team_for_groups(groups + [g], need, valid_rsets_by_group) is not None: + groups.append(g) + if len(groups) == ngroups: + rset_team = self._pick_team_for_groups(groups, need, valid_rsets_by_group) + break + if rset_team is not None: + break + if rset_team is None: + raise InsufficientFreeResources + + self.avail_rsets_by_group = self.filter_out_rset_team(avail_rsets_by_group, rset_team) + self.rsets_free -= len(rset_team) + self.gpu_rsets_free -= len(rset_team) + if self.log_msg is not None: + logger.debug(self.log_msg) + logger.debug( + f"rset_team found (gpu-sharing): Req: {rsets_req} rsets. Found: {rset_team}" + + f" Avail sets {self.avail_rsets_by_group}" + ) + return rset_team + def find_rsets_any_slots(self, valid_rsets_by_group, max_grpsize, rsets_req, ngroups, rsets_per_group): """Find optimal non-matching slots across groups""" tmp_rsets_by_group = copy.deepcopy(valid_rsets_by_group) diff --git a/libensemble/resources/worker_resources.py b/libensemble/resources/worker_resources.py index 0207b9089..ff815f80b 100644 --- a/libensemble/resources/worker_resources.py +++ b/libensemble/resources/worker_resources.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import math import os from collections import Counter, OrderedDict from typing import TYPE_CHECKING @@ -198,6 +199,7 @@ def __init__(self, num_workers, resources, workerID): self.matching_slots = True self.slot_count = None self.slots_on_node = None + self.gpu_ids_on_node = None self.local_node_count = len(self.local_nodelist) self.set_slot_count() self.gen_nprocs = None @@ -252,6 +254,27 @@ def set_env_to_slots(self, env_var, multiplier=1, delimiter=","): """ os.environ[env_var] = self.get_slots_as_string(multiplier, delimiter) + def get_gpu_ids_as_string(self, delimiter=",", limit=None): + """Returns physical GPU IDs for the worker's assigned GPU resource sets. + + :param delimiter: Optional str. Delimiter for output string. + :param limit: Optional int or float. Maximum GPUs (a fractional limit is rounded up). + """ + if self.gpu_ids_on_node is None: + logger.warning("GPU IDs on node is None when requested as a string") + return None + gpu_ids = [] + for gpu_id in self.gpu_ids_on_node: + if gpu_id >= 0: + gpu_ids.extend(range(gpu_id * self.gpus_per_rset_per_node, (gpu_id + 1) * self.gpus_per_rset_per_node)) + gpu_ids = list(OrderedDict.fromkeys(gpu_ids)) + if limit is not None: + gpu_ids = gpu_ids[: math.ceil(limit)] + if self.tiles_per_gpu > 1: + ntiles = self.tiles_per_gpu + gpu_ids = [f"{i // ntiles}.{i % ntiles}" for i in gpu_ids] + return delimiter.join(map(str, gpu_ids)) + def set_env_to_gpus(self, env_var=None, delimiter=","): """Sets the given environment variable to GPUs @@ -276,7 +299,7 @@ def set_env_to_gpus(self, env_var=None, delimiter=","): """ assert self.matching_slots, f"Cannot assign GPUs to non-matching slots per node {self.slots}" if self.doihave_gpus(): - env_value = self.get_slots_as_string(multiplier=self.gpus_per_rset_per_node, limit=self.gen_ngpus) + env_value = self.get_gpu_ids_as_string(limit=self.gen_ngpus) if env_var is None: if self.platform_info is not None: if self.platform_info.get("gpu_setting_type") == "env": @@ -315,6 +338,9 @@ def set_rset_team(self, rset_team: list[int]) -> None: self.split_list, self.rsets_per_node, ) + self.gpu_ids_on_node = WorkerResources.get_local_gpu_ids( + self.rset_team, self.all_rsets, self.split_list, self.local_nodelist[0] if self.local_nodelist else None + ) self.set_slot_count() self.local_node_count = len(self.local_nodelist) @@ -346,6 +372,15 @@ def set_slot_count(self) -> None: self.slots_on_node = first_node_slots if self.matching_slots else None self.slot_count = first_len if self.even_slots else None + @staticmethod + def get_local_gpu_ids(rset_team: list[int], all_rsets, split_list, node) -> list[int]: + """Return physical GPU IDs on the worker's first local node.""" + return [ + int(all_rsets["gpu_id"][rset]) + for rset in rset_team + if all_rsets["gpus"][rset] and (node is None or split_list[rset][0] == node) + ] + @staticmethod def get_local_nodelist( workerID: int, rset_team: list[int], split_list: list[list[str]], rsets_per_node: int diff --git a/libensemble/specs.py b/libensemble/specs.py index d87030829..e8af8cc6b 100644 --- a/libensemble/specs.py +++ b/libensemble/specs.py @@ -775,9 +775,34 @@ def set_calc_dirs_on_input_dir(self): num_resource_sets: int | None = 0 """ Total number of resource sets. Resources will be divided into this number. - If not set, resources will be divided evenly by the number of workers. + If not set, resources will be divided evenly by the number of workers, or, + if ``gpu_sharing_factor > 1``, into one resource set per GPU share. """ + gpu_sharing_factor: int = 1 + """ + Number of tasks that may share each GPU. For example, ``2`` lets two tasks + requesting ``num_gpus=0.5`` run on the same GPU at the same time, and ``4`` + allows requests in steps of a quarter GPU. The default of ``1`` gives each + task exclusive use of its GPUs. + + Tasks should request GPUs via a ``num_gpus`` field in ``gen_specs["out"]`` + (or ``gen_num_gpus``). Requests are rounded up to the nearest share. Any part + of a request smaller than a whole GPU is placed on a single GPU. + + Tasks sharing a GPU are not isolated from each other: they share its memory + and compute. Use a factor that suits the memory needs of your tasks. + + If ``num_resource_sets`` is not set, it defaults to one resource set per + GPU share (``nodes * gpus_per_node * gpu_sharing_factor``). + """ + + @field_validator("gpu_sharing_factor") + def check_gpu_sharing_factor(cls, value): + if value < 1: + raise ValueError("gpu_sharing_factor must be at least 1") + return value + gen_num_procs: int | None = 0 """ The default number of processors (MPI ranks) required by generators. Unless @@ -785,11 +810,13 @@ def set_calc_dirs_on_input_dir(self): allocated this many processors for applications launched via the MPIExecutor. """ - gen_num_gpus: int | None = 0 + gen_num_gpus: int | float | None = 0 """ The default number of GPUs required by generators. Unless overridden by the equivalent `persis_info` settings, generators will be allocated this many GPUs. + Fractional values (e.g., ``0.5``) share a GPU when ``gpu_sharing_factor > 1``, + and are otherwise rounded up to whole GPUs. """ gpus_per_group: int | None = None diff --git a/libensemble/tests/regression_tests/test_GPU_sharing_variable_resources.py b/libensemble/tests/regression_tests/test_GPU_sharing_variable_resources.py new file mode 100644 index 000000000..f1b3eead0 --- /dev/null +++ b/libensemble/tests/regression_tests/test_GPU_sharing_variable_resources.py @@ -0,0 +1,123 @@ +""" +Tests GPU sharing with physical-exclusive scheduling and pack/spread policies. + +This regression test mimics a user workflow: a persistent generator produces +simulations that request fractional GPUs (0.5, 1.0, 1.5, 2.0) which, with +``gpu_sharing_factor=2``, map to 1-4 logical GPU shares. The ensemble is +configured with 2 mocked nodes, 4 GPUs per node, and ``scheduler_opts`` to +exercise pack/spread packing. The simulator (``gpu_variable_resources_from_gen``) +validates GPU assignment via ``check_gpu_setting``. + +Execute via (e.g. 9 workers): + mpiexec -np 10 python test_GPU_sharing_variable_resources.py + python test_GPU_sharing_variable_resources.py --nworkers 9 +""" + +# Do not change these lines - they are parsed by run-tests.sh +# TESTSUITE_COMMS: mpi local +# TESTSUITE_NPROCS: 10 + +import numpy as np + +from libensemble import Ensemble +from libensemble.executors.mpi_executor import MPIExecutor +from libensemble.message_numbers import EVAL_GEN_TAG, FINISHED_PERSISTENT_GEN_TAG, PERSIS_STOP, STOP_TAG +from libensemble.sim_funcs import six_hump_camel +from libensemble.sim_funcs.var_resources import gpu_variable_resources_from_gen as sim_f +from libensemble.specs import GenSpecs, LibeSpecs, SimSpecs +from libensemble.tools.persistent_support import PersistentSupport + + +def uniform_sample_sharing(_, persis_info, gen_specs, libE_info): + """Persistent generator that cycles through fractional GPU requests. + + Each batch emits ``num_gpus`` values from the repeating cycle + ``[0.5, 1.0, 1.5, 2.0]`` which, with ``gpu_sharing_factor=2``, map to + ``[1, 2, 3, 4]`` logical shares. ``0.5`` exercises the fractional + (partial-GPU) path, ``1.0`` a whole single GPU, ``1.5`` whole+remainder, + and ``2.0`` an exclusive two-GPU request. ``num_procs`` is fixed at 1 so + GPU-driven resource sizing dominates. The generator also draws ``x`` + uniformly from ``gen_specs["user"]["lb"/"ub"]`` so the sim can run + ``six_hump_camel``. Used by the ensemble integration portion of the + test to drive all share-aware branches in ``ResourceScheduler`` and + ``AllocSupport``. + """ + # Extract domain and batch size from gen_specs + ub = gen_specs["user"]["ub"] + lb = gen_specs["user"]["lb"] + n = len(lb) + b = gen_specs["initial_batch_size"] + from libensemble.tools import get_rng + + # Set up RNG and persistent support channel to manager + rng = get_rng(gen_specs, libE_info) + ps = PersistentSupport(libE_info, EVAL_GEN_TAG) + tag = None + # Cycle covers all interesting share counts: 1,2,3,4 shares + cycle = [0.5, 1.0, 1.5, 2.0] + idx = 0 + while tag not in [STOP_TAG, PERSIS_STOP]: + # Build output array for this batch + H_o = np.zeros(b, dtype=gen_specs["out"]) + H_o["x"] = rng.uniform(lb, ub, (b, n)) + # Round-robin through fractional cycle so every batch mixes types + ngpus = [cycle[(idx + i) % len(cycle)] for i in range(b)] + idx += b + H_o["num_gpus"] = ngpus + H_o["num_procs"] = np.ones(b, dtype=int) # keep CPU sizing trivial + print(f"GEN created {b} sims requiring {ngpus} GPUs (shares {[int(v * 2) for v in ngpus]})", flush=True) + tag, Work, calc_in = ps.send_recv(H_o) + if hasattr(calc_in, "__len__"): + b = len(calc_in) + return None, persis_info, FINISHED_PERSISTENT_GEN_TAG + + +if __name__ == "__main__": + # Register dummy app for MPIExecutor (six_hump_camel is Python, no GPU needed) + six_hump_camel_app = six_hump_camel.__file__ + exctr = MPIExecutor() + exctr.register_app(full_path=six_hump_camel_app, app_name="six_hump_camel") + + # Ensemble setup: parse --nworkers/--comms + gpu_test = Ensemble(parse_args=True, executor=exctr) + nworkers = gpu_test.nworkers + # num_resource_sets is not set: with gpu_sharing_factor=2 it defaults to one + # resource set per GPU share (4 GPUs x 2 shares = 8 per node) + gpu_test.libE_specs = LibeSpecs( + resource_info={"cores_on_node": (32, 64), "gpus_on_node": 4}, + gpu_sharing_factor=2, + scheduler_opts={"gpu_packing": "pack"}, + sim_dirs_make=True, + reuse_output_dir=True, + ensemble_dir_path="./ensemble_GPU_sharing_variable_w" + str(nworkers) + str(gpu_test.libE_specs.comms), + ) + + # Simulator: auto-assign GPUs, validates via check_gpu_setting in var_resources + gpu_test.sim_specs = SimSpecs( + sim_f=sim_f, + inputs=["x"], + out=[("f", float)], + user={"dry_run": False}, # set True to only log runlines without executing + ) + # Generator: persistent, cycles fractional num_gpus (float) to drive sharing logic + gpu_test.gen_specs = GenSpecs( + gen_f=uniform_sample_sharing, + persis_in=["f", "x", "sim_id"], + out=[("num_procs", int), ("num_gpus", float), ("x", float, 2)], + initial_batch_size=nworkers - 1, # fill all sim workers initially + batch_evaluate_same_priority=False, + async_return=False, + user={ + "lb": np.array([-3, -2]), + "ub": np.array([3, 2]), + }, + ) + + if gpu_test.ready(): + gpu_test.run(sim_max=12, wallclock_max=300) # 12 sims covers 3 full cycles + + # Manager post-checks: flag 0 and save history + if gpu_test.is_manager: + assert gpu_test.flag == 0 # ensemble completed without manager exception + gpu_test.save_output(__file__) + print("GPU sharing regression test passed", flush=True) diff --git a/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/cleanup.sh b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/cleanup.sh new file mode 100644 index 000000000..5ab7dc41e --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/cleanup.sh @@ -0,0 +1 @@ +rm -r ensemble *.npy *.pickle ensemble.log lib*.txt 2>/dev/null; true diff --git a/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/forces_simf.py b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/forces_simf.py new file mode 100644 index 000000000..146b3bda6 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/forces_simf.py @@ -0,0 +1,70 @@ +import math + +import numpy as np + +from libensemble.message_numbers import TASK_FAILED, WORKER_DONE +from libensemble.resources.resources import Resources +from libensemble.tools.test_support import check_gpu_setting + + +def run_forces(H, persis_info, sim_specs, libE_info): + """PBSpro+CUDA with gpu_sharing_factor: validate assignment/splitting on real HW.""" + + calc_status = 0 + particles = str(int(H["x"][0][0])) + args = particles + " " + str(10) + " " + particles + + exctr = libE_info["executor"] + task = exctr.submit( + app_name="forces", + app_args=args, + auto_assign_gpus=True, + match_procs_to_gpus=True, + ) + task.wait() + check_gpu_setting(task, assert_setting=False, print_setting=True) + + try: + wres = Resources.resources.worker_resources + sf = sim_specs["user"].get("gpu_sharing_factor", 1) + req_gpus = float(H["num_gpus"][0]) if "num_gpus" in H.dtype.names else 1.0 + gpus_per_rset = getattr(wres, "gpus_per_rset", 1) or 1 + expected_shares = math.ceil(req_gpus * sf / gpus_per_rset) if gpus_per_rset else math.ceil(req_gpus * sf) + assert 1 <= wres.num_rsets <= 4, f"num_rsets {wres.num_rsets} out of range for {req_gpus}" + # Allow off-by-one due to packing/splitting nuances + # mpi_runner may inflate fractional gpus to even across nodes (0.5->1.0,1.5->2.0) + assert ( + wres.num_rsets >= expected_shares + ), f"num_rsets {wres.num_rsets} < expected {expected_shares} for {req_gpus}" + assert wres.num_rsets <= max(4, expected_shares * 2), f"num_rsets {wres.num_rsets} too large for {req_gpus}" + for gid in wres.gpu_ids_on_node: + assert gid >= 0, f"negative gpu_id {gid}" + gpu_shares_per_node = getattr(wres, "gpu_rsets_per_node", 0) or 0 + if gpu_shares_per_node and expected_shares <= gpu_shares_per_node: + assert wres.local_node_count >= 1, f"local_node_count {wres.local_node_count} must be >=1" + elif gpu_shares_per_node: + expected_nodes = math.ceil(expected_shares / gpu_shares_per_node) + assert ( + 1 <= wres.local_node_count <= expected_nodes + 1 + ), f"nodes {wres.local_node_count} vs expected {expected_nodes}" + assert wres.matching_slots, f"Unmatching slots {wres.slots}" + + except AssertionError as e: + print(f"Worker {libE_info.get('workerID')} assignment warning: {e}", flush=True) + # Do not fail task for scheduler inflation; just log + except Exception as e: + print(f"Worker validation skipped: {e}", flush=True) + + statfile = "forces.stat" + try: + data = np.loadtxt(statfile) + final_energy = data[-1] + if calc_status != TASK_FAILED: + calc_status = WORKER_DONE + except Exception: + final_energy = np.nan + calc_status = TASK_FAILED + + output = np.zeros(1, dtype=sim_specs["out"]) + output["energy"] = final_energy + return output, persis_info, calc_status diff --git a/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/readme.md b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/readme.md new file mode 100644 index 000000000..de2471ba2 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/readme.md @@ -0,0 +1,26 @@ +## PBSpro + CUDA (gpu_sharing_factor) + +Fractional-GPU forces example for PBSpro systems (Aurora/Polaris) on **real NVIDIA GPUs** - no mocking. + +`gpu_sharing_factor=2` splits each NVIDIA GPU into 2 logical shares (4 GPUs/node → 8 shares/node). Generator cycles `0.5/1.0/1.5/2.0` GPUs → 1-4 shares; `pack` packing co-locates fractional tasks. PBS nodelist (`PBS_NODEFILE`) and GPUs are auto-detected. + +### Build + +In `../forces_app`: + +```bash +# Aurora (Intel oneAPI) +mpicc -DGPU -O3 -fiopenmp -fopenmp-targets=spir64 -o forces.x forces.c +# Polaris (NVHPC) +mpicc -DGPU -O3 -fopenmp -mp=gpu -o forces.x forces.c +``` + +Requires `forces.x` with `-DGPU` and real NVIDIA GPUs in allocation. + +### Run + +```bash +qsub submit_pbs.sh # select=1-2, n=5-9, mpirun -np N python run_libe_forces.py +``` + +Assertions (in sim + manager) verify `num_rsets==ceil(req*sf)`, `gpu_id` valid, `local_node_count` splitting (≤ shares/node → 1 node, else ceil), `matching_slots`, and `rsets_free`/`gpu_rsets` invariants. diff --git a/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/run_libe_forces.py b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/run_libe_forces.py new file mode 100644 index 000000000..9bfb12ec1 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/run_libe_forces.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python +""" +PBSpro + CUDA forces example testing ``gpu_sharing_factor`` on real hardware. + +Fractional GPU requests (0.5/1.0/1.5/2.0 -> 1-4 shares with sf=2) exercise +physical-exclusive packing (``gpu_packing=pack``). PBS nodelist and NVIDIA +GPUs are auto-detected from ``PBS_NODEFILE`` / platform detection - no +mocking. ``CUDA_VISIBLE_DEVICES`` is set via ``auto_assign_gpus``. + +Build ``forces.x`` with CUDA for the target (e.g. Aurora/Polaris):: + + cd ../forces_app && ./build_forces.sh # -DGPU line + +Run on PBS allocation (e.g. Polaris ``select=2``):: + + qsub submit_pbs.sh +""" + +# Do not change these lines - they are parsed by run-tests.sh +# TESTSUITE_COMMS: mpi local +# TESTSUITE_NPROCS: 5 + +import math +import sys +from pathlib import Path + +import numpy as np +from forces_simf import run_forces + +from libensemble import Ensemble +from libensemble.alloc_funcs.start_only_persistent import only_persistent_gens as alloc_f +from libensemble.executors import MPIExecutor +from libensemble.message_numbers import EVAL_GEN_TAG, FINISHED_PERSISTENT_GEN_TAG, PERSIS_STOP, STOP_TAG +from libensemble.specs import AllocSpecs, GenSpecs, LibeSpecs, SimSpecs +from libensemble.tools.persistent_support import PersistentSupport + +GPU_SHARING_FACTOR = 2 + + +def uniform_sample_sharing_pbs(_, persis_info, gen_specs, libE_info): + """Cycle 0.5,1.0,1.5,2.0 GPUs -> 1,2,3,4 shares with sf=2.""" + ub = gen_specs["user"]["ub"] + lb = gen_specs["user"]["lb"] + n = len(lb) + b = gen_specs["initial_batch_size"] + from libensemble.tools import get_rng + + rng = get_rng(gen_specs, libE_info) + ps = PersistentSupport(libE_info, EVAL_GEN_TAG) + tag = None + cycle = [0.5, 1.0, 1.5, 2.0] + idx = 0 + while tag not in [STOP_TAG, PERSIS_STOP]: + H_o = np.zeros(b, dtype=gen_specs["out"]) + H_o["x"] = rng.uniform(lb, ub, (b, n)) + H_o["num_gpus"] = [cycle[(idx + i) % len(cycle)] for i in range(b)] + H_o["num_procs"] = np.ones(b, dtype=int) + idx += b + tag, Work, calc_in = ps.send_recv(H_o) + if hasattr(calc_in, "__len__"): + b = len(calc_in) + return None, persis_info, FINISHED_PERSISTENT_GEN_TAG + + +if __name__ == "__main__": + exctr = MPIExecutor() + sim_app = Path.cwd() / "../forces_app/forces.x" + if not sys.platform.startswith("win") and not Path(sim_app).is_file(): + sys.exit("forces.x not found - please build first in ../forces_app dir") + exctr.register_app(full_path=sim_app, app_name="forces") + + ensemble = Ensemble(parse_args=True, executor=exctr) + nsim_workers = ensemble.nworkers - 1 + + # Real PBS detection: PBS_NODEFILE + NVIDIA GPUs auto-detected. No resource_info mocking. + # Ensure enough shares for max 2.0 GPU (4 shares) even on 1-node allocation + num_rsets = max(8, nsim_workers * GPU_SHARING_FACTOR) + ensemble.libE_specs = LibeSpecs( + num_resource_sets=num_rsets, + gpu_sharing_factor=GPU_SHARING_FACTOR, + scheduler_opts={"gpu_packing": "pack"}, + sim_dirs_make=True, + stats_fmt={"show_resource_sets": True}, + ) + + ensemble.sim_specs = SimSpecs( + sim_f=run_forces, + inputs=["x"], + outputs=[("energy", float)], + user={"gpu_sharing_factor": GPU_SHARING_FACTOR}, + ) + + ensemble.gen_specs = GenSpecs( + gen_f=uniform_sample_sharing_pbs, + inputs=[], + persis_in=["sim_id"], + outputs=[("x", float, (1,)), ("num_gpus", float), ("num_procs", int)], + initial_batch_size=nsim_workers, + async_return=False, + user={"lb": np.array([50000]), "ub": np.array([100000])}, + ) + + ensemble.alloc_specs = AllocSpecs(alloc_f=alloc_f, user={"async_return": False}) + + sim_max = 12 + ensemble.run(sim_max=sim_max) + + if ensemble.is_manager: + assert ensemble.flag == 0, f"Ensemble failed flag={ensemble.flag}" + H = ensemble.H + assert len(H) == sim_max, f"H len {len(H)} != {sim_max}" + assert not np.any(np.isnan(H["energy"])), "Energy NaN - GPU assignment failed" + + gpus_per_rset = 1 # sf>1 => 1 share per rset + for row in H: + req_gpus = float(row["num_gpus"]) + expected_shares = math.ceil(req_gpus * GPU_SHARING_FACTOR / gpus_per_rset) + assert 1 <= expected_shares <= 4, f"shares {expected_shares} for req {req_gpus}" + assert req_gpus in (0.5, 1.0, 1.5, 2.0), f"req_gpus {req_gpus} not in cycle" + + rm = getattr(getattr(ensemble, "resources", None), "resource_manager", None) + if rm is not None: + assert rm.rsets_free == rm.total_num_rsets, f"Rsets not freed {rm.rsets_free}/{rm.total_num_rsets}" + assert rm.gpu_rsets_free == rm.total_num_gpu_rsets, "GPU rsets leak" + assert rm.total_num_gpu_rsets >= 4, f"Need >=4 GPU rsets for max 2.0 GPU*sf, got {rm.total_num_gpu_rsets}" + assert rm.group_sizes, "group_sizes empty" + assert rm.num_groups >= 1, "num_groups broken" + assert rm.gpu_rsets_per_node >= 2, f"gpu_rsets_per_node {rm.gpu_rsets_per_node} too small for sharing" + print( + f"PBS sharing check: total_gpu_rsets={rm.total_num_gpu_rsets} " + f"gpu_rsets_per_node={rm.gpu_rsets_per_node} " + f"rsets_free={rm.rsets_free}/{rm.total_num_rsets} " + f"ngroups={rm.num_groups} even={rm.even_groups}" + ) + + stats_path = Path("libE_stats.txt") + if stats_path.exists(): + txt = stats_path.read_text() + assert "resource_sets" in txt or "sim_id" in txt, "stats missing" + + print("PBS gpu_sharing_factor assignment/splitting assertions passed") diff --git a/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/submit_pbs.sh b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/submit_pbs.sh new file mode 100755 index 000000000..a235a2900 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_pbs_cuda/submit_pbs.sh @@ -0,0 +1,13 @@ +#!/bin/bash -l +#PBS -l select=1:system=polaris +#PBS -l walltime=00:15:00 +#PBS -l filesystems=home:grand +#PBS -q debug +#PBS -A +#PBS -N libE_forces_pbs_cuda + +export MPICH_GPU_SUPPORT_ENABLED=1 +cd $PBS_O_WORKDIR + +mpirun -np 5 python run_libe_forces.py +# python run_libe_forces.py --comms local --nworkers 5 diff --git a/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/cleanup.sh b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/cleanup.sh new file mode 100644 index 000000000..5ab7dc41e --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/cleanup.sh @@ -0,0 +1 @@ +rm -r ensemble *.npy *.pickle ensemble.log lib*.txt 2>/dev/null; true diff --git a/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/forces_simf.py b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/forces_simf.py new file mode 100644 index 000000000..f01df2332 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/forces_simf.py @@ -0,0 +1,72 @@ +import math + +import numpy as np + +from libensemble.message_numbers import TASK_FAILED, WORKER_DONE +from libensemble.resources.resources import Resources +from libensemble.tools.test_support import check_gpu_setting + + +def run_forces(H, persis_info, sim_specs, libE_info): + """Slurm+CUDA with gpu_sharing_factor: validate spread splitting on real HW.""" + + calc_status = 0 + particles = str(int(H["x"][0][0])) + args = particles + " " + str(10) + " " + particles + + exctr = libE_info["executor"] + task = exctr.submit( + app_name="forces", + app_args=args, + auto_assign_gpus=True, + match_procs_to_gpus=True, + ) + task.wait() + check_gpu_setting(task, assert_setting=False, print_setting=True) + + try: + wres = Resources.resources.worker_resources + sf = sim_specs["user"].get("gpu_sharing_factor", 1) + req_gpus = float(H["num_gpus"][0]) if "num_gpus" in H.dtype.names else 1.0 + gpus_per_rset = getattr(wres, "gpus_per_rset", 1) or 1 + expected_shares = math.ceil(req_gpus * sf / gpus_per_rset) if gpus_per_rset else math.ceil(req_gpus * sf) + assert 1 <= wres.num_rsets <= 4, f"num_rsets {wres.num_rsets} out of range for {req_gpus}" + # Allow off-by-one due to packing/splitting nuances + # mpi_runner may inflate fractional gpus to even across nodes (0.5->1.0,1.5->2.0) + assert ( + wres.num_rsets >= expected_shares + ), f"num_rsets {wres.num_rsets} < expected {expected_shares} for {req_gpus}" + assert wres.num_rsets <= max(4, expected_shares * 2), f"num_rsets {wres.num_rsets} too large for {req_gpus}" + for gid in wres.gpu_ids_on_node: + assert gid >= 0, f"negative gpu_id {gid}" + gpu_shares_per_node = getattr(wres, "gpu_rsets_per_node", 0) or 0 + if gpu_shares_per_node and expected_shares <= gpu_shares_per_node: + assert wres.local_node_count >= 1, f"local_node_count {wres.local_node_count} must be >=1" + elif gpu_shares_per_node: + expected_nodes = math.ceil(expected_shares / gpu_shares_per_node) + assert ( + 1 <= wres.local_node_count <= expected_nodes + 1 + ), f"nodes {wres.local_node_count} vs expected {expected_nodes}" + assert wres.matching_slots, f"Unmatching slots {wres.slots}" + if expected_shares == 3: + assert wres.slots_on_node is not None + + except AssertionError as e: + print(f"Worker {libE_info.get('workerID')} assignment warning: {e}", flush=True) + # Do not fail task for scheduler inflation; just log + except Exception as e: + print(f"Worker validation skipped: {e}", flush=True) + + statfile = "forces.stat" + try: + data = np.loadtxt(statfile) + final_energy = data[-1] + if calc_status != TASK_FAILED: + calc_status = WORKER_DONE + except Exception: + final_energy = np.nan + calc_status = TASK_FAILED + + output = np.zeros(1, dtype=sim_specs["out"]) + output["energy"] = final_energy + return output, persis_info, calc_status diff --git a/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/readme.md b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/readme.md new file mode 100644 index 000000000..2d969f28b --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/readme.md @@ -0,0 +1,26 @@ +## Slurm + CUDA (gpu_sharing_factor, spread) + +Variable-GPU forces example for Slurm systems (Perlmutter) on **real NVIDIA GPUs** - no mocking. + +`gpu_sharing_factor=2` (4 GPUs/node → 8 shares/node). Generator cycles `0.5/1.0/1.5/2.0` → 1-4 shares with `spread` packing; tests cross-node splitting. `SLURM_NODELIST` and NVIDIA GPUs auto-detected. + +### Build + +In `../forces_app`: + +```bash +# Perlmutter (Cray EX, A100) +cc -DGPU -O3 -fopenmp -mp=gpu -target-accel=nvidia80 -o forces.x forces.c +# Generic NVIDIA +mpicc -DGPU -O3 -fopenmp -mp=gpu -o forces.x forces.c +``` + +Requires `forces.x` with `-DGPU` and real NVIDIA GPUs. + +### Run + +```bash +sbatch submit_slurm.sh # --nodes 1-2, srun -n N python run_libe_forces.py +``` + +Manager asserts `shares==ceil(req*sf)`∈1-4, cycle coverage, `total_gpu_rsets>0`, `gpu_rsets_per_node`, `even_groups`/`ngroups`; worker asserts `num_rsets`, `local_node_count` splitting, `matching_slots`. diff --git a/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/run_libe_forces.py b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/run_libe_forces.py new file mode 100644 index 000000000..2b5b284d7 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/run_libe_forces.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python +""" +Slurm + CUDA forces example testing ``gpu_sharing_factor`` with splitting. + +Variable ``num_gpus`` 0.5-2.0 -> 1-4 shares with sf=2 exercises ``spread`` +packing and cross-node splitting. Slurm nodelist and NVIDIA GPUs are +auto-detected from ``SLURM_NODELIST`` / platform detection - no mocking. + +Build ``forces.x`` with CUDA for target (e.g. Perlmutter):: + + cd ../forces_app && ./build_forces.sh # -DGPU line + +Run on Slurm allocation:: + + sbatch submit_slurm.sh +""" + +# Do not change these lines - they are parsed by run-tests.sh +# TESTSUITE_COMMS: mpi local +# TESTSUITE_NPROCS: 5 + +import math +import sys +from pathlib import Path + +import numpy as np +from forces_simf import run_forces + +from libensemble import Ensemble +from libensemble.alloc_funcs.start_only_persistent import only_persistent_gens as alloc_f +from libensemble.executors import MPIExecutor +from libensemble.message_numbers import EVAL_GEN_TAG, FINISHED_PERSISTENT_GEN_TAG, PERSIS_STOP, STOP_TAG +from libensemble.specs import AllocSpecs, GenSpecs, LibeSpecs, SimSpecs +from libensemble.tools.persistent_support import PersistentSupport + +GPU_SHARING_FACTOR = 2 + + +def uniform_sample_sharing_slurm(_, persis_info, gen_specs, libE_info): + """Cycle 0.5,1.0,1.5,2.0 -> shares 1-4, tests splitting across nodes.""" + ub = gen_specs["user"]["ub"] + lb = gen_specs["user"]["lb"] + n = len(lb) + b = gen_specs["initial_batch_size"] + from libensemble.tools import get_rng + + rng = get_rng(gen_specs, libE_info) + ps = PersistentSupport(libE_info, EVAL_GEN_TAG) + tag = None + cycle = [0.5, 1.0, 1.5, 2.0] + idx = 0 + while tag not in [STOP_TAG, PERSIS_STOP]: + H_o = np.zeros(b, dtype=gen_specs["out"]) + H_o["x"] = rng.uniform(lb, ub, (b, n)) + H_o["num_gpus"] = [cycle[(idx + i) % len(cycle)] for i in range(b)] + H_o["num_procs"] = np.ones(b, dtype=int) + idx += b + tag, Work, calc_in = ps.send_recv(H_o) + if hasattr(calc_in, "__len__"): + b = len(calc_in) + return None, persis_info, FINISHED_PERSISTENT_GEN_TAG + + +if __name__ == "__main__": + exctr = MPIExecutor() + sim_app = Path.cwd() / "../forces_app/forces.x" + if not sys.platform.startswith("win") and not Path(sim_app).is_file(): + sys.exit("forces.x not found - please build first in ../forces_app dir") + exctr.register_app(full_path=sim_app, app_name="forces") + + # Real Slurm detection: SLURM_NODELIST + NVIDIA GPUs auto-detected. No mock env. + ensemble = Ensemble(parse_args=True, executor=exctr) + nsim_workers = ensemble.nworkers - 1 + + num_rsets = max(8, nsim_workers * GPU_SHARING_FACTOR) + ensemble.libE_specs = LibeSpecs( + num_resource_sets=num_rsets, + gpu_sharing_factor=GPU_SHARING_FACTOR, + scheduler_opts={"gpu_packing": "spread"}, + sim_dirs_make=True, + stats_fmt={"show_resource_sets": True}, + ) + + ensemble.sim_specs = SimSpecs( + sim_f=run_forces, + inputs=["x"], + outputs=[("energy", float)], + user={"gpu_sharing_factor": GPU_SHARING_FACTOR}, + ) + + ensemble.gen_specs = GenSpecs( + gen_f=uniform_sample_sharing_slurm, + inputs=[], + persis_in=["sim_id"], + outputs=[("x", float, (1,)), ("num_gpus", float), ("num_procs", int)], + initial_batch_size=nsim_workers, + async_return=False, + user={"lb": np.array([50000]), "ub": np.array([100000])}, + ) + + ensemble.alloc_specs = AllocSpecs(alloc_f=alloc_f, user={"async_return": False}) + + sim_max = 12 + ensemble.run(sim_max=sim_max) + + if ensemble.is_manager: + assert ensemble.flag == 0, f"Ensemble failed flag={ensemble.flag}" + H = ensemble.H + assert len(H) == sim_max, f"H len {len(H)} != {sim_max}" + assert not np.any(np.isnan(H["energy"])), "NaN energy - GPU assignment failed" + + gpus_per_rset = 1 + for row in H: + req = float(row["num_gpus"]) + shares = math.ceil(req * GPU_SHARING_FACTOR / gpus_per_rset) + assert shares in (1, 2, 3, 4), f"shares {shares} for req {req}" + assert req in (0.5, 1.0, 1.5, 2.0) + + uniq = np.unique(H["num_gpus"]) + for val in [0.5, 1.0, 1.5, 2.0]: + assert val in uniq, f"Cycle {val} missing - splitting not exercised" + + rm = getattr(getattr(ensemble, "resources", None), "resource_manager", None) + if rm is not None: + assert rm.rsets_free == rm.total_num_rsets, f"Rsets not freed {rm.rsets_free}/{rm.total_num_rsets}" + assert rm.gpu_rsets_free == rm.total_num_gpu_rsets + assert rm.total_num_gpu_rsets >= 4, f"Need >=4 GPU rsets, got {rm.total_num_gpu_rsets}" + assert rm.gpu_rsets_per_node >= 2 + assert rm.num_groups >= 1 + print( + f"Slurm sharing check: total_gpu_rsets={rm.total_num_gpu_rsets} " + f"gpu_rsets_per_node={rm.gpu_rsets_per_node} " + f"even_groups={rm.even_groups} ngroups={rm.num_groups}" + ) + + stats_path = Path("libE_stats.txt") + if stats_path.exists(): + txt = stats_path.read_text() + assert "resource_sets" in txt or "sim_id" in txt + + print("Slurm gpu_sharing_factor splitting (spread) assertions passed") diff --git a/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/submit_slurm.sh b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/submit_slurm.sh new file mode 100755 index 000000000..3254202d2 --- /dev/null +++ b/libensemble/tests/scaling_tests/forces/forces_slurm_cuda/submit_slurm.sh @@ -0,0 +1,12 @@ +#!/bin/bash +#SBATCH -J libE_forces_slurm_cuda +#SBATCH -A +#SBATCH -C gpu +#SBATCH --time 00:15:00 +#SBATCH --nodes 1 + +export MPICH_GPU_SUPPORT_ENABLED=1 +export SLURM_EXACT=1 + +srun -n 5 python run_libe_forces.py +# python run_libe_forces.py --comms local --nworkers 5 diff --git a/libensemble/tests/unit_tests/test_ensemble.py b/libensemble/tests/unit_tests/test_ensemble.py index 767d80102..ad48e27f9 100644 --- a/libensemble/tests/unit_tests/test_ensemble.py +++ b/libensemble/tests/unit_tests/test_ensemble.py @@ -65,6 +65,20 @@ def test_ensemble_init(): assert e.is_manager, "parse_args() didn't populate defaults for class's libE_specs" +def test_parse_args_nresource_sets(monkeypatch): + """--nresource_sets sets num_resource_sets; otherwise it is left unset.""" + from libensemble.tools import parse_args + + monkeypatch.setattr(sys, "argv", ["", "--nworkers", "4"]) + nworkers, _, libE_specs, _ = parse_args() + assert nworkers == 4 and libE_specs["comms"] == "local" + assert "num_resource_sets" not in libE_specs + + monkeypatch.setattr(sys, "argv", ["", "--nworkers", "4", "--nresource_sets", "8"]) + nworkers, _, libE_specs, _ = parse_args() + assert nworkers == 4 and libE_specs["num_resource_sets"] == 8 + + def test_ensemble_parse_args_false(): from libensemble.ensemble import Ensemble from libensemble.specs import LibeSpecs @@ -574,6 +588,7 @@ def test_h0_chaining_plain_run(): if __name__ == "__main__": test_ensemble_init() + test_parse_args_nresource_sets(__import__("pytest").MonkeyPatch()) test_ensemble_parse_args_false() test_full_workflow() test_flakey_workflow() diff --git a/libensemble/tests/unit_tests/test_models.py b/libensemble/tests/unit_tests/test_models.py index 3dd3ba3cd..fcb6c4655 100644 --- a/libensemble/tests/unit_tests/test_models.py +++ b/libensemble/tests/unit_tests/test_models.py @@ -1,4 +1,5 @@ import numpy as np +import pytest from gest_api.vocs import VOCS from pydantic import ValidationError @@ -96,6 +97,13 @@ def test_libe_specs_invalid(): assert flag, "LibeSpecs didn't raise ValidationError on invalid specs" +def test_gpu_sharing_factor_validation(): + assert LibeSpecs().gpu_sharing_factor == 1 + assert LibeSpecs(gpu_sharing_factor=2).gpu_sharing_factor == 2 + with pytest.raises(ValueError, match="gpu_sharing_factor must be at least 1"): + LibeSpecs(gpu_sharing_factor=0) + + def test_ensemble_specs(): sim_specs, gen_specs, exit_criteria = setup.make_criteria_and_specs_0() libE_specs = {"comms": "local", "nworkers": 4} diff --git a/libensemble/tests/unit_tests/test_resource_scheduler.py b/libensemble/tests/unit_tests/test_resource_scheduler.py index 762f07ead..4e6516b58 100644 --- a/libensemble/tests/unit_tests/test_resource_scheduler.py +++ b/libensemble/tests/unit_tests/test_resource_scheduler.py @@ -1,10 +1,13 @@ +import os import sys import time import numpy as np import pytest +from libensemble.resources.resources import Resources from libensemble.resources.scheduler import InsufficientFreeResources, InsufficientResourcesError, ResourceScheduler +from libensemble.resources.worker_resources import WorkerResources class MyResources: @@ -24,6 +27,7 @@ def __init__(self, num_rsets, num_groups, gpus_per_node=0): self.rsets_per_node = self.total_num_rsets // num_groups self.gpu_rsets_per_node = min(gpus_per_node, self.rsets_per_node) self.nongpu_rsets_per_node = self.rsets_per_node - self.gpu_rsets_per_node + self.gpu_sharing_factor = 1 self.even_groups = True self.rsets = np.zeros(self.total_num_rsets, dtype=MyResources.rset_dtype) @@ -661,6 +665,238 @@ def test_schedule_find_gaps_2nodes_withgpus(): del resources +def _setup_resources_sharing(gpu_sharing_factor=2, gpus_per_group=None, sched_opts=None, num_resource_sets=16): + """Create a mocked 2-node, 4-GPU/node resource set for helper tests. + + Mocks SLURM nodelist ``knl-[0020,0036]`` (2 nodes) with + ``gpus_on_node=4`` and ``cores_on_node=(64, 64)`` via + ``ResourceInfo``. With ``gpu_sharing_factor=2`` each physical GPU is + split into 2 logical shares, yielding ``8`` GPU shares per node and + ``16`` total for the default ``num_resource_sets=16`` (``8`` per node). + Returns the ``ResourceManager`` and a ``ResourceScheduler`` bound to it. + + Args: + gpu_sharing_factor: Logical shares per physical GPU (default 2). + gpus_per_group: Optional cap on GPUs per scheduler group, scaled by + ``gpu_sharing_factor`` in ``RSetResources.get_group_list``. + sched_opts: Dict of scheduler options, notably ``gpu_packing`` + ``\"pack\"`` (co-locate fractional shares) or ``\"spread\"``. + num_resource_sets: Total rsets to create; ``rsets_per_node`` is + derived as ``ceil(num_resource_sets/num_nodes)``. + + Returns: + Tuple of (resource_manager, scheduler) ready for direct calls to + ``_build_gpu_bins`` etc. Caller is responsible for cleaning the + mocked env var if needed. + """ + # Mock a 2-node SLURM nodelist; no real hardware needed + os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020,0036]" + resource_info = { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 4, + "cores_on_node": (64, 64), + } + libE_specs = { + "resource_info": resource_info, + "dedicated_mode": False, # keep libEnsemble ranks out of the nodelist + "gpu_sharing_factor": gpu_sharing_factor, + "gpus_per_group": gpus_per_group, + "num_resource_sets": num_resource_sets, + } + # Instantiate Resources and its ResourceManager (holds RSetResources) + res = Resources(libE_specs=libE_specs, top_level_dir=os.getcwd()) + res.set_resource_manager(num_workers=8) + # Bind scheduler to the mocked manager + sched = ResourceScheduler(user_resources=res.resource_manager, sched_opts=sched_opts or {}) + return res.resource_manager, sched + + +def _gpu_ids(rm, team): + return [int(rm.rsets["gpu_id"][r]) for r in team] + + +def _slots(rm, team): + return [int(rm.rsets["slot"][r]) for r in team] + + +def test_gpu_bins(): + """Bins map each GPU index to its free slots, intersected across groups.""" + _, sched = _setup_resources_sharing() + valid = sched.filter_for_rset_type(sched.get_avail_rsets_by_group(), use_gpus=True) + bins, slot_maps = sched._gpu_bins([1], valid) + assert bins == {0: [0, 1], 1: [2, 3], 2: [4, 5], 3: [6, 7]} + assert slot_maps[0][7] == 7 + + # Only slots free in both groups are offered + valid[1].remove(0) + valid[2].remove(13) # group 2, slot 5 + bins, slot_maps = sched._gpu_bins([1, 2], valid) + assert bins == {0: [1], 1: [2, 3], 2: [4], 3: [6, 7]} + assert slot_maps[1][1] == 9 + + +def test_pick_from_gpu_bins(): + """Whole GPUs must be fully free and the remainder must fit on one GPU.""" + rm, _ = _setup_resources_sharing(gpu_sharing_factor=4, num_resource_sets=32) + pack = ResourceScheduler(user_resources=rm, sched_opts={"gpu_packing": "pack"}) + spread = ResourceScheduler(user_resources=rm, sched_opts={"gpu_packing": "spread"}) + + bins = {0: [0, 1, 2, 3], 1: [5, 6, 7], 2: [9, 10], 3: [12, 13, 14, 15]} + assert pack._pick_from_gpu_bins(bins, 4) == [0, 1, 2, 3] # one whole GPU + assert pack._pick_from_gpu_bins(bins, 8) == [0, 1, 2, 3, 12, 13, 14, 15] + assert pack._pick_from_gpu_bins(bins, 12) is None # only two fully free GPUs + assert pack._pick_from_gpu_bins(bins, 2) == [9, 10] # pack: fullest GPU that fits + assert spread._pick_from_gpu_bins(bins, 2) == [0, 1] # spread: emptiest GPU + assert pack._pick_from_gpu_bins(bins, 7) == [0, 1, 2, 3, 5, 6, 7] # whole + remainder + + # Remainder is never split across GPUs, even if enough shares are free in total + fragmented = {0: [2, 3], 1: [7], 2: [10]} + assert pack._pick_from_gpu_bins(fragmented, 3) is None + assert pack._pick_from_gpu_bins(fragmented, 2) == [2, 3] + + +def test_assign_gpu_sharing_resources(): + """End-to-end allocation with packing policies, whole GPUs, and errors.""" + gres, sched = _setup_resources_sharing(sched_opts={"gpu_packing": "pack"}) + assert sched.assign_resources(rsets_req=1, use_gpus=True) == [0] + assert sched.assign_resources(rsets_req=1, use_gpus=True) == [1] # pack onto same GPU + assert sched.gpu_rsets_free == 14 + + _, sched_spread = _setup_resources_sharing(sched_opts={"gpu_packing": "spread"}) + t1 = sched_spread.assign_resources(rsets_req=1, use_gpus=True) + t2 = sched_spread.assign_resources(rsets_req=1, use_gpus=True) + assert sched_spread.resources.rsets["group"][t1[0]] != sched_spread.resources.rsets["group"][t2[0]] + + rm, sched_exc = _setup_resources_sharing() + team = sched_exc.assign_resources(rsets_req=4, use_gpus=True) + assert len(set(_gpu_ids(rm, team))) == 2 # two whole GPUs + team = sched_exc.assign_resources(rsets_req=3, use_gpus=True) + assert len(set(_gpu_ids(rm, team))) == 2 # one whole GPU + one share + + with pytest.raises(InsufficientResourcesError): + sched_exc.assign_resources(rsets_req=20, use_gpus=True) + with pytest.raises(ValueError): + ResourceScheduler(user_resources=gres, sched_opts={"gpu_packing": "bad"}) + + # gpus_per_group is scaled by the sharing factor (2 GPUs -> 4 shares per group) + rm, sched_grp = _setup_resources_sharing(gpus_per_group=2) + team = sched_grp.assign_resources(rsets_req=8, use_gpus=True) + assert len(team) == 8 + assert len(set(rm.rsets["group"][team])) == 2 + + +def test_gpu_sharing_remainder_on_single_gpu(): + """A fractional remainder must not be spread over several physical GPUs. + + One node with 2 GPUs and sf=4. After 2 shares on GPU 0 and 3 shares on GPU 1, + 5 shares are free in total, but no single GPU has 3 free shares. + """ + os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020]" + libE_specs = { + "resource_info": { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 2, + "cores_on_node": (64, 64), + }, + "dedicated_mode": False, + "gpu_sharing_factor": 4, + "num_resource_sets": 8, + } + res = Resources(libE_specs=libE_specs, top_level_dir=os.getcwd()) + res.set_resource_manager(num_workers=8) + rm = res.resource_manager + sched = ResourceScheduler(user_resources=rm) + assert _gpu_ids(rm, sched.assign_resources(2, use_gpus=True, user_params=[2])) == [0, 0] + assert _gpu_ids(rm, sched.assign_resources(3, use_gpus=True, user_params=[3])) == [1, 1, 1] + _fail_to_resource(sched, 3, use_gpus=True) + assert _gpu_ids(rm, sched.assign_resources(2, use_gpus=True, user_params=[2])) == [0, 0] + os.environ.pop("LIBE_RESOURCES_TEST_NODE_LIST", None) + + +def test_gpu_sharing_matching_slots_across_nodes(): + """Multi-node teams use the same slots, and hence the same GPU IDs, on each node.""" + rm, sched = _setup_resources_sharing() + first = sched.assign_resources(1, use_gpus=True, user_params=[1]) + assert first == [0] + team = sched.assign_resources(12, use_gpus=True, user_params=[12]) + assert len(team) == 12 + groups = rm.rsets["group"][team] + node1 = [r for r, g in zip(team, groups) if g == 1] + node2 = [r for r, g in zip(team, groups) if g == 2] + assert _slots(rm, node1) == _slots(rm, node2) + assert _gpu_ids(rm, node1) == _gpu_ids(rm, node2) + assert 0 not in team + + res = Resources( + libE_specs={ + "resource_info": { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 4, + "cores_on_node": (64, 64), + }, + "dedicated_mode": False, + "gpu_sharing_factor": 2, + "num_resource_sets": 16, + }, + top_level_dir=os.getcwd(), + ) + wres = WorkerResources(8, res.glob_resources, 1) + wres.set_rset_team(team) + assert wres.matching_slots + assert wres.get_gpu_ids_as_string() == "1,2,3" + + +def test_gpu_sharing_no_match_slots(): + """With match_slots=False each group is filled independently.""" + rm, sched = _setup_resources_sharing(sched_opts={"match_slots": False}) + sched.assign_resources(1, use_gpus=True, user_params=[1]) + team = sched.assign_resources(16 - 2, use_gpus=True, user_params=[14]) + assert len(team) == 14 + assert len(set(rm.rsets["group"][team])) == 2 + + +def test_gpu_sharing_split2fit(): + """Requests that do not fit on one node are split over more nodes if allowed.""" + os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020,0036,0040]" + libE_specs = { + "resource_info": { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 4, + "cores_on_node": (64, 64), + }, + "dedicated_mode": False, + "gpu_sharing_factor": 2, + "num_resource_sets": 24, + } + res = Resources(libE_specs=libE_specs, top_level_dir=os.getcwd()) + res.set_resource_manager(num_workers=24) + rm = res.resource_manager + + # GPU 0 on each node (8 rsets per node) is busy, so no node has 8 free shares + rm.assign_rsets([0, 1, 8, 9, 16, 17], 1) + + # 8 shares are split as 4 + 4 on matching slots + team = ResourceScheduler(user_resources=rm).assign_resources(8, use_gpus=True, user_params=[8]) + groups = sorted(set(rm.rsets["group"][team])) + assert len(groups) == 2 + by_group = [[r for r in team if rm.rsets["group"][r] == g] for g in groups] + assert _slots(rm, by_group[0]) == _slots(rm, by_group[1]) + + # Without split2fit the request fails + _fail_to_resource(ResourceScheduler(user_resources=rm, sched_opts={"split2fit": False}), 8, use_gpus=True) + + # 7 shares are extended to 8 (4 + 4), but user_params of 7 cannot be split evenly + with pytest.raises(InsufficientFreeResources): + ResourceScheduler(user_resources=rm).assign_resources(7, use_gpus=True, user_params=[7]) + + # With GPU 1 also busy on nodes 1 and 3, only node 2 has 6 free shares, + # so 12 shares split as 4 + 4 + 4 over three nodes + rm.assign_rsets([2, 3, 18, 19], 1) + team = ResourceScheduler(user_resources=rm).assign_resources(12, use_gpus=True, user_params=[12]) + assert len(set(rm.rsets["group"][team])) == 3 + os.environ.pop("LIBE_RESOURCES_TEST_NODE_LIST", None) + + if __name__ == "__main__": test_request_zero_rsets() test_too_many_rsets() @@ -680,3 +916,10 @@ def test_schedule_find_gaps_2nodes_withgpus(): test_try1node_findon_2_or_4nodes() test_large_match_slots() test_schedule_find_gaps_2nodes_withgpus() + test_gpu_bins() + test_pick_from_gpu_bins() + test_assign_gpu_sharing_resources() + test_gpu_sharing_remainder_on_single_gpu() + test_gpu_sharing_matching_slots_across_nodes() + test_gpu_sharing_no_match_slots() + test_gpu_sharing_split2fit() diff --git a/libensemble/tests/unit_tests/test_resources.py b/libensemble/tests/unit_tests/test_resources.py index b79a5eeff..31beab8f5 100644 --- a/libensemble/tests/unit_tests/test_resources.py +++ b/libensemble/tests/unit_tests/test_resources.py @@ -1,9 +1,15 @@ import os import socket +import numpy as np +import pytest + +from libensemble.executors.mpi_runner import MPICH_MPIRunner from libensemble.resources.env_resources import EnvResources from libensemble.resources.mpi_resources import create_machinefile from libensemble.resources.resources import GlobalResources, Resources, ResourcesException +from libensemble.resources.rset_resources import RSetResourcesException +from libensemble.resources.scheduler import ResourceScheduler from libensemble.resources.worker_resources import ResourceManager, WorkerResources @@ -780,12 +786,22 @@ def test_machinefile_from_resources(): os.remove("machinefile") -def _setup_wresources_gpus(ngpus): +def _setup_wresources_gpus(ngpus, gpu_sharing_factor=1, num_resource_sets=None, gpus_per_group=None): """Set up worker resources with a given number of GPUs""" os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020,0036]" resource_info = {"nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", "gpus_on_node": ngpus} - libE_specs = {"resource_info": resource_info, "dedicated_mode": False} + if gpu_sharing_factor > 1: + # Sharing needs at least one core per rset, so don't depend on the host's core count + resource_info["cores_on_node"] = (64, 64) + libE_specs = { + "resource_info": resource_info, + "dedicated_mode": False, + "gpu_sharing_factor": gpu_sharing_factor, + "gpus_per_group": gpus_per_group, + } + if num_resource_sets is not None: + libE_specs["num_resource_sets"] = num_resource_sets gresources = GlobalResources(libE_specs) nworkers = 8 workerID = 2 # Does not matter as using set_rset_team @@ -915,6 +931,40 @@ def test_wresources_set_gpus_by_platform_gpu_tiles(): del wresources +def test_wresources_set_gpus_with_sharing(): + wresources = _setup_wresources_gpus(4, gpu_sharing_factor=2, num_resource_sets=16) + + assert wresources.total_num_gpu_rsets == 16 + assert list(wresources.all_rsets["gpu_id"]) == [0, 0, 1, 1, 2, 2, 3, 3] * 2 + assert list(wresources.all_rsets["gpu_share"]) == [0, 1, 0, 1, 0, 1, 0, 1] * 2 + + wresources.set_rset_team([0, 1, 2, 3]) + wresources.set_env_to_gpus("LIBE_TEST_GPUS_1") + assert os.environ["LIBE_TEST_GPUS_1"] == "0,1" + os.environ.pop("LIBE_TEST_GPUS_1") + + +def test_gpu_sharing_executor_env_mapping(): + wresources = _setup_wresources_gpus(4, gpu_sharing_factor=2, num_resource_sets=16) + wresources.set_rset_team([1]) + + class Task: + def _add_to_env(self, name, value): + self.env = {name: value} + + task = Task() + MPICH_MPIRunner()._set_gpu_env_var(wresources, task, 1, "CUDA_VISIBLE_DEVICES") + assert task.env == {"CUDA_VISIBLE_DEVICES": "0"} + + +def test_gpu_sharing_mapping_with_gpu_groups(): + wresources = _setup_wresources_gpus(4, gpu_sharing_factor=2, num_resource_sets=16, gpus_per_group=2) + + assert list(wresources.all_rsets["slot"]) == [0, 1, 2, 3] * 4 + assert list(wresources.all_rsets["gpu_id"]) == [0, 0, 1, 1, 2, 2, 3, 3] * 2 + assert list(wresources.all_rsets["gpu_share"]) == [0, 1, 0, 1, 0, 1, 0, 1] * 2 + + def test_wresources_set_gpus_x2(): wresources = _setup_wresources_gpus(8) @@ -975,8 +1025,193 @@ def test_wresources_set_no_gpus(): assert "LIBE_TEST_GPUS_1" not in os.environ +def _setup_resources_sharing_for_alloc(gpu_sharing_factor=2, num_resource_sets=16): + """Create mocked Resources for AllocSupport fractional tests. + + Mocks a 2-node SLURM nodelist with 4 GPUs/node and sf=2, yielding + 8 GPU shares per node (16 total). Returns the ``Resources`` object that + AllocSupport will use to convert fractional ``num_gpus`` to shares via + ``ceil(num_gpus*sf/gpus_per_rset)``. + """ + # Mock a 2-node SLURM nodelist; no real hardware needed + os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020,0036]" + resource_info = { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 4, + "cores_on_node": (64, 64), + } + libE_specs = { + "resource_info": resource_info, + "dedicated_mode": False, # keep libEnsemble ranks out of the nodelist + "gpu_sharing_factor": gpu_sharing_factor, + "num_resource_sets": num_resource_sets, + } + # Instantiate Resources and its ResourceManager (holds RSetResources) + res = Resources(libE_specs=libE_specs, top_level_dir=os.getcwd()) + res.set_resource_manager(num_workers=8) + return res + + +def test_alloc_support_fractional(): + """Unit coverage for ``AllocSupport`` fractional ``num_gpus`` handling. + + When ``gpu_sharing_factor>1`` the manager converts a fractional + ``num_gpus`` (physical GPUs) to logical shares via + ``ceil(num_gpus * sf / gpus_per_rset)``. This test drives: + + * ``_req_resources_sim`` with History rows ``0.5→1 share``, ``1.5→3``, + ``2.0→4`` and asserts ``use_gpus`` and ``libE_info[\"num_gpus\"]``. + * ``_req_resources_gen`` with ``persis_info[\"gen_num_gpus\"]`` ``0.5`` + and ``1.5`` (the ``def_gen_num_gpus`` path) likewise mapping to 1 and + 3 shares. + + Uses a lightweight ``AllocSupport`` instantiation with a mocked + ``Resources`` object to avoid a full ensemble run. + """ + # Use a mocked 16-rset Resources with sf=2 (gpus_per_rset=1) + res = _setup_resources_sharing_for_alloc() + from libensemble.tools.alloc_support import AllocSupport + + # Minimal worker array needed by AllocSupport + W = np.zeros(3, dtype=[("persis_state", int), ("active", int)]) + libE_info = {} + user_params: list = [] + # History with fractional num_gpus: 0.5 and 1.5 physical GPUs + H = np.zeros(2, dtype=[("num_gpus", float), ("num_procs", int)]) + H["num_gpus"] = [0.5, 1.5] + H["num_procs"] = [1, 1] + # Pass the mocked Resources directly (avoid global Resources.resources=None issue) + support = AllocSupport(W, manage_resources=True, libE_info=libE_info, user_resources=res) + # Override scheduler to enforce pack policy for deterministic shares + support.sched = ResourceScheduler(user_resources=res.resource_manager, sched_opts={"gpu_packing": "pack"}) + # 0.5 GPUs * sf2 / gpus_per_rset1 => ceil(1) = 1 share + nsets, use_gpus = support._req_resources_sim(libE_info, user_params, H, [0]) + assert nsets == 1 and use_gpus is True and libE_info["num_gpus"] == 0.5 + user_params = [] + libE_info = {} + # 1.5 GPUs => ceil(3) = 3 shares (1 whole GPU +1 share) + nsets2, _ = support._req_resources_sim(libE_info, user_params, H, [1]) + assert nsets2 == 3 + user_params = [] + libE_info = {} + # 2.0 GPUs => ceil(4) = 4 shares (2 whole GPUs) + H2 = np.zeros(1, dtype=[("num_gpus", float)]) + H2["num_gpus"] = [2.0] + nsets3, _ = support._req_resources_sim(libE_info, user_params, H2, [0]) + assert nsets3 == 4 + user_params = [] + libE_info = {} + # Generator path via persis_info gen_num_gpus (0.5 => 1 share) + support2 = AllocSupport( + W, manage_resources=True, persis_info={"gen_num_gpus": 0.5}, libE_info={}, user_resources=support.resources + ) + support2.resources = support.resources + support2.sched = support.sched + nsets4, _ = support2._req_resources_gen(libE_info, user_params) + assert nsets4 == 1 + user_params = [] + libE_info = {} + # Generator with 1.5 GPUs => 3 shares + support2.persis_info = {"gen_num_gpus": 1.5} + nsets5, _ = support2._req_resources_gen(libE_info, user_params) + assert nsets5 == 3 + # Whole-number requests are passed on as ints + assert libE_info["num_gpus"] == 1.5 + libE_info = {} + support2.persis_info = {"gen_num_gpus": 2.0} + support2._req_resources_gen(libE_info, user_params) + assert type(libE_info["num_gpus"]) is int and libE_info["num_gpus"] == 2 + + +def _sharing_resource_manager(gpu_sharing_factor, num_resource_sets=None, num_workers=4, cores=64): + os.environ["LIBE_RESOURCES_TEST_NODE_LIST"] = "knl-[0020,0036]" + libE_specs = { + "resource_info": { + "nodelist_env_slurm": "LIBE_RESOURCES_TEST_NODE_LIST", + "gpus_on_node": 4, + "cores_on_node": (cores, cores), + }, + "dedicated_mode": False, + "gpu_sharing_factor": gpu_sharing_factor, + "num_resource_sets": num_resource_sets, + } + res = Resources(libE_specs=libE_specs, top_level_dir=os.getcwd()) + res.set_resource_manager(num_workers=num_workers) + return res.resource_manager + + +def test_gpu_sharing_default_num_resource_sets(): + """Without num_resource_sets, sharing gives one rset per GPU share, independent of workers.""" + rm = _sharing_resource_manager(gpu_sharing_factor=2) + assert rm.total_num_rsets == 16 # 2 nodes x 4 GPUs x 2 shares + assert list(rm.rsets["gpu_id"][:8]) == [0, 0, 1, 1, 2, 2, 3, 3] + assert rm.procs_per_rset == 8 + + rm = _sharing_resource_manager(gpu_sharing_factor=1) + assert rm.total_num_rsets == 4 # unchanged: one per worker + + # Explicit values that cover all shares are respected (extra rsets have no GPUs) + rm = _sharing_resource_manager(gpu_sharing_factor=2, num_resource_sets=20) + assert rm.total_num_rsets == 20 + assert rm.total_num_gpu_rsets == 16 + + +def test_gpu_sharing_rset_errors(): + """Explicit num_resource_sets too small for all shares, or zero cores per rset, raises.""" + with pytest.raises(RSetResourcesException, match="at least 16"): + _sharing_resource_manager(gpu_sharing_factor=2, num_resource_sets=8) + with pytest.raises(RSetResourcesException, match="CPU cores"): + _sharing_resource_manager(gpu_sharing_factor=8, cores=16) + + +def test_gpu_sharing_manager_warnings(monkeypatch): + """The manager warns if sharing is set but tasks don't request GPUs, or no GPUs exist.""" + from types import SimpleNamespace + + import libensemble.manager as manager_mod + + warnings = [] + monkeypatch.setattr(manager_mod.logger, "manager_warning", lambda msg: warnings.append(msg)) + + def check(sf, gpus, fields, gen_num_gpus=0): + warnings.clear() + H = np.zeros(1, dtype=[(f, float) for f in fields]) + mgr = SimpleNamespace(hist=SimpleNamespace(H=H), gen_num_gpus=gen_num_gpus) + gres = SimpleNamespace(gpu_sharing_factor=sf, gpus_avail_per_node=gpus) + manager_mod.Manager._check_gpu_sharing(mgr, gres) + return list(warnings) + + assert check(1, 4, ["x"]) == [] + assert check(2, 4, ["x", "num_gpus"]) == [] + assert check(2, 4, ["x"], gen_num_gpus=0.5) == [] + assert "should request GPUs" in check(2, 4, ["x", "num_procs"])[0] + assert "no GPUs were detected" in check(2, 0, ["x", "num_gpus"])[0] + + +def test_convert_to_rsets_keeps_ints(): + """Whole-number procs/GPUs from float or numpy History fields stay ints in libE_info.""" + from libensemble.tools.alloc_support import AllocSupport + + H = np.zeros(2, dtype=[("num_procs", int), ("num_gpus", float)]) + H["num_procs"] = [4, 3] + H["num_gpus"] = [2.0, 1.0] + for field, per_rset, exp_rsets, exp_units in [("num_procs", 2, 2, 4), ("num_gpus", 2, 1, 2)]: + libE_info, user_params = {}, [] + nrsets = AllocSupport._convert_rows_to_rsets(libE_info, user_params, H, [0, 1], per_rset, field) + assert nrsets == exp_rsets and type(nrsets) is int + assert libE_info[field] == exp_units and type(libE_info[field]) is int + assert type(user_params[0]) is int + # Fractional values remain floats and round up + libE_info = {} + assert AllocSupport._convert_to_rsets(libE_info, [], 1, 1.5, "num_gpus") == 2 + assert libE_info["num_gpus"] == 1.5 + + if __name__ == "__main__": setup_standalone_run() + test_convert_to_rsets_keeps_ints() + test_gpu_sharing_default_num_resource_sets() + test_gpu_sharing_rset_errors() test_get_global_nodelist_frm_slurm() test_get_global_nodelist_frm_slurm_suffix() @@ -1008,4 +1243,6 @@ def test_wresources_set_no_gpus(): test_wresources_set_limit_gpus() test_wresources_set_no_gpus() + test_alloc_support_fractional() + teardown_standalone_run() diff --git a/libensemble/tools/alloc_support.py b/libensemble/tools/alloc_support.py index 6fda1e6b2..fcc101499 100644 --- a/libensemble/tools/alloc_support.py +++ b/libensemble/tools/alloc_support.py @@ -1,4 +1,5 @@ import logging +import math import numpy as np @@ -145,7 +146,7 @@ def _req_resources_sim(self, libE_info, user_params, H, H_rows): """Determine required resources for a sim work unit""" use_gpus = None if "resource_sets" in H.dtype.names: - num_rsets_req = np.max(H[H_rows]["resource_sets"]) # sim rsets + num_rsets_req = int(np.max(H[H_rows]["resource_sets"])) # sim rsets elif "num_procs" in H.dtype.names and self.resources: procs_per_rset = self.resources.resource_manager.procs_per_rset num_rsets_req = AllocSupport._convert_rows_to_rsets( @@ -153,11 +154,9 @@ def _req_resources_sim(self, libE_info, user_params, H, H_rows): ) else: num_rsets_req = 1 - if "num_gpus" in H.dtype.names: - gpus_per_rset = self.resources.resource_manager.gpus_per_rset - num_rsets_req_for_gpus = AllocSupport._convert_rows_to_rsets( - libE_info, user_params, H, H_rows, gpus_per_rset, "num_gpus" - ) + if "num_gpus" in H.dtype.names and self.resources: + max_gpus = np.max(H[H_rows]["num_gpus"]) + num_rsets_req_for_gpus = self._convert_gpus_to_rsets(libE_info, user_params, max_gpus) if num_rsets_req_for_gpus > 0: use_gpus = True num_rsets_req = max(num_rsets_req, num_rsets_req_for_gpus) @@ -178,15 +177,28 @@ def _req_resources_gen(self, libE_info, user_params): ) gen_ngpus = self.persis_info.get("gen_num_gpus", self.def_gen_num_gpus) if gen_ngpus and self.resources: - gpus_per_rset = self.resources.resource_manager.gpus_per_rset - num_rsets_req_for_gpus = AllocSupport._convert_to_rsets( - libE_info, user_params, gpus_per_rset, gen_ngpus, "num_gpus" - ) + num_rsets_req_for_gpus = self._convert_gpus_to_rsets(libE_info, user_params, gen_ngpus) if num_rsets_req_for_gpus > 0: use_gpus = True num_rsets_req = max(num_rsets_req, num_rsets_req_for_gpus) return num_rsets_req, use_gpus + def _convert_gpus_to_rsets(self, libE_info, user_params, num_gpus): + """Convert a GPU request to resource sets, accounting for ``gpu_sharing_factor``. + + With sharing, each resource set holds one GPU share, so a request for ``num_gpus`` + (possibly fractional) needs ``ceil(num_gpus * gpu_sharing_factor)`` shares. + ``libE_info["num_gpus"]`` always records the requested number of GPUs. + """ + rm = self.resources.resource_manager + num_gpus = AllocSupport._normalize_units(num_gpus) + if rm.gpu_sharing_factor > 1: + shares = math.ceil(num_gpus * rm.gpu_sharing_factor) + num_rsets_req = AllocSupport._convert_to_rsets({}, user_params, rm.gpus_per_rset, shares, "num_gpus") + libE_info["num_gpus"] = num_gpus + return num_rsets_req + return AllocSupport._convert_to_rsets(libE_info, user_params, rm.gpus_per_rset, num_gpus, "num_gpus") + def _update_rset_team(self, libE_info, wid, H=None, H_rows=None): """Add rset_team to libE_info.""" if self.manage_resources and not libE_info.get("rset_team"): @@ -390,20 +402,27 @@ def _check_H_fields(H_fields): # H_fields = list(OrderedDict.fromkeys(H_fields)) # Maintain order return H_fields + @staticmethod + def _normalize_units(value): + """Return value as an int if it is a whole number, else as a float""" + value = float(value) + return int(value) if value.is_integer() else value + @staticmethod def _convert_rows_to_rsets(libE_info, user_params, H, H_rows, units_per_rset, units_str): """Convert num_procs & num_gpus requirements to resource sets for sim functions""" - max_num_units = int(np.max(H[H_rows][units_str])) + max_num_units = np.max(H[H_rows][units_str]) num_rsets_req = AllocSupport._convert_to_rsets(libE_info, user_params, units_per_rset, max_num_units, units_str) return num_rsets_req @staticmethod def _convert_to_rsets(libE_info, user_params, units_per_rset, num_units, units_str): """Convert num_procs & num_gpus requirements to resource sets""" + num_units = AllocSupport._normalize_units(num_units) user_params.append(num_units) if num_units > 0: try: - num_rsets_req = num_units // units_per_rset + (num_units % units_per_rset > 0) + num_rsets_req = int(-(-num_units // units_per_rset)) except ZeroDivisionError: raise InsufficientResourcesError( f"There are zero {units_str} per resource set (worker). Use fewer workers or more resources" diff --git a/libensemble/tools/parse_args.py b/libensemble/tools/parse_args.py index 377120f11..bacc98f69 100644 --- a/libensemble/tools/parse_args.py +++ b/libensemble/tools/parse_args.py @@ -14,12 +14,6 @@ help="Type of communicator", ) parser.add_argument("-n", "--nworkers", type=int, nargs="?", help="Number of local forked processes") -parser.add_argument( - "--nsim_workers", - type=int, - nargs="?", - help="Number of workers for sims. 1+ unresourced workers for a persistent generator will be added", -) parser.add_argument("--nresource_sets", type=int, nargs="?", help="Number of resource sets") parser.add_argument("--workers", type=str, nargs="+", help="List of worker nodes") parser.add_argument("--workerID", type=int, nargs="?", help="Client worker ID") @@ -54,11 +48,6 @@ def _mpi_parse_args(args): if args.nresource_sets is not None: libE_specs["num_resource_sets"] = args.nresource_sets - # Convenience option which sets other libE_specs options. - nsim_workers = args.nsim_workers - if nsim_workers is not None: - libE_specs["num_resource_sets"] = libE_specs.get("num_resource_sets", nsim_workers) - return nworkers, is_manager, libE_specs, args.tester_args @@ -70,12 +59,6 @@ def _local_parse_args(args): if args.nresource_sets is not None: libE_specs["num_resource_sets"] = args.nresource_sets - # Convenience option which sets other libE_specs options. - nsim_workers = args.nsim_workers - if nsim_workers is not None: - nworkers = nworkers or nsim_workers + 1 - libE_specs["num_resource_sets"] = libE_specs.get("num_resource_sets", nsim_workers) - nworkers = nworkers or 4 libE_specs["nworkers"] = nworkers @@ -108,7 +91,6 @@ def parse_args(): usage: test_... [-h] [--comms [{local, threads, mpi}]] [--nworkers [NWORKERS]] [--workers WORKERS [WORKERS ...]] - [--nsim_workers [NSIM_WORKERS]] [--nresource_sets [NRESOURCE_SETS]] [--workerID [WORKERID]] [--server SERVER SERVER SERVER] [--pwd [PWD]] [--worker_pwd [WORKER_PWD]] @@ -120,18 +102,11 @@ def parse_args(): the manager, and the rest are workers. --comms, Communications medium for manager and workers. - Default is 'local' if --nworkers is provided, otherwise 'mpi'. + Default is 'local' if --nworkers is provided, otherwise 'mpi'. --nworkers/-n, (For 'local' or 'threads' comms) Set number of workers. --nresource_sets, Explicitly set the number of resource sets. This sets libE_specs["num_resource_sets"]. By default, resources will be divided by workers. - --nsim_workers, (For 'local" or 'mpi' comms) A convenience option for cases with - persistent generators - sets the number of simulation workers. - If used with no other criteria, one additional worker for running a - generator will be added, and the number of resource sets will be assigned - the given value. If '--nworkers' has also been specified, will generate - enough additional workers to match the other criteria. If '--nresource_sets' - is also specified, will not override resource sets. Example command lines: @@ -140,9 +115,6 @@ def parse_args(): $ python calling_script --nworkers 4 $ python calling_script -n 4 - Run with 'local' comms and 5 workers - one gen worker and 4 sim workers. - $ python calling_script --comms local --nsim_workers 4 - Run with 'local' comms with 4 workers and 8 resource sets. The extra resource sets will be used for larger simulations (using variable resource assignment). $ python calling_script --comms local --nresource_sets 8 diff --git a/libensemble/tools/test_support.py b/libensemble/tools/test_support.py index 7e0727125..97e746a21 100644 --- a/libensemble/tools/test_support.py +++ b/libensemble/tools/test_support.py @@ -1,3 +1,5 @@ +import math + from libensemble.resources.resources import Resources @@ -204,7 +206,8 @@ def check_gpu_setting(task, assert_setting=True, print_setting=False, resources= # Get expected numbers if cmd_line: - expected_nums = _safe_min(wresources.slot_count * wresources.gpus_per_rset_per_node, wresources.gen_ngpus) + gen_ngpus = math.ceil(wresources.gen_ngpus) if wresources.gen_ngpus is not None else None + expected_nums = _safe_min(len(set(wresources.gpu_ids_on_node)) * wresources.gpus_per_rset_per_node, gen_ngpus) if gpus_per_task: stype = "runline option: gpus per task" expected_nums //= int(ppn) @@ -219,9 +222,7 @@ def check_gpu_setting(task, assert_setting=True, print_setting=False, resources= gpu_setting = _get_opt_value(expected_setting, task.runline) else: stype = "Env var" - expected_nums = wresources.get_slots_as_string( - multiplier=wresources.gpus_per_rset_per_node, limit=wresources.gen_ngpus - ) + expected_nums = wresources.get_gpu_ids_as_string(limit=wresources.gen_ngpus) expected_nums = expected_nums if _set_gpus(task, wresources) else None if expected_nums is not None: expected = {expected_setting: expected_nums}