shogi_server/timeout_queue.rb

   1 #   queue = Queue.new
   2 #   timeout(5) do
   3 #     queue.deq
   4 #   end
   5 #
   6 # is not good since not all of stdlib is safe with respect to
   7 # asynchronous exceptions.
   8 # This class is a safe implementation.
   9 # See: http://www.ruby-forum.com/topic/107864
  10 #
  11
  12 require 'thread'
  13 require 'monitor'
  14
  15 module ShogiServer
  16
  17 class TimeoutQueue
  18   def initialize
  19     @lock = Mutex.new
  20     @messages = []
  21     @readers  = []
  22   end
  23
  24   def enq(msg)
  25     @lock.synchronize do
  26       unless @readers.empty?
  27         @readers.pop << msg
  28       else
  29         @messages.push msg
  30       end
  31     end
  32   end
  33
  34   #
  35   # @param timeout
  36   # @return nil if timeout
  37   #
  38   def deq(timeout=5)
  39     timeout_thread = nil
  40     mon = nil
  41     empty_cond = nil
  42
  43     begin
  44       reader = nil
  45       @lock.synchronize do
  46         unless @messages.empty?
  47           # fast path
  48           return @messages.shift
  49         else
  50           reader = Queue.new
  51           @readers.push reader
  52           if timeout
  53             mon = Monitor.new
  54             empty_cond = mon.new_cond
  55
  56             timeout_thread = Thread.new do
  57               mon.synchronize do
  58                 if empty_cond.wait(timeout)
  59                   # timeout
  60                   @lock.synchronize do
  61                     @readers.delete reader
  62                     reader << nil
  63                   end
  64                 else
  65                   # timeout_thread was waked up before timeout
  66                 end
  67               end
  68             end # thread
  69           end
  70         end
  71       end
  72       # either timeout or writer will send to us
  73       return reader.shift
  74     ensure
  75       # (try to) clean up timeout thread
  76       if timeout_thread
  77         mon.synchronize { empty_cond.signal }
  78         Thread.pass
  79       end
  80     end
  81   end
  82 end
  83
  84 end