Class: AtomicRuby::AtomicConditionVariable

Inherits:
Object
  • Object
show all
Defined in:
lib/atomic-ruby/atomic_condition_variable.rb,
sig/generated/atomic-ruby/atomic_condition_variable.rbs,
ext/atomic_ruby/atomic_ruby.c

Overview

Note:

This class is NOT Ractor-safe as it parks Thread references, which cannot be shared across ractors.

Provides lock-free wait/signal coordination using atomic operations.

AtomicConditionVariable lets one or more threads park until another thread signals them, without the paired Mutex that Ruby's ConditionVariable requires. The set of parked threads is tracked in a native doubly-linked list, so registering a waiter, removing a waiter, and shifting the head off for a signal are all O(1) and allocation-light. Parking still uses Thread.stop and Thread#wakeup, which are the standard kernel-level primitives Ruby exposes for sleeping a thread.

The lost-wakeup race in a naive check-then-park consumer is avoided by the #wait contract: a waiter registers itself before re-evaluating the predicate, so any signal that fires after the producer makes the predicate true is guaranteed to see the waiter and wake it. Ruby also remembers pending wakeups across Thread.stop, so a wakeup that arrives between the predicate check and the actual park is not lost.

Examples:

Basic usage

condvar = AtomicConditionVariable.new
ready = AtomicBoolean.new(false)

waiter = Thread.new do
  condvar.wait { ready.true? }
  puts "ready"
end

ready.make_true
condvar.signal

Worker loop draining an atomic queue

condvar.wait do
  work = queue.pop
  work || shutdown.true?
end

Defined Under Namespace

Classes: Waiter

Instance Method Summary collapse

Constructor Details

#initialize ⇒ AtomicConditionVariable

Creates a new condition variable with no parked threads.

Examples:

condvar = AtomicConditionVariable.new


52
53
54
 
# File 'lib/atomic-ruby/atomic_condition_variable.rb', line 52

def initialize
  _initialize
end

Instance Method Details

#broadcast ⇒ Integer

Wakes every parked waiter.

Each woken thread observes the wake the same way as with #signal.

Examples:

condvar = AtomicConditionVariable.new
condvar.broadcast #=> 0

Returns:

  • (Integer) —

    The number of waiters signalled



105
106
107
108
109
 
# File 'lib/atomic-ruby/atomic_condition_variable.rb', line 105

def broadcast
  threads = _drain_threads
  threads.each { |thread| thread.wakeup rescue nil }
  threads.size
end

#signal ⇒ true, false

Wakes one parked waiter, or no-ops if none are parked.

If a waiter has registered itself but is not yet inside Thread.stop, Ruby remembers the wakeup and the next Thread.stop returns immediately.

Examples:

condvar = AtomicConditionVariable.new
condvar.signal #=> false

Returns:

  • (true, false) —

    true if a waiter was signalled, false otherwise



86
87
88
89
90
91
92
 
# File 'lib/atomic-ruby/atomic_condition_variable.rb', line 86

def signal
  thread = _shift_thread
  return false unless thread

  thread.wakeup rescue nil
  true
end

#wait { ... } ⇒ untyped

Blocks until the given block returns a truthy value, then returns that value.

The block is evaluated optimistically first. If it returns truthy on that pass, no waiter is registered and the call returns immediately. Otherwise the calling thread registers itself, re-evaluates the block, and parks via Thread.stop until a #signal or #broadcast wakes it. The block may run more than once and may run concurrently with a signalling thread.

Examples:

Simple wait

condvar = AtomicConditionVariable.new
ready = AtomicBoolean.new(false)

Thread.new do
  sleep(1)
  ready.make_true
  condvar.signal
end

condvar.wait { ready.true? } #=> true

Yields:

Yield Returns:

  • (untyped) —

    Truthy to wake, falsy to keep waiting

Returns:

  • (untyped) —

    The first truthy value returned by the block



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/atomic-ruby/atomic_condition_variable.rb', line 137

def wait
  result = yield
  return result if result

  self_thread = Thread.current
  loop do
    waiter = _add_waiter(self_thread)
    result = yield
    if result
      _remove_waiter(waiter)
      return result
    end

    Thread.stop
    _remove_waiter(waiter)
  end
end

#waiter_count ⇒ Integer

Returns the number of currently parked waiters.

This operation is atomic and thread-safe. The returned value reflects the state at the time of the call, but may change immediately after in concurrent environments.

Examples:

condvar = AtomicConditionVariable.new
puts condvar.waiter_count #=> 0

Returns:

  • (Integer) —

    The number of currently parked waiters



69
70
71
 
# File 'lib/atomic-ruby/atomic_condition_variable.rb', line 69

def waiter_count
  _waiter_count
end