Class: Farce::TimerQueue

Inherits:
Abstract::TimerQueue show all
Includes:
Shareable::Unfreezable, Internal::ManagedQueue
Defined in:
lib/farce/timer_queue.rb

Overview

A shareable timer queue with transfer modes for unshareable values.

Instance Attribute Summary

Attributes inherited from Abstract::Queue

#capacity, #mode, #num_waiting, #size

Instance Method Summary collapse

Methods included from Shareable

#ractor_shareable?

Methods inherited from Abstract::TimerQueue

#delete, #first_timestamp, #last_timestamp, #overdue?, #overdue_by, #wait_pop

Methods inherited from Abstract::Queue

#age_tracking?, #clear, #close, #closed?, #deq, #duplicable?, #empty?, #enq, #full?, #generation, #length, #max, #oldest_age, #oldest_enqueued_at, #seal, #sealed?, #wait_pop, #wait_push

Constructor Details

#initialize(capacity: nil, mode: :smart, track_age: false) ⇒ BasicObject

Valid modes are:

  • :smart - Applies smart rules or a dedicated proxy. Raises for unsupported values. This is the default for most mode-backed classes.
  • :smart_raise - Applies smart rules or a dedicated proxy, with a raising fallback.
  • :smart_copy - Applies smart rules or a dedicated proxy, then falls back to copying or Marshal. This is the default for Port.
  • :smart_move - Applies smart rules or a dedicated proxy, then falls back to moving.
  • :copy - The value will be copied between Ractors.
  • :marshal - The value will be serialized immediately and reconstructed once per Ractor with Ruby's Marshal protocol. Contents must support Marshal. Nested shareable values may be reconstructed too.
  • :make_shareable - The value will be made Ractor-shareable using Ractor.make_shareable.
  • :move - The value will be moved between Ractors. This saves memory compared to copying, and supports values that can't be copied but moved (like IO objects). However, the value will no longer be accessible on the Ractor that pushed it.
  • :mutable - A Mutable instance will be created for the value. This isn't done recursively and thus will fail for nested unshareable values.
  • :local - The value will be kept local to the Ractor that pushed it. Another ractor trying to receive it will get an error. Useful for usage contained within a single Ractor.
  • :proxy - The value will be wrapped in a Farce::Proxy that executes calls in the original Ractor.
  • :raise - An error will be raised if the value is not Ractor-shareable. Useful for enforcing shareability.
  • :dedup - The value will be deduplicated using Farce.dedup, then made Ractor-shareable. This may update and freeze the original. Already-shareable values pass through unchanged.
  • :shareable_copy - The value will be copied and the copy will be made Ractor-shareable.

Parameters:

  • capacity (Integer, nil) (defaults to: nil) —

    the maximum number of values, or nil for an unbounded queue

  • mode (Symbol) (defaults to: :smart) —

    the transfer mode for unshareable values



14
15
16
17
 
# File 'lib/farce/timer_queue.rb', line 14

def initialize(capacity: nil, mode: :smart, track_age: false)
  @manager = ModeManager.new(mode:)
  super(capacity:, track_age:)
end

Instance Method Details

#peek { ... } ⇒ BasicObject?

Note:

Peeking at a moved value claims it for this Ractor while leaving it queued.

Return the earliest value without removing it, whether or not it is due.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



63
64
65
66
67
68
 
# File 'lib/farce/timer_queue.rb', line 63

def peek
  empty = false
  value = @queue.peek { empty = true }
  return @manager.unwrap(value) unless empty
  yield if block_given?
end

#pop(non_block = false, timeout: nil) { ... } ⇒ BasicObject?

Remove the earliest value, waiting until its timestamp is reached.

Parameters:

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

    maximum number of seconds to wait

Yields:

  • called when the timeout expires first

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/farce/timer_queue.rb', line 41

def pop(non_block = false, timeout: nil) # rubocop:disable Style/OptionalBooleanParameter
  return try_pop { raise ThreadError, "queue empty" } if non_block
  deadline = timeout_at(timeout) unless timeout.nil?

  while true
    empty = false
    value = @queue.pop_before(Clock.now) { empty = true }
    return @manager.unwrap(value) unless empty
    return block_given? ? yield : nil if UNDEFINED.equal?(wait_for_timestamp(deadline))
  end
end

#push(value, non_block = false, mode: nil, timeout: nil, **time_options) ⇒ Entry, false

Add a value to become available at the given time. Valid modes are:

  • :smart - Applies smart rules or a dedicated proxy. Raises for unsupported values. This is the default for most mode-backed classes.
  • :smart_raise - Applies smart rules or a dedicated proxy, with a raising fallback.
  • :smart_copy - Applies smart rules or a dedicated proxy, then falls back to copying or Marshal. This is the default for Port.
  • :smart_move - Applies smart rules or a dedicated proxy, then falls back to moving.
  • :copy - The value will be copied between Ractors.
  • :marshal - The value will be serialized immediately and reconstructed once per Ractor with Ruby's Marshal protocol. Contents must support Marshal. Nested shareable values may be reconstructed too.
  • :make_shareable - The value will be made Ractor-shareable using Ractor.make_shareable.
  • :move - The value will be moved between Ractors. This saves memory compared to copying, and supports values that can't be copied but moved (like IO objects). However, the value will no longer be accessible on the Ractor that pushed it.
  • :mutable - A Mutable instance will be created for the value. This isn't done recursively and thus will fail for nested unshareable values.
  • :local - The value will be kept local to the Ractor that pushed it. Another ractor trying to receive it will get an error. Useful for usage contained within a single Ractor.
  • :proxy - The value will be wrapped in a Farce::Proxy that executes calls in the original Ractor.
  • :raise - An error will be raised if the value is not Ractor-shareable. Useful for enforcing shareability.
  • :dedup - The value will be deduplicated using Farce.dedup, then made Ractor-shareable. This may update and freeze the original. Already-shareable values pass through unchanged.
  • :shareable_copy - The value will be copied and the copy will be made Ractor-shareable.

Parameters:

  • value (BasicObject) —

    the value to add

  • non_block (Boolean) (defaults to: false) —

    whether to raise an exception when the queue is at capacity

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

    maximum number of seconds to wait for capacity

  • time_options (Hash{Symbol => Object}) —

    a scheduling option accepted by Clock.parse: at:, time:, timeout_at:, delay:, in:, offset:, wait:, or clock:. With no scheduling option, the value is available immediately. Since timeout: controls the capacity wait, use delay: or wait: to schedule a relative offset.

  • mode (Symbol) (defaults to: nil) —

    the transfer mode for unshareable values

Returns:

  • (Entry, false) —

    the cancellation handle, or false if the capacity wait expires

Raises:

  • (ThreadError) —

    when the queue is at capacity and non_block is true



22
23
24
25
26
27
 
# File 'lib/farce/timer_queue.rb', line 22

def push(value, non_block = false, mode: nil, timeout: nil, **time_options) # rubocop:disable Style/OptionalBooleanParameter
  at    = Clock.parse(time_options)
  value = @manager.wrap(value, mode:)
  entry = Entry.new(value, at, self)
  push_to_storage(at, non_block, value, timeout:) && entry
end

#try_pop { ... } ⇒ BasicObject?

Try to remove the earliest value only if its timestamp has been reached.

Yields:

  • called when no value is ready

Returns:

  • (BasicObject, nil) —

    the value or the fallback result



54
55
56
57
58
59
 
# File 'lib/farce/timer_queue.rb', line 54

def try_pop
  empty = false
  value = @queue.pop_before(Clock.now) { empty = true }
  return @manager.unwrap(value) unless empty
  yield if block_given?
end

#try_push(value, mode: nil, **time_options) { ... } ⇒ Entry, BasicObject

Try to add a value without waiting for capacity. Valid modes are:

  • :smart - Applies smart rules or a dedicated proxy. Raises for unsupported values. This is the default for most mode-backed classes.
  • :smart_raise - Applies smart rules or a dedicated proxy, with a raising fallback.
  • :smart_copy - Applies smart rules or a dedicated proxy, then falls back to copying or Marshal. This is the default for Port.
  • :smart_move - Applies smart rules or a dedicated proxy, then falls back to moving.
  • :copy - The value will be copied between Ractors.
  • :marshal - The value will be serialized immediately and reconstructed once per Ractor with Ruby's Marshal protocol. Contents must support Marshal. Nested shareable values may be reconstructed too.
  • :make_shareable - The value will be made Ractor-shareable using Ractor.make_shareable.
  • :move - The value will be moved between Ractors. This saves memory compared to copying, and supports values that can't be copied but moved (like IO objects). However, the value will no longer be accessible on the Ractor that pushed it.
  • :mutable - A Mutable instance will be created for the value. This isn't done recursively and thus will fail for nested unshareable values.
  • :local - The value will be kept local to the Ractor that pushed it. Another ractor trying to receive it will get an error. Useful for usage contained within a single Ractor.
  • :proxy - The value will be wrapped in a Farce::Proxy that executes calls in the original Ractor.
  • :raise - An error will be raised if the value is not Ractor-shareable. Useful for enforcing shareability.
  • :dedup - The value will be deduplicated using Farce.dedup, then made Ractor-shareable. This may update and freeze the original. Already-shareable values pass through unchanged.
  • :shareable_copy - The value will be copied and the copy will be made Ractor-shareable.

Parameters:

  • value (BasicObject) —

    the value to add

  • time_options (Hash{Symbol => Object}) —

    a scheduling option accepted by Clock.parse: at:, time:, timeout_at:, delay:, in:, offset:, timeout:, wait:, or clock:. With no scheduling option, the value is available immediately.

  • mode (Symbol) (defaults to: nil) —

    the transfer mode for unshareable values

Yields:

  • called when the queue is at capacity

Returns:

  • (Entry, BasicObject) —

    the cancellation handle, or the fallback result when full



32
33
34
35
36
37
38
 
# File 'lib/farce/timer_queue.rb', line 32

def try_push(value, mode: nil, **time_options)
  at    = Clock.parse(time_options)
  value = @manager.wrap(value, mode:)
  entry = Entry.new(value, at, self)
  return entry if @queue.push(at, value)
  block_given? ? yield : false
end