Skip to content

Validate maximum number of workers for flux #527

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Dec 20, 2024
Merged

Validate maximum number of workers for flux #527

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1641409
Validate maximum number of workers for flux
jan-janssen Dec 19, 2024
14c9381
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Dec 20, 2024
94c0d8e
fix
jan-janssen Dec 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions executorlib/interactive/executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
from executorlib.standalone.thread import RaisingThread

try: # The PyFluxExecutor requires flux-base to be installed.
from executorlib.interactive.flux import FluxPythonSpawner
from executorlib.interactive.flux import FluxPythonSpawner, validate_max_workers
except ImportError:
pass

Expand Down Expand Up @@ -226,13 +226,19 @@ def create_executor(
resource_dict["flux_executor_nesting"] = flux_executor_nesting
if block_allocation:
resource_dict["init_function"] = init_function
max_workers = validate_number_of_cores(
max_cores=max_cores,
max_workers=max_workers,
cores_per_worker=cores_per_worker,
set_local_cores=False,
)
validate_max_workers(
max_workers=max_workers,
cores=cores_per_worker,
threads_per_core=resource_dict["threads_per_core"],
)
return InteractiveExecutor(
max_workers=validate_number_of_cores(
max_cores=max_cores,
max_workers=max_workers,
cores_per_worker=cores_per_worker,
set_local_cores=False,
),
max_workers=max_workers,
executor_kwargs=resource_dict,
spawner=FluxPythonSpawner,
)
Expand Down
14 changes: 14 additions & 0 deletions executorlib/interactive/flux.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,25 @@
import os
from typing import Optional

import flux
import flux.job

from executorlib.standalone.interactive.spawner import BaseSpawner


def validate_max_workers(max_workers, cores, threads_per_core):
handle = flux.Flux()
cores_total = flux.resource.list.resource_list(handle).get().up.ncores
cores_requested = max_workers * cores * threads_per_core
if cores_total < cores_requested:
raise ValueError(
"The number of requested cores is larger than the available cores "
+ str(cores_total)
+ " < "
+ str(cores_requested)
)
Comment on lines +10 to +20
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Improve error handling and type safety.

The validation function could benefit from the following improvements:

  1. Use f-strings for better readability
  2. Add type hints
  3. Ensure proper resource cleanup

Consider this implementation:

-def validate_max_workers(max_workers, cores, threads_per_core):
+def validate_max_workers(max_workers: int, cores: int, threads_per_core: int) -> None:
     handle = flux.Flux()
-    cores_total = flux.resource.list.resource_list(handle).get().up.ncores
-    cores_requested = max_workers * cores * threads_per_core
-    if cores_total < cores_requested:
-        raise ValueError(
-            "The number of requested cores is larger than the available cores "
-            + str(cores_total)
-            + " < "
-            + str(cores_requested)
-        )
+    try:
+        cores_total = flux.resource.list.resource_list(handle).get().up.ncores
+        cores_requested = max_workers * cores * threads_per_core
+        if cores_total < cores_requested:
+            raise ValueError(
+                f"The number of requested cores is larger than the available cores: "
+                f"{cores_total} < {cores_requested}"
+            )
+    finally:
+        handle.close()
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def validate_max_workers(max_workers, cores, threads_per_core):
handle = flux.Flux()
cores_total = flux.resource.list.resource_list(handle).get().up.ncores
cores_requested = max_workers * cores * threads_per_core
if cores_total < cores_requested:
raise ValueError(
"The number of requested cores is larger than the available cores "
+ str(cores_total)
+ " < "
+ str(cores_requested)
)
def validate_max_workers(max_workers: int, cores: int, threads_per_core: int) -> None:
handle = flux.Flux()
try:
cores_total = flux.resource.list.resource_list(handle).get().up.ncores
cores_requested = max_workers * cores * threads_per_core
if cores_total < cores_requested:
raise ValueError(
f"The number of requested cores is larger than the available cores: "
f"{cores_total} < {cores_requested}"
)
finally:
handle.close()



class FluxPythonSpawner(BaseSpawner):
"""
A class representing the FluxPythonInterface.
Expand Down
10 changes: 10 additions & 0 deletions tests/test_executor_backend_flux.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,3 +113,13 @@ def test_internal_memory(self):
self.assertFalse(f.done())
self.assertEqual(f.result(), np.array([5]))
self.assertTrue(f.done())

def test_validate_max_workers(self):
with self.assertRaises(ValueError):
Executor(
max_workers=10,
resource_dict={"cores": 10, "threads_per_core": 10},
flux_executor=self.executor,
backend="flux_allocation",
block_allocation=True,
)
Loading