Rubyを用いたマルチスレッド対応Queueの作り方

はじめに

マルチスレッドとキューについていろいろ書いてきている。とりあえず今回は、Rubyでマルチスレッド対応なQueueを考えてみる。

いろいろなご意見

皆様からいろいろなご意見を頂いたので、記載しておく

　・キューの上限が必要なケースがよくわからなかった

　・ディープクローンが発生するとパフォーマンスが悪くなる

　・lock-freeなアルゴリズムがあるのでは

これらについては、今後また書きたい

rubyのマルチスレッド対応キュー

Rubyでマルチスレッド対応Queueが欲しければ、以下のようにすればいい。

require 'thread'
q = Queue.new

q.enq( '123' )
q.deq()
 
 

既にクラスがあるので、それを利用すればいい...

自分で作る

いや、あえて作る必要がないのであるが、作ってみる

require 'thread'

class MyQueue
  def initialize
    @empty = ConditionVariable.new
    @mutex = Mutex.new
    @q = []
  end

  def count
    @q.size
  end

  def enq(v)
    @mutex.synchronize do
      @q.push v
      @empty.signal if count == 1
    end
  end

  def deq
    @mutex.synchronize do
      @empty.wait(@mutex) if count == 0
      v = @q.shift
      v
    end
  end
end
 
 

ConditionVariable

このConditionVariableが肝である。

キューにアイテムを追加する(enq)とき、@q オブジェクトを排他するためにmutexで同期している。同様にキューからアイテムを取得する(deq)とき、@q オブジェクトを排他するためにmutexで同期している。

もしキューにアイテムがなければアイテムが来るまで、ブロックすることを考えるが、mutexで同期している最中にブロックすると誰もキュー@qにアイテムを追加できなくなるので、デッドロックしてしまう。

そこで出てくるのがConditionVariable。 ConditionVariableにmutexを渡し wait すると、アトミックに「mutexの排他を解除」しさらに「その場で待つ」という事をする。もし、waitした ConditionVariableにsignalを送り、ブロックを解除すると、waitから抜けると同時に、渡されたmutexを握ってくれる。

これを用い、deq の中で、サイズを調べもし空であれば、waitする。アイテムがない状態でenqが呼ばれると、deqの中でwaitしているConditionVariableにsignalが送られ、 waitから抜けてくる。

ちなみに

上記サンプルは ruby の online マニュアル等にも出ている例であるが、長時間負荷をかけると mutex内部で異常が発生し例外が発生する。

なお、Queue クラスではそんな事が発生しなかったので、こちらを利用した方がいい。

まとめ

rubyで マルチスレッド対応な　Queueの作り方について書いた。なお、キュー全体をロックしているので多くのスレッドが同時にアクセスするとパフォーマンスが低下する。