我在 Ruby 1.9.3 中有一个构建RubyTree的程序。我的数据最好被描述为有向无环图(DAG);请注意,它不是多树。好吧,至少数据应该是一个 DAG,尽管用户尽最大努力用坏数据挫败我的程序。
我通过解析 XML 文档动态构建 DAG。XML 文档没有明确指定树结构,但确实提供了整数 ID 的交叉引用,这些 ID 在文档中的元素之间建立了链接。
我需要确保 RubyTree 不包含任何循环。源数据可能(错误地)有一个循环,如果有,我的程序需要知道它,而不是进入无限循环或崩溃。目前,为了实现这一点,我将 Ruby 标准库的TSort模块混合到 RubyTree 的Tree::TreeNode
类中。这使用 Tarjan 的算法在每次添加节点时对图执行拓扑排序。在拓扑排序期间,如果检测到循环,则会引发异常——这正是我想要的。
例如:
module Tree
class TreeNode
include TSort
def tsort_each_node(&block)
self.each(&block)
end
def tsort_each_child(node, &block)
node.get_children().each { |child| yield child }
end
def add(child, at_index = -1)
#The standard RubyTree implementation of add goes here
begin
self.tsort()
rescue TSort::Cyclic => exce
self.remove!(child)
raise exce
end
return child
end
end
end
我还必须修改其他一些方法。基本上任何需要遍历树或需要实现 TSort 的子节点,或者摆脱它对遍历的依赖(例如,我简化Tree::TreeNode#to_s()
为 return Tree::TreeNode#name
。)
现在,我的程序在功能上是正确的。我已经完成了重要的测试,结果运行良好:我所要做的就是TSort::Cyclic
在代码中的正确位置进行救援,如果我尝试添加一个导致循环的节点,该节点将被删除并且我可以登录报告中待处理的问题(通过修复源数据)。
问题是,在大小为 75000 左右的 RubyTree 上,边数非常接近等于顶点数减 1,迭代运行 Tarjan 算法会产生看起来相当二次的算法复杂度。Tarjan 本身是O(|V| + |E|)
,在我的情况下是 about O(2*|V|)
,但每次我调用 时add()
,都会|V|
增加 1,因为我逐个节点地构建图形节点。我不能简单地在最后调用 Tarjan,因为我可能需要在 read-compare-add 循环期间遍历图形或其中的一部分,并且任何遍历尝试都可能会挂起程序或使其崩溃,如果事实上存在一个循环。(不用说,我的代码是单线程的;如果不是,我们就会遇到一个大问题。就目前而言,我依赖于这样一个事实:add()
如果存在循环,则永远不会返回而不引发异常,即使存在循环,节点也会以在返回之前清除循环的方式被删除add()
。)
但是太慢了!仅此一个循环就需要半个多小时,而我的程序由其他几个步骤组成,这些步骤占用了自己相当多的时间。但就目前而言,从ruby-perf
. 我尝试在 RubyTree 实现中从数组切换到链表each
,但它通过删除大量Array#concat
调用仅将运行时间减少了约 1%。
我发现了Tarjan 的这篇很棒的论文,他发明了 Ruby 所依赖的强连接组件算法,似乎增量循环检测是一个活跃的研究领域。然而,论文中的对话水平明显超出了我的想象,而且我很确定我缺乏将论文的发现转化为 Ruby 代码的数学背景。不仅如此,通过阅读论文的备注部分,他们的尽力而为的算法似乎有自己的令人担忧的最坏情况运行时,所以它甚至可能不会比我目前的方法快,这取决于我的具体情况数据。TSort
我是否在这里遗漏了一些愚蠢的东西,或者我最好的选择是投入脑力去分析 Tarjan 的论文并尝试提出其中一种算法的 Ruby 实现?请注意,我并不特别关心算法的拓扑排序方面;这是我真正想要的副作用。如果树没有经过拓扑排序但仍然保证没有循环,我会非常高兴。
另外值得注意的是,在我的源数据中,周期有点少见。也就是说,循环可能由于数据输入过程中的手动错误而发生,但它们永远不会故意发生,并且应该始终向程序报告,以便它可以告诉我,这样我就可以用比利俱乐部击败某人进入错误的数据。此外,即使程序检测到一个特别讨厌的循环,它也绝对必须继续运行,所以我不能只是把头埋在沙子里,希望不会有任何循环。
实际问题是什么?
应某些人的要求,这是一个演示,您可以运行以查看工作中的问题。
安装 RubyTree 的稳定版本(我使用 MRI 1.9.3)。然后比较这两个程序的输出:
图表 1:打印“第三次”后,主线程上的 CPU 使用率为 100% 时永远挂起
require 'tree'
a = Tree::TreeNode.new('a', nil)
b = Tree::TreeNode.new('b', nil)
c = Tree::TreeNode.new('c', nil)
a.add(b)
a.add(c)
puts "First time"
b.add(c)
puts "Second time"
b.add(a)
puts "Third time"
c.add(b)
puts "Fourth time"
c.add(a)
puts "Fifth time"
puts "Done"
图表2:一路遍历并打印“Done”,结果没有循环
请注意,我通常会在rescue
块内做一些事情来记录发生的循环,并大声抱怨创造这些循环的人类罪犯。
require 'tree'
require 'tsort'
module Tree
class TreeNode
include TSort
def tsort_each_node(&block)
self.each(&block)
end
def tsort_each_child(node, &block)
node.get_children().each { |child| yield child}
end
def to_s
name
end
def get_children()
return @children
end
def add(child, at_index = -1)
unless child
raise ArgumentError, "Attempting to add a nil node" # Only handles the immediate child scenario
end
if self.equal?(child)
raise TSort::Cyclic, "Cycle detected: [#{child.name}, #{child.name}]"
end
# Lazy man's unique test, won't test if children of child are unique in this tree too.
if @children_hash.include?(child.name)
raise "Child #{child.name} already added!"
end
if insertion_range.include?(at_index)
@children.insert(at_index, child)
else
raise "Attempting to insert a child at a non-existent location (#{at_index}) when only positions from #{insertion_range.min} to #{insertion_range.max} exist."
end
@children_hash[child.name] = child
child.parent = self
#CYCLE DETECTION - raises TSort::Cyclic if this caused a cycle
begin
self.tsort()
rescue TSort::Cyclic => exce
self.remove!(child)
raise exce
end
return child
end
end
end
a = Tree::TreeNode.new('a', nil)
b = Tree::TreeNode.new('b', nil)
c = Tree::TreeNode.new('c', nil)
a.add(b)
a.add(c)
puts "First time"
b.add(c)
puts "Second time"
begin
b.add(a)
rescue
end
puts "Third time"
begin
c.add(b)
rescue
end
puts "Fourth time"
begin
c.add(a)
rescue
end
puts "Fifth time"
puts "Done"
对我来说,目标是开发功能上等同于图表 2 的代码,但可以更好地扩展到更大数量的顶点(我预计不会有超过 10^6 个顶点,在这种情况下我会很好)在现代桌面工作站上花费几分钟(“去喝杯咖啡”),但不是几个小时或更长时间。)