[ruby-list:6794] Re: Mutex/ConditionVariable/Queue

前田です。

In message "[ruby-list:6791] Re: Mutex/ConditionVariable/Queue"
Shugo Maeda <[e-mail]> wrote:

|と思ったのですが、signalしたスレッドから条件を偽にしようとする
|スレッドに移った後、waitしたスレッドに移らずにまたsignalしたス
|レッドに戻ってMutexが解放され、その後さらにwaitしたスレッドに
|移らず、条件を偽にしようとするスレッドに移る、というケースがもし
|あるとまずいですね。
|# Rubyのスケジューラでこういうことが起きないことが保証されていれば
|# 大丈夫なのですが。

eval.cを覗いてみたのですが、こういうことが起きる心配はないことが
わかりました。
# signalしたスレッドから他のスレッドに制御が移った場合、再び
# signalしたスレッドに制御が移る前に必ずwaitしたスレッドにも
# 制御が移ると思います。たぶん(^^;

厳密に言えばwaitしたスレッドに制御が渡っても__cond_lockする前に
他のスレッドに制御が移ってしまうとまずいのですが、それはsignal
した後すぐにwaitしたスレッドに制御を渡すようにした場合でも同じ
ですよね。
実際問題としてそういうことはまずないと思うのですが...。

In message "[ruby-list:6792] Re: Mutex/ConditionVariable/Queue"
Yukihiro Matsumoto <[e-mail]> wrote:

|||eval.cをいじってくれても全然構いませんよ ^^;;;
||
||というのはやはり辛いので(^^;
|
|私でも出来るんで対して辛いことはないと思いますよ．^^;;;

多分これでeval.cをいじらなくても大丈夫だと思います(^^;

実際に実装してみました。
SizedQueueもConditionVariableを使って書いてみましたが、きちんと
動作確認していないのでうまく動くかわかりません(^^;

Mutex#__cond_lockというのが苦しいところです。
C++ならfriendにするんでしょうね(^^;

class Mutex
  
  attr :owner
  
  def initialize
    @waiting = []
    @cond_waiting = []
    @owner = nil
  end

  def locked?
    return @owner != nil
  end

  def try_lock
    result = false
    Thread.critical = true
    if @owner.nil?
      @owner = Thread.current
      result = true
    end
    Thread.critical = false
    return result
  end

  def lock
    while (Thread.critical = true; @owner != nil)
      @waiting.push(Thread.current)
      Thread.stop
    end
    @owner = Thread.current
    Thread.critical = false
  end

  def unlock
    if @owner != Thread.current
      raise ThreadError, "current thread not owner"
    end
    Thread.critical = true
    if @cond_waiting.empty?
      t = @waiting.shift
    else
      t = @cond_waiting.shift
    end
    @owner = nil
    Thread.critical = false
    t.run if t
  end

  def synchronize
    begin
      lock
      yield
    ensure
      unlock
    end
  end
  
  # DO NOT CALL THIS METHOD
  def __cond_lock
    while (Thread.critical = true; @owner != nil)
      @cond_waiting.push(Thread.current)
      Thread.stop
    end
    @owner = Thread.current
    Thread.critical = false
  end
end

class ConditionVariable
  def initialize(mutex)
    @mutex = mutex
    @waiting = []
    @waiting_mutex = Mutex.new
  end
  
  def wait
    @mutex.unlock
    @waiting_mutex.synchronize {
      @waiting.push(Thread.current)
    }
    Thread.stop
    @mutex.__cond_lock
  end
  
  def signal
    if @mutex.owner != Thread.current
      raise ThreadError, "current thread not owner"
    end
    @waiting_mutex.synchronize {
      t = @waiting.shift
      t.run if t
    }
  end
    
  def broadcast
    if @mutex.owner != Thread.current
      raise ThreadError, "current thread not owner"
    end
    @waiting_mutex.synchronize {
      for t in @waiting
	t.run
      end
      @waiting.clear
    }
  end
  
  def num_waiting
    return @waiting.length
  end
end

class Queue
  def initialize
    @que = []
    @que_mutex = Mutex.new
    @que_cond = ConditionVariable.new(@que_mutex)
  end

  def push(obj)
    @que_mutex.synchronize {
      @que.push(obj)
      @que_cond.signal
    }
  end

  def pop(noblock = false)
    @que_mutex.synchronize {
      if noblock and @que.length == 0
	raise ThreadError, "queue empty"
      end
      while @que.empty?
	@que_cond.wait
      end
      return @que.shift
    }
  end

  def empty?
    @que.length == 0
  end

  def length
    @que.length
  end
  alias size length
  
  def num_waiting
    return @que_cond.num_waiting
  end
end

class SizedQueue < Queue
  
  attr :max
  
  def initialize(max)
    @max = max
    @sized_que_mutex = Mutex.new
    @sized_que_cond = ConditionVariable.new(@sized_que_mutex)
    super()
  end

  def push(obj)
    @sized_que_mutex.synchronize {
      while @que.length >= @max
	@sized_que_cond.wait
      end
      super(obj)
    }
  end

  def pop(*args)
    @sized_que_mutex.synchronize {
      obj = super
      if @que.length < @max
	@sized_que_cond.signal
      end
      return obj
    }
  end
  
  def max=(max)
    @sized_que_mutex.synchronize {
      @max = max
      @sized_que_cond.broadcast
    }
  end
  
  def num_waiting
    return super + @sized_que_cond.num_waiting
  end
end

-- 
前田 修吾