Merge pull request #222 from NeurodataWithoutBorders/negative_job_slicing

Add negative job slicing

Author: CodyCBakerPhD

nwbinspector/nwbinspector.py

@@ -29,7 +29,7 @@
 )
 from .register_checks import InspectorMessage, Importance
 from .tools import get_s3_urls_and_dandi_paths
-from .utils import FilePathType, PathType, OptionalListOfStrings, robust_s3_read
+from .utils import FilePathType, PathType, OptionalListOfStrings, robust_s3_read, calculate_number_of_cpu
 
 INTERNAL_CONFIGS = dict(dandi=Path(__file__).parent / "internal_configs" / "dandi.inspector_config.yaml")
 
@@ -318,6 +318,8 @@ def inspect_all(
         The default is the lowest level, BEST_PRACTICE_SUGGESTION.
     n_jobs : int
         Number of jobs to use in parallel. Set to -1 to use all available resources.
+        This may also be a negative integer x from -2 to -(number_of_cpus - 1) which acts like negative slicing by using
+        all available CPUs minus x.
         Set to 1 (also the default) to disable.
     skip_validate : bool, optional
         Skip the PyNWB validation step. This may be desired for older NWBFiles (< schema version v2.10).
@@ -341,6 +343,7 @@ def inspect_all(
         Importance[importance_threshold] if isinstance(importance_threshold, str) else importance_threshold
     )
     modules = modules or []
+    n_jobs = calculate_number_of_cpu(requested_cpu=n_jobs)
     if progress_bar_options is None:
         progress_bar_options = dict(position=0, leave=False)
         if stream:

nwbinspector/tools.py

@@ -7,7 +7,7 @@
 
 from pynwb import NWBFile
 
-from .utils import is_module_installed
+from .utils import is_module_installed, calculate_number_of_cpu
 
 
 def make_minimal_nwbfile():
@@ -43,6 +43,7 @@ def get_s3_urls_and_dandi_paths(dandiset_id: str, version_id: Optional[str] = No
     ), "The specified 'path' is not a proper DANDISet ID. It should be a six-digit numeric identifier."
 
     s3_urls_to_dandi_paths = dict()
+    n_jobs = calculate_number_of_cpu(requested_cpu=n_jobs)
     if n_jobs != 1:
         with DandiAPIClient() as client:
             dandiset = client.get_dandiset(dandiset_id=dandiset_id, version_id=version_id)

nwbinspector/utils.py

@@ -1,4 +1,5 @@
 """Commonly reused logic for evaluating conditions; must not have external dependencies."""
+import os
 import re
 import json
 import numpy as np
@@ -130,3 +131,25 @@ def robust_s3_read(
         except Exception as exc:
             raise exc
     raise TimeoutError(f"Unable to complete the command ({command.__name__}) after {max_retries} attempts!")
+
+
+def calculate_number_of_cpu(requested_cpu: int = 1) -> int:
+    """
+    Calculate the number CPUs to use with respect to negative slicing and check against maximal available resources.
+
+    Parameters
+    ----------
+    requested_cpu : int, optional
+        The desired number of CPUs to use.
+
+        The default is 1.
+    """
+    total_cpu = os.cpu_count()
+    assert requested_cpu <= total_cpu, f"Requested more CPUs ({requested_cpu}) than are available ({total_cpu})!"
+    assert requested_cpu >= -(
+        total_cpu - 1
+    ), f"Requested fewer CPUs ({requested_cpu}) than are available ({total_cpu})!"
+    if requested_cpu > 0:
+        return requested_cpu
+    else:
+        return total_cpu + requested_cpu

tests/test_utils.py

@@ -1,9 +1,16 @@
+import os
 from packaging import version
 
 from hdmf.testing import TestCase
 
 from nwbinspector import Importance
-from nwbinspector.utils import format_byte_size, check_regular_series, is_dict_in_string, get_package_version
+from nwbinspector.utils import (
+    format_byte_size,
+    check_regular_series,
+    is_dict_in_string,
+    get_package_version,
+    calculate_number_of_cpu,
+)
 
 
 def test_format_byte_size():
@@ -104,3 +111,27 @@ def test_get_package_version_type():
 
 def test_get_package_version_value():
     assert get_package_version("hdmf") >= version.parse("3.1.1")  # minimum supported PyNWB version
+
+
+class TestCalulcateNumberOfCPU(TestCase):
+    total_cpu = os.cpu_count()
+
+    def test_request_more_than_available_assert(self):
+        requested_cpu = 2500
+        with self.assertRaisesWith(
+            exc_type=AssertionError,
+            exc_msg=f"Requested more CPUs ({requested_cpu}) than are available ({self.total_cpu})!",
+        ):
+            calculate_number_of_cpu(requested_cpu=requested_cpu)
+
+    def test_request_fewer_than_available_assert(self):
+        requested_cpu = -2500
+        with self.assertRaisesWith(
+            exc_type=AssertionError,
+            exc_msg=f"Requested fewer CPUs ({requested_cpu}) than are available ({self.total_cpu})!",
+        ):
+            calculate_number_of_cpu(requested_cpu=requested_cpu)
+
+    def test_calculate_number_of_cpu_negative_value(self):
+        requested_cpu = -1  # CI only has 2 jobs available
+        assert calculate_number_of_cpu(requested_cpu=requested_cpu) == requested_cpu % self.total_cpu