Ruby中的并发栈实现(互斥/CAS的相对性能？)

Question

这段代码非常简单，但它的目的是在不同平台上测试互斥/CAS的相对性能。最新版本可在以下网站找到：

https://github.com/alexdowad/showcase/blob/master/ruby-threads/concurrent_stack.rb

它附带了一个内置的基准脚本。请尝试运行并发布结果，以及您的操作系统、Ruby版本和CPU。如果您能找到任何改进风格或性能的方法，也请张贴。

为方便起见，守则的第一版转载如下：

# 3 thread-safe stack implementations
# Written by Alex Dowad

# Usage:
# stack.push(1)
# stack.peek    => 1 (1 is not removed from stack)
# stack.pop     => 1 (now 1 is removed from stack)

require 'rubygems' # for compatibility with MRI 1.8, JRuby
require 'thread'
require 'atomic'   # atomic gem must be installed

# The easy one first
class ThreadSafeStack
  def initialize
    @s,@m = [],Mutex.new
  end
  def push(value)
    @m.synchronize { @s.push(value) }
  end
  def pop
    @m.synchronize { @s.pop }
  end
  def peek
    @m.synchronize { @s.last }
  end
end

# a non-locking version which uses compare-and-swap to update stack
class ConcurrentStack
  Node = Struct.new(:value,:next)

  def initialize
    @top = Atomic.new(nil)
  end
  def push(value)
    node = Node.new(value,nil)
    @top.update { |current| node.next = current; node }
  end
  def pop
    node = nil
    @top.update do |current|
      node = current
      return if node.nil?
      node.next
    end
    node.value
  end
  def peek
    node = @top.value
    return if node.nil?
    node.value
  end
end

# same as ConcurrentStack, but additionally recycles popped nodes
# (to reduce load on GC)
# a global free list is used, and is also updated using CAS,
#   in exactly the same way as the stacks themselves
class RecyclingConcurrentStack
  Node      = Struct.new(:value,:next)
  FREE_LIST = Atomic.new(nil)

  def initialize
    @top = Atomic.new(nil)
  end
  def push(value)
    node = get_node(value)
    @top.update { |current| node.next = current; node }
  end
  def pop
    node = nil
    @top.update do |current|
      return if (node = current).nil?
      node.next
    end
    result = node.value
    FREE_LIST.update do |current|
      node.next = current
      node
    end
    result
  end
  def peek
    node = @top.value
    return if node.nil?
    node.value
  end

  private
  def get_node(val)
    # if contention causes the CAS to fail, just allocate a new node
    node = FREE_LIST.value
    if node && FREE_LIST.compare_and_swap(node,node.next)
      node.value = val
      node
    else
      Node.new(val,nil)
    end
  end
end

# Test driver
if __FILE__ == $0
  require 'benchmark'
  ITERATIONS_PER_TEST = 1000000
  QUEUE,MUTEX = ConditionVariable.new,Mutex.new

  def wait_for_signal
    MUTEX.synchronize { QUEUE.wait(MUTEX) }
  end
  def send_signal
    MUTEX.synchronize { QUEUE.broadcast }
  end

  def test(klass)
    test_with_threads(klass,1)
    test_with_threads(klass,5)
    test_with_threads(klass,25)
  end

  def test_with_threads(klass,n_threads)
    stack = klass.new
    iterations = ITERATIONS_PER_TEST / n_threads
    puts "Testing #{klass} with #{n_threads} thread#{'s' if n_threads>1}, iterating #{iterations}x each"

    threads = n_threads.times.collect do
                Thread.new do
                  wait_for_signal
                  iterations.times do
                    stack.push(rand(100))
                    stack.peek
                    stack.pop
                  end
                end
              end
    n_gc = GC.count
    sleep(0.001)

    result = Benchmark.measure do
      send_signal
      threads.each { |t| t.join }
    end
    puts result
    puts "Garbage collector ran #{GC.count - n_gc} times"
  end

  test(ThreadSafeStack)
  test(ConcurrentStack)
  test(RecyclingConcurrentStack)
end

Answer 1

我从github获取了最新的源代码，并开始对该项目进行审查。以下是我的基准测试结果：

# Intel(R) Core(TM)2 Duo CPU     P7570  @ 2.26GHz
# ruby 1.9.2p290 (2011-07-09 revision 32553) [i686-linux]
Testing ThreadSafeStack with 1 thread, iterating 1000000x each
  2.250000   0.000000   2.250000 (  2.254392)
Garbage collector ran 0 times
Testing ThreadSafeStack with 5 threads, iterating 200000x each
  2.350000   0.020000   2.370000 (  2.533645)
Garbage collector ran 0 times
Testing ThreadSafeStack with 25 threads, iterating 40000x each
  2.890000   0.780000   3.670000 (  3.170657)
Garbage collector ran 0 times
Testing ConcurrentStack with 1 thread, iterating 1000000x each
  2.840000   0.000000   2.840000 (  2.836545)
Garbage collector ran 54 times
Testing ConcurrentStack with 5 threads, iterating 200000x each
  2.850000   0.010000   2.860000 (  2.871540)
Garbage collector ran 54 times
Testing ConcurrentStack with 25 threads, iterating 40000x each
  2.880000   0.000000   2.880000 (  2.881745)
Garbage collector ran 56 times
Testing RecyclingConcurrentStack with 1 thread, iterating 1000000x each
  3.910000   0.000000   3.910000 (  3.906049)
Garbage collector ran 0 times
Testing RecyclingConcurrentStack with 5 threads, iterating 200000x each
  3.900000   0.000000   3.900000 (  3.910919)
Garbage collector ran 0 times
Testing RecyclingConcurrentStack with 25 threads, iterating 40000x each
  3.960000   0.000000   3.960000 (  4.024026)
Garbage collector ran 0 times

使用JRuby1.6.5(ruby-1.8.7-P 330)，我遇到了一个死锁。在JRruby中，我每次运行基准时都会复制它，但是到目前为止，我只使用了一次MRI。我会设法找出原因，并在这里发布一个更新。

更新

我回顾了并发堆栈的实现，没有发现任何可能导致死锁或其他问题的东西。看起来死锁是由基准测试代码造成的。对于JRuby，它只是挂起(非常频繁)。通过MRI，我得到了“死锁检测”的异常：

Testing ConcurrentStack with 25 threads, iterating 40000x each
/home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `join': deadlock detected (fatal)
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `block (2 levels) in test_with_threads'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `each'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `block in test_with_threads'
    from /usr/share/ruby-rvm/rubies/ruby-1.9.2-p290/lib/ruby/1.9.1/benchmark.rb:295:in `measure'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:141:in `test_with_threads'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:120:in `test'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:150:in `<top (required)>'
    from -e:1:in `load'
    from -e:1:in `<main>'

和

/home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `join': deadlock detected (fatal)
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `block (2 levels) in test_with_threads'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `each'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:143:in `block in test_with_threads'
Testing RecyclingConcurrentStack with 1 thread, iterating 1000000x each
    from /usr/share/ruby-rvm/rubies/ruby-1.9.2-p290/lib/ruby/1.9.1/benchmark.rb:295:in `measure'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:141:in `test_with_threads'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:118:in `test'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:152:in `block in <top (required)>'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:151:in `loop'
    from /home/alex/Projects/read-write-lock/concurrent-stack.rb:151:in `<top (required)>'
    from -e:1:in `load'
    from -e:1:in `<main>'

据我所见，唯一可能导致死锁的情况是主线程执行QUEUE.broadcast，然后在线程上执行Thread#join，但该线程尚未执行QUEUE.wait。然后主线程无限地等待这个线程，MRI可以检测到这样的死锁情况。

当我增加从0.001到0.01的睡眠时间时，死锁似乎已经消失了。看起来，这个问题是由线程初始化时间过长引起的--有时它无法在QUEUE.broadcast调用之前对所有线程启动块执行。

另外，作为一个小小的改进，可以将以下代码提取到私有函数(如cache_node )中，以提高pop方法的可读性：

FREE_LIST.update do |current|
  node.next = current
  node
end

堆栈实现看起来正确而清晰，我没有发现其他问题。

对于睡眠==，JRuby的0.05基准测试如下：

# jruby 1.6.5 (ruby-1.8.7-p330) (2011-10-25 9dcd388) (Java HotSpot(TM) Server VM 1.6.0_26) [linux-i386-java]
Testing ThreadSafeStack with 1 thread, iterating 1000000x each
  2.881000   0.000000   2.881000 (  2.881000)
Testing ThreadSafeStack with 5 threads, iterating 200000x each
  2.658000   0.000000   2.658000 (  2.658000)
Testing ThreadSafeStack with 25 threads, iterating 40000x each
  2.885000   0.000000   2.885000 (  2.885000)
Testing ConcurrentStack with 1 thread, iterating 1000000x each
  2.142000   0.000000   2.142000 (  2.142000)
Testing ConcurrentStack with 5 threads, iterating 200000x each
  1.084000   0.000000   1.084000 (  1.084000)
Testing ConcurrentStack with 25 threads, iterating 40000x each
  0.969000   0.000000   0.969000 (  0.970000)
Testing RecyclingConcurrentStack with 1 thread, iterating 1000000x each
  2.628000   0.000000   2.628000 (  2.628000)
Testing RecyclingConcurrentStack with 5 threads, iterating 200000x each
  1.436000   0.000000   1.436000 (  1.436000)
Testing RecyclingConcurrentStack with 25 threads, iterating 40000x each
  1.473000   0.000000   1.473000 (  1.473000)