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

けいじゅ＠日本ラショナルソフトウェアです.

In [ruby-list :06794 ] the message: "[ruby-list:6794] Re:
Mutex/ConditionVariable/Queue ", on Feb/26 20:42(JST) Shugo Maeda
writes:

>前田です。

>実際に実装してみました。
>SizedQueueもConditionVariableを使って書いてみましたが、きちんと
>動作確認していないのでうまく動くかわかりません(^^;
>
>Mutex#__cond_lockというのが苦しいところです。
>C++ならfriendにするんでしょうね(^^;

それもありますが. 本来必要のない実装がMutexに入り込んでいますよね. 

この場合は, MutexのサブクラスとしてConditionVariableに必要な機能を持つ
ものを実装した方が良くないでしょうか? すると, Mutex#__cond_lock ->
ConditionMutex#lock となって気持ち良くoverloadできますしね.

ついでに, そのサブクラスをConditionVariableのスコープに入れてしまいま
す.

さらに, ConditionVariable.newでmutexを渡すようになっていますが, 外部か
ら渡した方が良いんですかね? 勝手に必要がないとしてinitializeの中に組み
込んでみました.

# 必要ならば, ConditionVariable::ConditionMutex.new を行なう必要が出てく
# るので, もうちょっと工夫の必要があるかな?

すると以下のようなプログラムになりました. いかがなものでしょう?

# やっぱり動作確認しておりません.

class Mutex
#  オリジナルMutexと同じ
end

class ConditionVariable
  class ConditionMutex
    attr :owner
  
    def initialize
      super
      @cond_waiting = []
      @owner = nil
    end

    def try_lock
      if ret = super
	@owner = Thread.current
      end
      ret
    end

    def lock
      while (Thread.critical = true; @owner != nil)
	@cond_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
  end

  def initialize
    @mutex = ConditionMutex.new
    @waiting = []
    @waiting_mutex = Mutex.new
  end

  def synchronize
    @mutex.syncronize do
      yield
    end
  end

  def wait
    @mutex.unlock
    @waiting_mutex.synchronize {
      @waiting.push(Thread.current)
    }
    Thread.stop
    @mutex.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_cond = ConditionVariable.new
  end

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

  def pop(noblock = false)
    @que_cond.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_cond = ConditionVariable.new
    super()
  end

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

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


>
>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
>
>-- 
>前田 修吾
>

__
................................石塚 圭樹＠日本ラショナルソフトェア...
----------------------------------->> e-mail: [e-mail] <<---