lib/monitor.rb

   1 =begin
   2
   3 = monitor.rb
   4
   5 Copyright (C) 2001  Shugo Maeda <shugo@ruby-lang.org>
   6 Copyright (C) 2008  MenTaLguY <mental@rydia.net>
   7
   8 This library is distributed under the terms of the Ruby license.
   9 You can freely distribute/modify this library.
  10
  11 == example
  12
  13 This is a simple example.
  14
  15   require 'monitor.rb'
  16
  17   buf = []
  18   buf.extend(MonitorMixin)
  19   empty_cond = buf.new_cond
  20
  21   # consumer
  22   Thread.start do
  23     loop do
  24       buf.synchronize do
  25         empty_cond.wait_while { buf.empty? }
  26         print buf.shift
  27       end
  28     end
  29   end
  30
  31   # producer
  32   while line = ARGF.gets
  33     buf.synchronize do
  34       buf.push(line)
  35       empty_cond.signal
  36     end
  37   end
  38
  39 The consumer thread waits for the producer thread to push a line
  40 to buf while buf.empty?, and the producer thread (main thread)
  41 reads a line from ARGF and push it to buf, then call
  42 empty_cond.signal.
  43
  44 =end
  45
  46 require 'thread'
  47
  48 #
  49 # Adds monitor functionality to an arbitrary object by mixing the module with
  50 # +include+.  For example:
  51 #
  52 #    require 'monitor.rb'
  53 #
  54 #    buf = []
  55 #    buf.extend(MonitorMixin)
  56 #    empty_cond = buf.new_cond
  57 #
  58 #    # consumer
  59 #    Thread.start do
  60 #      loop do
  61 #        buf.synchronize do
  62 #          empty_cond.wait_while { buf.empty? }
  63 #          print buf.shift
  64 #        end
  65 #      end
  66 #    end
  67 #
  68 #    # producer
  69 #    while line = ARGF.gets
  70 #      buf.synchronize do
  71 #        buf.push(line)
  72 #        empty_cond.signal
  73 #      end
  74 #    end
  75 #
  76 # The consumer thread waits for the producer thread to push a line
  77 # to buf while buf.empty?, and the producer thread (main thread)
  78 # reads a line from ARGF and push it to buf, then call
  79 # empty_cond.signal.
  80 #
  81 module MonitorMixin
  82   #
  83   # FIXME: This isn't documented in Nutshell.
  84   #
  85   # Since MonitorMixin.new_cond returns a ConditionVariable, and the example
  86   # above calls while_wait and signal, this class should be documented.
  87   #
  88   class ConditionVariable
  89     # Create a new timer with the argument timeout, and add the
  90     # current thread to the list of waiters.  Then the thread is
  91     # stopped.  It will be resumed when a corresponding #signal
  92     # occurs.
  93     def wait(timeout = nil)
  94       condition = @condition
  95       @monitor.instance_eval { mon_wait_for_cond(condition, timeout) }
  96     end
  97
  98     # call #wait while the supplied block returns +true+.
  99     def wait_while
 100       while yield
 101         wait
 102       end
 103     end
 104
 105     # call #wait until the supplied block returns +true+.
 106     def wait_until
 107       until yield
 108         wait
 109       end
 110     end
 111
 112     # Wake up and run the next waiter
 113     def signal
 114       condition = @condition
 115       @monitor.instance_eval { mon_signal_cond(condition) }
 116       nil
 117     end
 118
 119     # Wake up all the waiters.
 120     def broadcast
 121       condition = @condition
 122       @monitor.instance_eval { mon_broadcast_cond(condition) }
 123       nil
 124     end
 125
 126     def count_waiters
 127       condition = @condition
 128       @monitor.instance_eval { mon_count_cond_waiters(condition) }
 129     end
 130
 131     private
 132
 133     def initialize(monitor, condition)
 134       @monitor = monitor
 135       @condition = condition
 136     end
 137   end
 138
 139   def self.extend_object(obj)
 140     super(obj)
 141     obj.instance_eval {mon_initialize()}
 142   end
 143
 144   #
 145   # Attempts to enter exclusive section.  Returns +false+ if lock fails.
 146   #
 147   def mon_try_enter
 148     @mon_mutex.synchronize do
 149       @mon_owner = Thread.current unless @mon_owner
 150       if @mon_owner == Thread.current
 151         @mon_count += 1
 152         true
 153       else
 154         false
 155       end
 156     end
 157   end
 158   # For backward compatibility
 159   alias try_mon_enter mon_try_enter
 160
 161   #
 162   # Enters exclusive section.
 163   #
 164   def mon_enter
 165     @mon_mutex.synchronize do
 166       mon_acquire(@mon_entering_cond)
 167       @mon_count += 1
 168     end
 169   end
 170
 171   #
 172   # Leaves exclusive section.
 173   #
 174   def mon_exit
 175     @mon_mutex.synchronize do
 176       mon_check_owner
 177       @mon_count -= 1
 178       mon_release if @mon_count.zero?
 179       nil
 180     end
 181   end
 182
 183   #
 184   # Enters exclusive section and executes the block.  Leaves the exclusive
 185   # section automatically when the block exits.  See example under
 186   # +MonitorMixin+.
 187   #
 188   def mon_synchronize
 189     mon_enter
 190     begin
 191       yield
 192     ensure
 193       mon_exit
 194     end
 195   end
 196   alias synchronize mon_synchronize
 197
 198   #
 199   # FIXME: This isn't documented in Nutshell.
 200   #
 201   # Create a new condition variable for this monitor.
 202   # This facilitates control of the monitor with #signal and #wait.
 203   #
 204   def new_cond
 205     condition = ::ConditionVariable.new
 206     condition.instance_eval { @mon_n_waiters = 0 }
 207     return ConditionVariable.new(self, condition)
 208   end
 209
 210   private
 211
 212   def initialize(*args)
 213     super
 214     mon_initialize
 215   end
 216
 217   # called by initialize method to set defaults for instance variables.
 218   def mon_initialize
 219     @mon_mutex = Mutex.new
 220     @mon_owner = nil
 221     @mon_count = 0
 222     @mon_total_waiting = 0
 223     @mon_entering_cond = ::ConditionVariable.new
 224     @mon_waiting_cond = ::ConditionVariable.new
 225     self
 226   end
 227
 228   # Throw a ThreadError exception if the current thread
 229   # does't own the monitor
 230   def mon_check_owner
 231     # called with @mon_mutex held
 232     if @mon_owner != Thread.current
 233       raise ThreadError, "current thread not owner"
 234     end
 235   end
 236
 237   def mon_acquire(condition)
 238     # called with @mon_mutex held
 239     while @mon_owner && @mon_owner != Thread.current
 240       condition.wait @mon_mutex
 241     end
 242     @mon_owner = Thread.current
 243   end
 244
 245   def mon_release
 246     # called with @mon_mutex held
 247     @mon_owner = nil
 248     if @mon_total_waiting.nonzero?
 249       @mon_waiting_cond.signal
 250     else
 251       @mon_entering_cond.signal
 252     end
 253   end
 254
 255   def mon_wait_for_cond(condition, timeout)
 256     @mon_mutex.synchronize do
 257       mon_check_owner
 258       count = @mon_count
 259       @mon_count = 0
 260       condition.instance_eval { @mon_n_waiters += 1 }
 261       begin
 262         mon_release
 263         if timeout
 264           condition.wait(@mon_mutex, timeout)
 265         else
 266           condition.wait(@mon_mutex)
 267           true
 268         end
 269       ensure
 270         @mon_total_waiting += 1
 271         # TODO: not interrupt-safe
 272         mon_acquire(@mon_waiting_cond)
 273         @mon_total_waiting -= 1
 274         @mon_count = count
 275         condition.instance_eval { @mon_n_waiters -= 1 }
 276       end
 277     end
 278   end
 279
 280   def mon_signal_cond(condition)
 281     @mon_mutex.synchronize do
 282       mon_check_owner
 283       condition.signal
 284     end
 285   end
 286
 287   def mon_broadcast_cond(condition)
 288     @mon_mutex.synchronize do
 289       mon_check_owner
 290       condition.broadcast
 291     end
 292   end
 293
 294   def mon_count_cond_waiters(condition)
 295     @mon_mutex.synchronize do
 296       condition.instance_eval { @mon_n_waiters }
 297     end
 298   end
 299 end
 300
 301 # Monitors provide means of mutual exclusion for Thread programming.
 302 # A critical region is created by means of the synchronize method,
 303 # which takes a block.
 304 # The condition variables (created with #new_cond) may be used
 305 # to control the execution of a monitor with #signal and #wait.
 306 #
 307 # the Monitor class wraps MonitorMixin, and provides aliases
 308 #  alias try_enter try_mon_enter
 309 #  alias enter mon_enter
 310 #  alias exit mon_exit
 311 # to access its methods more concisely.
 312 class Monitor
 313   include MonitorMixin
 314   alias try_enter try_mon_enter
 315   alias enter mon_enter
 316   alias exit mon_exit
 317 end
 318
 319
 320 # Documentation comments:
 321 #  - All documentation comes from Nutshell.
 322 #  - MonitorMixin.new_cond appears in the example, but is not documented in
 323 #    Nutshell.
 324 #  - All the internals (internal modules Accessible and Initializable, class
 325 #    ConditionVariable) appear in RDoc.  It might be good to hide them, by
 326 #    making them private, or marking them :nodoc:, etc.
 327 #  - The entire example from the RD section at the top is replicated in the RDoc
 328 #    comment for MonitorMixin.  Does the RD section need to remain?
 329 #  - RDoc doesn't recognise aliases, so we have mon_synchronize documented, but
 330 #    not synchronize.
 331 #  - mon_owner is in Nutshell, but appears as an accessor in a separate module
 332 #    here, so is hard/impossible to RDoc.  Some other useful accessors
 333 #    (mon_count and some queue stuff) are also in this module, and don't appear
 334 #    directly in the RDoc output.
 335 #  - in short, it may be worth changing the code layout in this file to make the
 336 #    documentation easier
 337
 338 # Local variables:
 339 # mode: Ruby
 340 # tab-width: 8
 341 # End: