lib/thwait.rb

   1 #
   2 #   thwait.rb - thread synchronization class
   3 #       $Release Version: 0.9 $
   4 #       $Revision: 1.3 $
   5 #       by Keiju ISHITSUKA(Nihpon Rational Software Co.,Ltd.)
   6 #
   7 # --
   8 #  feature:
   9 #  provides synchronization for multiple threads.
  10 #
  11 #  class methods:
  12 #  * ThreadsWait.all_waits(thread1,...)
  13 #    waits until all of specified threads are terminated.
  14 #    if a block is supplied for the method, evaluates it for
  15 #    each thread termination.
  16 #  * th = ThreadsWait.new(thread1,...)
  17 #    creates synchronization object, specifying thread(s) to wait.
  18 #
  19 #  methods:
  20 #  * th.threads
  21 #    list threads to be synchronized
  22 #  * th.empty?
  23 #    is there any thread to be synchronized.
  24 #  * th.finished?
  25 #    is there already terminated thread.
  26 #  * th.join(thread1,...)
  27 #    wait for specified thread(s).
  28 #  * th.join_nowait(threa1,...)
  29 #    specifies thread(s) to wait.  non-blocking.
  30 #  * th.next_wait
  31 #    waits until any of specified threads is terminated.
  32 #  * th.all_waits
  33 #    waits until all of specified threads are terminated.
  34 #    if a block is supplied for the method, evaluates it for
  35 #    each thread termination.
  36 #
  37
  38 require "thread.rb"
  39 require "e2mmap.rb"
  40
  41 #
  42 # This class watches for termination of multiple threads.  Basic functionality
  43 # (wait until specified threads have terminated) can be accessed through the
  44 # class method ThreadsWait::all_waits.  Finer control can be gained using
  45 # instance methods.
  46 #
  47 # Example:
  48 #
  49 #   ThreadsWait.all_wait(thr1, thr2, ...) do |t|
  50 #     STDERR.puts "Thread #{t} has terminated."
  51 #   end
  52 #
  53 class ThreadsWait
  54   RCS_ID='-$Id: thwait.rb,v 1.3 1998/06/26 03:19:34 keiju Exp keiju $-'
  55
  56   extend Exception2MessageMapper
  57   def_exception("ErrNoWaitingThread", "No threads for waiting.")
  58   def_exception("ErrNoFinishedThread", "No finished threads.")
  59
  60   #
  61   # Waits until all specified threads have terminated.  If a block is provided,
  62   # it is executed for each thread termination.
  63   #
  64   def ThreadsWait.all_waits(*threads) # :yield: thread
  65     tw = ThreadsWait.new(*threads)
  66     if block_given?
  67       tw.all_waits do |th|
  68         yield th
  69       end
  70     else
  71       tw.all_waits
  72     end
  73   end
  74
  75   #
  76   # Creates a ThreadsWait object, specifying the threads to wait on.
  77   # Non-blocking.
  78   #
  79   def initialize(*threads)
  80     @threads = []
  81     @wait_queue = Queue.new
  82     join_nowait(*threads) unless threads.empty?
  83   end
  84
  85   # Returns the array of threads in the wait queue.
  86   attr :threads
  87
  88   #
  89   # Returns +true+ if there are no threads to be synchronized.
  90   #
  91   def empty?
  92     @threads.empty?
  93   end
  94
  95   #
  96   # Returns +true+ if any thread has terminated.
  97   #
  98   def finished?
  99     !@wait_queue.empty?
 100   end
 101
 102   #
 103   # Waits for specified threads to terminate.
 104   #
 105   def join(*threads)
 106     join_nowait(*threads)
 107     next_wait
 108   end
 109
 110   #
 111   # Specifies the threads that this object will wait for, but does not actually
 112   # wait.
 113   #
 114   def join_nowait(*threads)
 115     threads.flatten!
 116     @threads.concat threads
 117     for th in threads
 118       Thread.start(th) do |t|
 119         begin
 120           t.join
 121         ensure
 122           @wait_queue.push t
 123         end
 124       end
 125     end
 126   end
 127
 128   #
 129   # Waits until any of the specified threads has terminated, and returns the one
 130   # that does.
 131   #
 132   # If there is no thread to wait, raises +ErrNoWaitingThread+.  If +nonblock+
 133   # is true, and there is no terminated thread, raises +ErrNoFinishedThread+.
 134   #
 135   def next_wait(nonblock = nil)
 136     ThreadsWait.fail ErrNoWaitingThread if @threads.empty?
 137     begin
 138       @threads.delete(th = @wait_queue.pop(nonblock))
 139       th
 140     rescue ThreadError
 141       ThreadsWait.fail ErrNoFinishedThread
 142     end
 143   end
 144
 145   #
 146   # Waits until all of the specified threads are terminated.  If a block is
 147   # supplied for the method, it is executed for each thread termination.
 148   #
 149   # Raises exceptions in the same manner as +next_wait+.
 150   #
 151   def all_waits
 152     until @threads.empty?
 153       th = next_wait
 154       yield th if block_given?
 155     end
 156   end
 157 end
 158
 159 ThWait = ThreadsWait
 160
 161
 162 # Documentation comments:
 163 #  - Source of documentation is evenly split between Nutshell, existing
 164 #    comments, and my own rephrasing.
 165 #  - I'm not particularly confident that the comments are all exactly correct.
 166 #  - The history, etc., up the top appears in the RDoc output.  Perhaps it would
 167 #    be better to direct that not to appear, and put something else there
 168 #    instead.