added get_chunks, updated heatmap(better speedups)

nx_parallel/algorithms/bipartite/redundancy.py

@@ -2,14 +2,22 @@
 from networkx.algorithms.bipartite.redundancy import _node_redundancy
 import networkx as nx
 import nx_parallel as nxp
+from itertools import chain
 
 
 __all__ = ["node_redundancy"]
 
 
-def node_redundancy(G, nodes=None):
-    """In the parallel implementation we compute the node redundancy coefficients
-    for all the nodes in the bipartite graph `G` concurrently.
+def node_redundancy(G, nodes=None, get_chunks="chunks"):
+    """In the parallel implementation we divide the nodes into chunks and compute
+    the node redundancy coefficients for all `node_chunk` in parallel.
+
+    Parameters
+    ------------
+    get_chunks : str, function (default = "chunks")
+        A function that takes in an iterable of all the nodes as input and returns
+        an iterable `node_chunks`. The default chunking is done by slicing the
+        `G.nodes` (or `nodes`) into `n` chunks, where `n` is the number of CPU cores.
 
     networkx.bipartite.node_redundancy : https://networkx.org/documentation/stable/reference/algorithms/generated/networkx.algorithms.bipartite.redundancy.node_redundancy.html"""
 
@@ -23,7 +31,15 @@ def node_redundancy(G, nodes=None):
             " that has fewer than two neighbors."
         )
     total_cores = nxp.cpu_count()
+    if get_chunks == "chunks":
+        num_in_chunk = max(len(nodes) // total_cores, 1)
+        node_chunks = nxp.chunks(nodes, num_in_chunk)
+    else:
+        node_chunks = get_chunks(nodes)
     node_redundancies = Parallel(n_jobs=total_cores)(
-        delayed(lambda G, v: (v, _node_redundancy(G, v)))(G, v) for v in nodes
+        delayed(
+            lambda G, node_chunk: [(v, _node_redundancy(G, v)) for v in node_chunk]
+        )(G, node_chunk)
+        for node_chunk in node_chunks
     )
-    return dict(node_redundancies)
+    return dict(chain.from_iterable((node_redundancies)))

timing/timing_individual_function.py

@@ -14,44 +14,49 @@
 # for bipartite graphs
 # n = [50, 100, 200, 400]
 # m = [25, 50, 100, 200]
-number_of_nodes_list = [10, 50, 100, 300, 500]
+number_of_nodes_list = [75, 150, 300, 600]
+weighted = False
 pList = [1, 0.8, 0.6, 0.4, 0.2]
 currFun = nx.bipartite.node_redundancy
 for p in pList:
     for num in range(len(number_of_nodes_list)):
         # create original and parallel graphs
-        G = nx.fast_gnp_random_graph(num, p, seed=42, directed=True)
+        G = nx.fast_gnp_random_graph(
+            number_of_nodes_list[num], p, seed=42, directed=True
+        )
+
         """
         # for bipartite.node_redundancy
         G = nx.bipartite.random_graph(n[num], m[num], p, seed=42, directed=True)
         for i in G.nodes:
             l = list(G.neighbors(i))
-            if len(l)==0:
+            if len(l) == 0:
                 v = random.choice(list(G.nodes) - [i,])
                 G.add_edge(i, v)
-                G.add_edge(i, random.choice(list(G.nodes) - [i, v]))
-            elif len(l)==1:
-                G.add_edge(i, random.choice(list(G.nodes) - [i, list(G.neighbors(i))[0]]))
+                G.add_edge(i, random.choice([node for node in G.nodes if node != i]))
+            elif len(l) == 1:
+                G.add_edge(i, random.choice([node for node in G.nodes if node != i and node not in list(G.neighbors(i))]))
         """
 
         # for weighted graphs
-        random.seed(42)
-        for u, v in G.edges():
-            G[u][v]["weight"] = random.random()
+        if weighted:
+            random.seed(42)
+            for u, v in G.edges():
+                G[u][v]["weight"] = random.random()
 
         H = nx_parallel.ParallelGraph(G)
 
         # time both versions and update heatmapDF
         t1 = time.time()
-        c = currFun(H)
-        if isinstance(c, types.GeneratorType):
-            d = dict(c)
+        c1 = currFun(H)
+        if isinstance(c1, types.GeneratorType):
+            d = dict(c1)
         t2 = time.time()
         parallelTime = t2 - t1
         t1 = time.time()
-        c = currFun(G)
-        if isinstance(c, types.GeneratorType):
-            d = dict(c)
+        c2 = currFun(G)
+        if isinstance(c2, types.GeneratorType):
+            d = dict(c2)
         t2 = time.time()
         stdTime = t2 - t1
         timesFaster = stdTime / parallelTime