スレッド間のバリア同期

Rubyでバリア同期です。たまに欲しくなるんだけど、標準ライブラリにはないみたいなので(もしあったら教えてください)。
バリア同期は、複数のスレッド間で処理を再開するタイミングの同期を取る(待ち合わせをする)ために使うことができます。

コード

barrier.rb

#!/usr/bin/env ruby
# coding: utf-8
require 'monitor'

class Barrier
  def initialize(thresh)
    @thresh = thresh
    @obj = {count: 0}
    @obj.extend(MonitorMixin)
    @cond = @obj.new_cond
  end

  def wait
    @obj.synchronize do
      raise 'Cannot enter twice within same barrier' unless @obj[:count]

      @obj[:count] += 1
      @cond.broadcast
      @cond.wait_until { @obj[:count].nil? or @obj[:count] >= @thresh }

      @obj[:count] = nil
    end

    true
  end
end

使い方

まずBarrierインスタンスを作ります。ここでは、3人がwaitしたら動き出すインスタンスを作っておきます。

@barrier = Barrier.new(3)

次に、いくつかのスレッドを起動します。

alice = Thread.start do
  puts 'Alice: prepare...'
  sleep(5)  # 初期化に5秒かかるスレッドを想定

  puts 'Alice: waiting'
  @barrier.wait  # ここで待ち合わせ

  puts 'Alice: Starting the task'
  puts 'Alice: Foo'
  puts 'Alice: finished'
end

bob = Thread.start do
  puts 'Bob: prepare...'
  sleep(1)  # 初期化に1秒かかるスレッドを想定

  puts 'Bob: waiting'
  @barrier.wait  # ここで待ち合わせ

  puts 'Bob: Starting the task'
  puts 'Bob: Bar'
  puts 'Bob: finished'
end

puts 'MainThread: prepare...'
puts 'MainThread: waiting'
@barrier.wait  # ここで待ち合わせ

puts 'MainThread: Starting the task'
alice.join
bob.join
puts 'MainThread: Finished'

alice, bob, そしてメインスレッドの3者が同じBarrierインスタンスの#waitを呼び出すまで待機し、待ち合わせが完了したら同時に動き出します。

実行例

Bob: prepare...
MainThread: prepare...
Alice: prepare...
MainThread: waiting
Bob: waiting
Alice: waiting
Alice: Starting the task
Alice: Foo
Bob: Starting the task
Bob: Bar
Bob: finished
Alice: finished
MainThread: Starting the task
MainThread: Finished