[ENH]: parallel implementation of is_reachable() with numpy arrays.

_nx_parallel/__init__.py

@@ -105,7 +105,7 @@ def get_info():
                 },
             },
             "is_reachable": {
-                "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/tournament.py#L13",
+                "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/tournament.py#L15",
                 "additional_docs": "The function parallelizes the calculation of two neighborhoods of vertices in `G` and checks closure conditions for each neighborhood subset in parallel.",
                 "additional_parameters": {
                     'get_chunks : str, function (default = "chunks")': "A function that takes in a list of all the nodes as input and returns an iterable `node_chunks`. The default chunking is done by slicing the `nodes` into `n_jobs` number of chunks."
@@ -147,7 +147,7 @@ def get_info():
                 },
             },
             "tournament_is_strongly_connected": {
-                "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/tournament.py#L58",
+                "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/tournament.py#L76",
                 "additional_docs": "The parallel computation is implemented by dividing the nodes into chunks and then checking whether each node is reachable from each other node in parallel.",
                 "additional_parameters": {
                     'get_chunks : str, function (default = "chunks")': "A function that takes in a list of all the nodes as input and returns an iterable `node_chunks`. The default chunking is done by slicing the `nodes` into `n_jobs` number of chunks."

nx_parallel/algorithms/tournament.py

@@ -1,7 +1,9 @@
-from joblib import Parallel, delayed
+from joblib import Parallel, delayed, dump, load
 import nx_parallel as nxp
-from networkx.algorithms.simple_paths import is_simple_path as is_path
 import networkx as nx
+import tempfile
+import shutil
+import os
 
 __all__ = [
     "is_reachable",
@@ -25,33 +27,49 @@ def is_reachable(G, s, t, get_chunks="chunks"):
         into `n_jobs` number of chunks.
     """
 
-    def two_neighborhood_close(G, chunk):
-        tnc = []
+    def two_neighborhood_close(adjM_filepath, chunk):
+        adjM = load(adjM_filepath, mmap_mode="r")
+        node_indices = range(adjM.shape[0])
         for v in chunk:
             S = {
                 x
-                for x in G
-                if x == v or x in G[v] or any(is_path(G, [v, z, x]) for z in G)
+                for x in node_indices
+                if x == v
+                or adjM[v, x]
+                or any(adjM[v, z] and adjM[z, x] for z in node_indices)
             }
-            tnc.append(not (is_closed(G, S) and s in S and t not in S))
-        return all(tnc)
+            if s_ind in S and t_ind not in S and is_closed(adjM, node_indices, S):
+                return False
+        return True
 
-    def is_closed(G, nodes):
-        return all(v in G[u] for u in set(G) - nodes for v in nodes)
+    def is_closed(adjM, node_indices, S):
+        return all(u in S or adjM[u, v] for u in node_indices for v in S)
 
     if hasattr(G, "graph_object"):
         G = G.graph_object
 
+    nodelist = list(G)
+    adjM = nx.to_numpy_array(G, dtype=bool, nodelist=nodelist)
+    nodemap = {n: i for i, n in enumerate(nodelist)}
+    s_ind = nodemap[s]
+    t_ind = nodemap[t]
+
+    temp_folder = tempfile.mkdtemp()
+    adjM_filepath = os.path.join(temp_folder, "adjMatrix.mmap")
+    dump(adjM, adjM_filepath)
+
     n_jobs = nxp.get_n_jobs()
 
     if get_chunks == "chunks":
         node_chunks = nxp.chunks(G, n_jobs)
     else:
         node_chunks = get_chunks(G)
 
-    return all(
-        Parallel()(delayed(two_neighborhood_close)(G, chunk) for chunk in node_chunks)
+    results = Parallel()(
+        delayed(two_neighborhood_close)(adjM_filepath, chunk) for chunk in node_chunks
     )
+    shutil.rmtree(temp_folder)
+    return all(results)
 
 
 @nxp._configure_if_nx_active()

timing/timing_individual_function.py

@@ -6,7 +6,9 @@
 from matplotlib import pyplot as plt
 
 import nx_parallel as nxp
+import joblib
 
+joblib.parallel_config(n_jobs=-1)
 # Code to create README heatmaps for individual function currFun
 heatmapDF = pd.DataFrame()
 # for bipartite graphs
@@ -68,12 +70,12 @@
     G = nx.tournament.random_tournament(num, seed=42)
     H = nxp.ParallelGraph(G)
     t1 = time.time()
-    c = currFun(H, 1, num)
+    c = currFun(H, 1, num - 1)
     t2 = time.time()
     parallelTime = t2 - t1
     print(parallelTime)
     t1 = time.time()
-    c = currFun(G, 1, num)
+    c = currFun(G, 1, num - 1)
     t2 = time.time()
     stdTime = t2 - t1
     print(stdTime)