Class: Farce::Concurrent::CyclicBarrier

Inherits:
Object
  • Object
show all
Includes:
Internal::Noncopyable, Shareable::Delegated
Defined in:
lib/farce/concurrent/cyclic_barrier.rb

Overview

A replacement for concurrent-ruby's CyclicBarrier implementation.

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Internal::Noncopyable

#duplicable?

Methods included from Shareable

#ractor_shareable?

Constructor Details

#initialize(parties) { ... } ⇒ CyclicBarrier

Returns a new instance of CyclicBarrier.

Parameters:

  • parties (Integer) —

    the number of parties required to trip the barrier.

Yields:

  • Optional block to trigger after the last waiter has arrived at the barrier but before they are released. Will be made Ractor-shareable. If it raises, the barrier breaks and the exception propagates to its caller.

Raises:

  • (ArgumentError) —

    if parties is not a positive integer.



45
46
47
48
49
50
51
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 45

def initialize(parties, &)
  raise ArgumentError, "parties must be a positive integer" unless parties.is_a?(Integer) && parties.positive?
  @parties    = parties
  @action     = block_given? ? Ractor.shareable_proc(&) : nil
  @generation = Strict::Atom.new(Generation.new)
  super()
end

Instance Attribute Details

#parties ⇒ Integer (readonly)

Returns the number of parties required to trip the barrier.

Returns:

  • (Integer) —

    the number of parties required to trip the barrier.



38
39
40
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 38

def parties
  @parties
end

Instance Method Details

#broken? ⇒ Boolean

Returns whether the barrier is in a broken state.

Returns:

  • (Boolean) —

    whether the barrier is in a broken state.



54
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 54

def broken? = generation.state == :broken

#number_waiting ⇒ Integer

Returns the number of parties currently waiting at the barrier.

Returns:

  • (Integer) —

    the number of parties currently waiting at the barrier.



57
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 57

def number_waiting = generation.number_waiting

#reset ⇒ nil

Breaks the current generation and starts a new one. Existing waiters return false.

Returns:

  • (nil) 


94
95
96
97
98
99
100
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 94

def reset
  update do |current|
    current.state = :broken
    Generation.new
  end
  nil
end

#wait(timeout = nil) ⇒ Boolean

Waits for all parties to reach the barrier. Successful generations can be reused immediately. A timeout breaks the barrier for all waiters until #reset is called.

Parameters:

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

    the maximum time to wait for the barrier.

Returns:

  • (Boolean) —

    true if the barrier was successfully reached, false if it was broken by a timeout or reset.



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/farce/concurrent/cyclic_barrier.rb', line 63

def wait(timeout = nil) # rubocop:disable Naming/PredicateMethod
  deadline = Internal.timeout_deadline(timeout)
  waiting  = nil
  update do |current|
    waiting = current
    next current unless current.state == :waiting
    current.waiting.increment
    next current if current.waiting.value < @parties

    completed = false
    begin
      @action&.call
      completed = true
    ensure
      current.state = completed ? :fulfilled : :broken
    end
    Generation.new
  end

  unless waiting.wait(Clock.remaining(deadline))
    update do |current|
      # A trip or reset may have won the race with this waiter's timeout.
      waiting.state = :broken if waiting.state == :waiting
      current
    end
  end
  waiting.state == :fulfilled
end