Fixed kahn's topological sort and added parallel topological sort (#288)

pydatastructs/graphs/__init__.py

@@ -15,7 +15,8 @@
     strongly_connected_components,
     depth_first_search,
     shortest_paths,
-    topological_sort
+    topological_sort,
+    topological_sort_parallel
 )
 
 __all__.extend(algorithms.__all__)

pydatastructs/graphs/algorithms.py

@@ -19,7 +19,8 @@
     'strongly_connected_components',
     'depth_first_search',
     'shortest_paths',
-    'topological_sort'
+    'topological_sort',
+    'topological_sort_parallel'
 ]
 
 Stack = Queue = deque
@@ -772,27 +773,107 @@ def topological_sort(graph: Graph, algorithm: str) -> list:
     return getattr(algorithms, func)(graph)
 
 def _kahn_adjacency_list(graph: Graph) -> list:
-    S = set(graph.vertices)
-    in_degree = dict()
+    S = Queue()
+    in_degree = {u: 0 for u in graph.vertices}
     for u in graph.vertices:
         for v in graph.neighbors(u):
-            if v.name not in in_degree:
-                in_degree[v.name] = 0
             in_degree[v.name] += 1
-            if v.name in S:
-                S.remove(v.name)
+    for u in graph.vertices:
+        if in_degree[u] == 0:
+            S.append(u)
+            in_degree.pop(u)
 
     L = []
     while S:
-        n = S.pop()
+        n = S.popleft()
         L.append(n)
         for m in graph.neighbors(n):
             graph.remove_edge(n, m.name)
             in_degree[m.name] -= 1
             if in_degree[m.name] == 0:
-                S.add(m.name)
+                S.append(m.name)
                 in_degree.pop(m.name)
 
     if in_degree:
         raise ValueError("Graph is not acyclic.")
     return L
+
+def topological_sort_parallel(graph: Graph, algorithm: str, num_threads: int) -> list:
+    """
+    Performs topological sort on the given graph using given algorithm using
+    given number of threads.
+
+    Parameters
+    ==========
+
+    graph: Graph
+        The graph under consideration.
+    algorithm: str
+        The algorithm to be used.
+        Currently, following are supported,
+        'kahn' -> Kahn's algorithm as given in [1].
+    num_threads: int
+        The maximum number of threads to be used.
+
+    Returns
+    =======
+
+    list
+        The list of topologically sorted vertices.
+
+    Examples
+    ========
+
+    >>> from pydatastructs import Graph, AdjacencyListGraphNode, topological_sort_parallel
+    >>> v_1 = AdjacencyListGraphNode('v_1')
+    >>> v_2 = AdjacencyListGraphNode('v_2')
+    >>> graph = Graph(v_1, v_2)
+    >>> graph.add_edge('v_1', 'v_2')
+    >>> topological_sort_parallel(graph, 'kahn', 1)
+    ['v_1', 'v_2']
+
+    References
+    ==========
+
+    .. [1] https://en.wikipedia.org/wiki/Topological_sorting#Kahn's_algorithm
+    """
+    import pydatastructs.graphs.algorithms as algorithms
+    func = "_" + algorithm + "_" + graph._impl + '_parallel'
+    if not hasattr(algorithms, func):
+        raise NotImplementedError(
+        "Currently %s algorithm isn't implemented for "
+        "performing topological sort on %s graphs."%(algorithm, graph._impl))
+    return getattr(algorithms, func)(graph, num_threads)
+
+def _kahn_adjacency_list_parallel(graph: Graph, num_threads: int) -> list:
+    num_vertices = len(graph.vertices)
+
+    def _collect_source_nodes(graph: Graph) -> list:
+        S = []
+        in_degree = {u: 0 for u in graph.vertices}
+        for u in graph.vertices:
+            for v in graph.neighbors(u):
+                in_degree[v.name] += 1
+        for u in in_degree:
+            if in_degree[u] == 0:
+                S.append(u)
+        return list(S)
+
+    def _job(graph: Graph, u: str):
+        for v in graph.neighbors(u):
+            graph.remove_edge(u, v.name)
+
+    L = []
+    source_nodes = _collect_source_nodes(graph)
+    while source_nodes:
+        with ThreadPoolExecutor(max_workers=num_threads) as Executor:
+            for node in source_nodes:
+                L.append(node)
+                Executor.submit(_job, graph, node)
+        for node in source_nodes:
+            graph.remove_vertex(node)
+        source_nodes = _collect_source_nodes(graph)
+
+    if len(L) != num_vertices:
+        raise ValueError("Graph is not acyclic.")
+    return L

pydatastructs/graphs/tests/test_algorithms.py

@@ -1,7 +1,8 @@
 from pydatastructs import (breadth_first_search, Graph,
 breadth_first_search_parallel, minimum_spanning_tree,
 minimum_spanning_tree_parallel, strongly_connected_components,
-depth_first_search, shortest_paths, topological_sort)
+depth_first_search, shortest_paths, topological_sort,
+topological_sort_parallel)
 from pydatastructs.utils.raises_util import raises
 
 def test_breadth_first_search():
@@ -292,7 +293,7 @@ def _test_shortest_paths(ds, algorithm):
 
 def test_topological_sort():
 
-    def _test_topological_sort(ds, algorithm):
+    def _test_topological_sort(func, ds, algorithm, threads=None):
         import pydatastructs.utils.misc_util as utils
         GraphNode = getattr(utils, "Adjacency" + ds + "GraphNode")
         vertices = [GraphNode('2'), GraphNode('3'), GraphNode('5'),
@@ -309,7 +310,13 @@ def _test_topological_sort(ds, algorithm):
         graph.add_edge('11', '9')
         graph.add_edge('11', '10')
         graph.add_edge('8', '9')
-        l = topological_sort(graph, algorithm)
+        if threads is not None:
+            l = func(graph, algorithm, threads)
+        else:
+            l = func(graph, algorithm)
         assert all([(l1 in l[0:3]) for l1 in ('3', '5', '7')] +
                    [(l2 in l[3:5]) for l2 in ('8', '11')] +
                    [(l3 in l[5:]) for l3 in ('10', '9', '2')])
+
+    _test_topological_sort(topological_sort, "List", "kahn")
+    _test_topological_sort(topological_sort_parallel, "List", "kahn", 3)