以下函数式 Scala 代码生成一个映射,为图形的每个节点分配一个代表。每个代表标识一个强连通分量。该代码基于 Tarjan 的强连通分量算法。
为了理解算法,理解 dfs 函数的折叠和契约可能就足够了。
def scc[T](graph:Map[T,Set[T]]): Map[T,T] = {
//`dfs` finds all strongly connected components below `node`
//`path` holds the the depth for all nodes above the current one
//'sccs' holds the representatives found so far; the accumulator
def dfs(node: T, path: Map[T,Int], sccs: Map[T,T]): Map[T,T] = {
//returns the earliest encountered node of both arguments
//for the case both aren't on the path, `old` is returned
def shallowerNode(old: T,candidate: T): T =
(path.get(old),path.get(candidate)) match {
case (_,None) => old
case (None,_) => candidate
case (Some(dOld),Some(dCand)) => if(dCand < dOld) candidate else old
}
//handle the child nodes
val children: Set[T] = graph(node)
//the initially known shallowest back-link is `node` itself
val (newState,shallowestBackNode) = children.foldLeft((sccs,node)){
case ((foldedSCCs,shallowest),child) =>
if(path.contains(child))
(foldedSCCs, shallowerNode(shallowest,child))
else {
val sccWithChildData = dfs(child,path + (node -> path.size),foldedSCCs)
val shallowestForChild = sccWithChildData(child)
(sccWithChildData, shallowerNode(shallowest, shallowestForChild))
}
}
newState + (node -> shallowestBackNode)
}
//run the above function, so every node gets visited
graph.keys.foldLeft(Map[T,T]()){ case (sccs,nextNode) =>
if(sccs.contains(nextNode))
sccs
else
dfs(nextNode,Map(),sccs)
}
}
我仅在 Wikipedia 页面上的示例图上测试了代码。
与命令式版本的区别
与最初的实现相比,我的版本避免显式展开堆栈,并简单地使用适当的(非尾)递归函数。堆栈由名为path 的持久映射表示。在我的第一个版本中,我使用 List 作为堆栈;但这效率较低,因为必须搜索包含元素。
效率
代码相当高效。对于每条边,您必须更新和/或访问不可变映射path,成本为O(log|N|),总计O(|E| log|N|)。这与命令式版本实现的O(|E|) 形成对比。
线性时间实现
Chris Okasaki 的答案中的论文在 Haskell 中提供了一个线性时间解决方案,用于查找强连通分量。他们的实现是基于 Kosaraju 的用于寻找 SCC 的算法,该算法基本上需要两次深度优先遍历。这篇论文的主要贡献似乎是在 Haskell 中实现了一个惰性的线性时间 DFS。
要实现线性时间解决方案,他们需要一组带有 O(1) 单例添加和成员资格测试的集合。这基本上与使此答案中给出的解决方案具有比命令式解决方案更高的复杂性的问题相同。他们使用 Haskell 中的状态线程来解决它,这也可以在 Scala 中完成(参见 Scalaz)。所以如果愿意把代码弄得比较复杂的话,可以将 Tarjan 的 SCC 算法实现为一个功能性的O(|E|) 版本。