Class: ConditionVariable

Inherits:
Object
  • Object
show all
Defined in:
lib/thread.rb

Overview

ConditionVariable objects augment class Mutex. Using condition variables, it is possible to suspend while in the middle of a critical section until a resource becomes available.

Example:

require 'thread'

mutex = Mutex.new
resource = ConditionVariable.new

a = Thread.new {
  mutex.synchronize {
    # Thread 'a' now needs the resource
    resource.wait(mutex)
    # 'a' can now have the resource
  }
}

b = Thread.new {
  mutex.synchronize {
    # Thread 'b' has finished using the resource
    resource.signal
  }
}

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ConditionVariable

Creates a new ConditionVariable



54
55
56
57
 
# File 'lib/thread.rb', line 54

def initialize
  @waiters = []
  @waiters_mutex = Mutex.new
end

Instance Method Details

#broadcast ⇒ Object

Wakes up all threads waiting for this lock.



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/thread.rb', line 96

def broadcast
  # TODO: incomplete
  waiters0 = nil
  @waiters_mutex.synchronize do
    waiters0 = @waiters.dup
    @waiters.clear
  end
  for t in waiters0
    begin
      t.run
    rescue ThreadError
    end
  end
  self
end

#signal ⇒ Object

Wakes up the first thread in line waiting for this lock.



83
84
85
86
87
88
89
90
91
 
# File 'lib/thread.rb', line 83

def signal
  begin
    t = @waiters_mutex.synchronize { @waiters.shift }
    t.run if t
  rescue ThreadError
    retry
  end
  self
end

#wait(mutex, timeout = nil) ⇒ Object

Releases the lock held in mutex and waits; reacquires the lock on wakeup.

If timeout is given, this method returns after timeout seconds passed, even if no other thread doesn't signal.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/thread.rb', line 65

def wait(mutex, timeout=nil)
  begin
    # TODO: mutex should not be used
    @waiters_mutex.synchronize do
      @waiters.push(Thread.current)
    end
    mutex.sleep timeout
  ensure
    @waiters_mutex.synchronize do
      @waiters.delete(Thread.current)
    end
  end
  self
end