我通过解析XML文档动态构建DAG。XML文档没有明确指定树结构,但提供整数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()以返回Tree :: TreeNode#name)。
现在,我的程序在功能上是正确的。我进行了大量测试,结果正常:我只需要在代码中的适当位置拯救TSort :: Cyclic,如果我尝试添加导致循环的节点,则该节点将被删除,并且我可以记录问题以后处理(通过修复源数据)。
问题在于,在一个大小为75000的RubyTree上,边的数量非常接近于顶点数减1的情况下,迭代运行Tarjan算法会产生看起来相当二次的算法复杂度。Tarjan本身是O(|V| + |E|),在我的情况下大约是O(2 * |V|),但每次调用add()时,|V|增加1,因为我逐个建立节点构建图形。我不能简单地在最后调用Tarjan,因为我可能需要在读取-比较-添加循环期间遍历图形或其部分,并且任何遍历尝试都可能使程序挂起或崩溃,如果实际上有一个循环。(不用说,我的代码是单线程的;如果不是,我们将面临巨大的问题。就目前而言,我依赖于事实,即如果存在循环,则add()永远不会返回而会引发异常,即使存在循环,节点也会以清除循环的方式被删除,然后add()才会返回。) 但是它太慢了!仅对于这个循环,它花费了半个多小时,而我的程序包括几个其他步骤,它们也需要相应的时间。但就目前而言,仅执行Tarjan算法就占据了大部分性能,根据
ruby-perf
的结果判断。我尝试在RubyTree each
实现中从数组切换到链表,但只通过删除许多Array#concat
调用将运行时缩短了约1%。我发现了this一篇由Tarjan发明强连通分量算法的论文,Ruby的
TSort
就依赖于此。而且,增量循环检测似乎是一个活跃的研究领域。但是,这篇论文的对话水平远超过我的理解能力,我相信我缺乏将论文中的发现转化为Ruby代码的数学背景。不仅如此,在阅读论文的“备注”部分时,他们最好的算法自身的最坏运行时间也令人担忧,因此根据我的数据具体情况,它甚至可能不比我当前的方法更快。我是否遗漏了一些愚蠢的东西,或者我的最佳选择是付出精神努力来分析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个顶点,在这种情况下,我可以接受在现代台式工作站上需要几分钟(“去喝杯咖啡”),但不能是几个小时或更长时间。)
child.__id__
(Fixnum)作为@children_hash
的键,而不是child.name
(String),可能会带来一些性能改进。抱歉,虽然没有尝试过实际测试。如果某种方式hash
方法对于Fixnums比对于Strings更快,那可能会有所帮助。如果不是,比较相同的对象(字符串或Fixnums - 都无所谓)是很快的。如果我记得正确的话,那就是在找到潜在的相同键之后,Hash代码检查的第一件事情。 - moonflyhash = {}; hash[a] = true; t1 = Time.now; 10000.times { hash.include?(a) }; t2 = Time.now; t2-t1
和hash = {}; hash[a.__id__] = true; t1 = Time.now; 10000.times { hash.include?(a.__id__) }; t2 = Time.now; t2-t1
,其中a =“a”。没有区别:(。如果您比较第一种方法在a ="a"
和a=1
的性能,则有2x-1.5x的改进,但调用__id__
会消耗所有差异。因此,请忽略我的先前评论。 - moonfly