Class: RobotLab::Waiter Private

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

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Thread-safe waiter for blocking get operations on Memory.

Uses an IO.pipe pair instead of ConditionVariable so that IO.select integrates with the Async fiber scheduler hook. ConditionVariable#wait can block the event loop in Async contexts; IO.select yields to the scheduler correctly.

Multiple threads may wait on the same Waiter instance. signal() writes one byte per waiting thread so every blocked IO.select wakes exactly once.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Waiter

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Creates a new Waiter instance.



18
19
20
21
22
23
24
 
# File 'lib/robot_lab/waiter.rb', line 18

def initialize
  @read_io, @write_io = IO.pipe
  @mutex        = Mutex.new
  @value        = nil
  @signaled     = false
  @waiter_count = 0
end

Instance Method Details

#close ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Release the pipe file descriptors. Should be called after wait returns.



88
89
90
91
 
# File 'lib/robot_lab/waiter.rb', line 88

def close
  @read_io.close  rescue nil
  @write_io.close rescue nil
end

#signal(value) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Signal a value to all waiting threads.

Parameters:

  • value (Object) —

    the value to signal



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/robot_lab/waiter.rb', line 59

def signal(value)
  count = @mutex.synchronize do
    @value    = value
    @signaled = true
    @waiter_count
  end

  # Write one byte per waiting thread (min 1 to handle the race
  # where a thread passed the @signaled check but hasn't entered
  # IO.select yet — its IO.select will return immediately).
  bytes = [count, 1].max
  @write_io.write_nonblock("." * bytes) rescue nil
rescue IOError
  # pipe already closed — signal is a no-op
end

#signaled? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check if this waiter has been signaled.

Returns:

  • (Boolean) 


79
80
81
 
# File 'lib/robot_lab/waiter.rb', line 79

def signaled?
  @mutex.synchronize { @signaled }
end

#wait(timeout: nil) ⇒ Object, :timeout

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for a value to be signaled.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum seconds to wait (nil = indefinite)

Returns:

  • (Object, :timeout) —

    the signaled value, or :timeout if timed out



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/robot_lab/waiter.rb', line 31

def wait(timeout: nil)
  @mutex.synchronize do
    return @value if @signaled

    @waiter_count += 1
  end

  begin
    ready = IO.select([@read_io], nil, nil, timeout)

    @mutex.synchronize do
      @waiter_count -= 1
      return :timeout unless ready

      @read_io.read_nonblock(1) rescue nil  # drain one wake byte
      @value
    end
  rescue IOError
    @mutex.synchronize { @waiter_count -= 1 }
    :timeout
  end
end