Class: Celluloid::Condition

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

Overview

ConditionVariable-like signaling between tasks and threads

Defined Under Namespace

Classes: Waiter

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Condition

Returns a new instance of Condition.



30
31
32
33
 
# File 'lib/celluloid/condition.rb', line 30

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

Instance Method Details

#broadcast(value = nil) ⇒ Object

Broadcast a value to all waiting tasks and threads



79
80
81
82
83
84
 
# File 'lib/celluloid/condition.rb', line 79

def broadcast(value = nil)
  @mutex.synchronize do
    @waiters.each { |waiter| waiter << SignalConditionRequest.new(waiter.task, value) }
    @waiters.clear
  end
end

#signal(value = nil) ⇒ Object

Send a signal to the first task waiting on this condition



66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/celluloid/condition.rb', line 66

def signal(value = nil)
  @mutex.synchronize do
    if waiter = @waiters.shift
      waiter << SignalConditionRequest.new(waiter.task, value)
    else
      Internals::Logger.with_backtrace(caller(3)) do |logger|
        logger.debug("Celluloid::Condition signaled spuriously")
      end
    end
  end
end

#wait(timeout = nil) ⇒ Object

Wait for the given signal and return the associated value

Raises:



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/celluloid/condition.rb', line 36

def wait(timeout = nil)
  raise ConditionError, "cannot wait for signals while exclusive" if Celluloid.exclusive?

  if actor = Thread.current[:celluloid_actor]
    task = Task.current
    if timeout
      bt = caller
      timer = actor.timers.after(timeout) do
        exception = ConditionError.new("timeout after #{timeout.inspect} seconds")
        exception.set_backtrace bt
        task.resume exception
      end
    end
  else
    task = Thread.current
  end
  waiter = Waiter.new(self, task, Celluloid.mailbox, timeout)

  @mutex.synchronize do
    @waiters << waiter
  end

  result = Celluloid.suspend :condwait, waiter
  timer.cancel if timer
  raise result if result.is_a?(ConditionError)
  return yield(result) if block_given?
  result
end