Add should_run functionality (#123)

* add should_run functionality

* added a few basic tests

* improve naming, should_run message

* remove _default_should_run import in tests

* centralise should_run

* adding skip arguments feature

* ensured test accomodates tournament graphs

* added a policy for using certain number of cores

* modify default_should_run policy

* modify should_run tests

* update docstring

* change func name

* handling exports

* minor edit

* fix naming; imports

* concise imports

* revert prev due to circular import

* remove custom

Co-authored-by: Dan Schult <dschult@colgate.edu>

---------

Co-authored-by: Dan Schult <dschult@colgate.edu>

Author: akshitasure12

nx_parallel/algorithms/isolate.py

@@ -5,7 +5,7 @@
 __all__ = ["number_of_isolates"]
 
 
-@nxp._configure_if_nx_active()
+@nxp._configure_if_nx_active(should_run=nxp.should_skip_parallel)
 def number_of_isolates(G, get_chunks="chunks"):
     """The parallel computation is implemented by dividing the list
     of isolated nodes into chunks and then finding the length of each chunk in parallel

nx_parallel/algorithms/shortest_paths/unweighted.py

@@ -15,7 +15,7 @@
 ]
 
 
-@nxp._configure_if_nx_active()
+@nxp._configure_if_nx_active(should_run=nxp.should_skip_parallel)
 def all_pairs_shortest_path_length(G, cutoff=None, get_chunks="chunks"):
     """The parallel implementation first divides the nodes into chunks and then
     creates a generator to lazily compute shortest paths lengths for each node in

nx_parallel/interface.py

@@ -86,6 +86,30 @@ def convert_to_nx(result, *, name=None):
             return result.graph_object
         return result
 
+    @classmethod
+    def should_run(cls, name, args, kwargs):
+        """Determine whether this backend should run the specified algorithm
+        with the given arguments.
+
+        Parameters
+        ----------
+        cls : type
+            `BackendInterface` class
+        name : str
+            Name of the target algorithm
+        args : tuple
+            Positional arguments passed to the algorithm's `should_run`.
+        kwargs : dict
+            Keyword arguments passed to the algorithm's `should_run`.
+
+        Returns
+        -------
+        bool or str
+            If the algorithm should run, returns True.
+            Otherwise, returns a string explaining why parallel execution is skipped.
+        """
+        return getattr(cls, name).should_run(*args, **kwargs)
+
 
 for attr in ALGORITHMS:
     setattr(BackendInterface, attr, getattr(algorithms, attr))

nx_parallel/tests/test_should_run.py

@@ -0,0 +1,73 @@
+import nx_parallel as nxp
+from nx_parallel.interface import ALGORITHMS
+import networkx as nx
+import inspect
+import pytest
+import os
+import joblib
+
+
+def get_functions_with_should_run():
+    for name, obj in inspect.getmembers(nxp.algorithms, inspect.isfunction):
+        if callable(obj.should_run):
+            yield name
+
+
+def test_get_functions_with_should_run():
+    assert set(get_functions_with_should_run()) == set(ALGORITHMS)
+
+
+def test_default_should_run():
+    @nxp._configure_if_nx_active()
+    def dummy_default():
+        pass
+
+    with pytest.MonkeyPatch().context() as mp:
+        mp.delitem(os.environ, "PYTEST_CURRENT_TEST", raising=False)
+        assert (
+            dummy_default.should_run()
+            == "Parallel backend requires `n_jobs` > 1 to run"
+        )
+
+        with joblib.parallel_config(n_jobs=4):
+            assert dummy_default.should_run()
+
+
+def test_skip_parallel_backend():
+    @nxp._configure_if_nx_active(should_run=nxp.should_skip_parallel)
+    def dummy_skip_parallel():
+        pass
+
+    assert dummy_skip_parallel.should_run() == "Fast algorithm; skip parallel execution"
+
+
+def test_should_run_if_large():
+    @nxp._configure_if_nx_active(should_run=nxp.should_run_if_large)
+    def dummy_if_large(G):
+        pass
+
+    smallG = nx.fast_gnp_random_graph(20, 0.6, seed=42)
+    largeG = nx.fast_gnp_random_graph(250, 0.6, seed=42)
+
+    assert dummy_if_large.should_run(smallG) == "Graph too small for parallel execution"
+    assert dummy_if_large.should_run(largeG)
+
+
+@pytest.mark.parametrize("func_name", get_functions_with_should_run())
+def test_should_run(func_name):
+    tournament_funcs = [
+        "tournament_is_strongly_connected",
+    ]
+
+    if func_name in tournament_funcs:
+        G = nx.tournament.random_tournament(15, seed=42)
+    else:
+        G = nx.fast_gnp_random_graph(40, 0.6, seed=42)
+    H = nxp.ParallelGraph(G)
+    func = getattr(nxp, func_name)
+
+    result = func.should_run(H)
+    if not isinstance(result, (bool, str)):
+        raise AssertionError(
+            f"{func.__name__}.should_run has an invalid return type; {type(result)}"
+        )

nx_parallel/utils/__init__.py

@@ -1,2 +1,3 @@
 from .chunk import *
 from .decorators import *
+from .should_run_policies import *

nx_parallel/utils/decorators.py

@@ -3,12 +3,13 @@
 from functools import wraps
 import networkx as nx
 from joblib import parallel_config
+from nx_parallel.utils.should_run_policies import default_should_run
 
 
 __all__ = ["_configure_if_nx_active"]
 
 
-def _configure_if_nx_active():
+def _configure_if_nx_active(should_run=None):
     """Decorator to set the configuration for the parallel computation
     of the nx-parallel algorithms.
     """
@@ -29,6 +30,10 @@ def wrapper(*args, **kwargs):
                     return func(*args, **kwargs)
             return func(*args, **kwargs)
 
+        wrapper.should_run = default_should_run
+        if should_run:
+            wrapper.should_run = should_run
+
         return wrapper
 
     return decorator

nx_parallel/utils/should_run_policies.py

@@ -0,0 +1,25 @@
+import nx_parallel as nxp
+
+
+__all__ = [
+    "default_should_run",
+    "should_skip_parallel",
+    "should_run_if_large",
+]
+
+
+def should_skip_parallel(*_):
+    return "Fast algorithm; skip parallel execution"
+
+
+def should_run_if_large(G, *_):
+    if len(G) <= 200:
+        return "Graph too small for parallel execution"
+    return True
+
+
+def default_should_run(*_):
+    n_jobs = nxp.get_n_jobs()
+    if n_jobs in (None, 0, 1):
+        return "Parallel backend requires `n_jobs` > 1 to run"
+    return True