Class: Farce::Abstract::Pool Abstract

Inherits:
Executor show all
Includes:
Internal::Inspect
Defined in:
lib/farce/abstract/pool.rb

Overview

This class is abstract.

Use Pool, Unshared::Pool, or Local::Pool.

A task scheduler with a dynamically sized pool of workers.

Pools accept work through #schedule. #close rejects new tasks and drains queued work. Concrete pools use either Ractor or Thread workers.

Direct Known Subclasses

Local::Pool, Pool, Unshared::Pool

Instance Method Summary collapse

Methods inherited from Executor

#close_all, #execute, #local?, #ractor_shareable?

Methods included from Internal::Noncopyable

#duplicable?

Constructor Details

#initialize(min_size: 0, max_size: 4, max_inflight: 64, grow_after: 0.005, shrink_after: 30, capacity: 1024, backend: Internal::FROZEN_CONFIG.io_backend, &constructor) ⇒ Pool

Create and start an elastic pool using the configured fiber scheduler. An explicit constructor block overrides the configured default.

max_inflight is a soft limit. A worker may exceed it when its scheduler has no runnable fibers or when the pool has reached max_size.

Parameters:

  • min_size (Integer) (defaults to: 0) —

    Workers started immediately and retained while open.

  • max_size (Integer) (defaults to: 4) —

    Maximum workers allowed.

  • max_inflight (Integer, nil) (defaults to: 64) —

    Soft task limit for each worker.

  • grow_after (Numeric) (defaults to: 0.005) —

    Queue wait before adding another worker.

  • shrink_after (Numeric, nil) (defaults to: 30) —

    Idle time before an extra worker retires.

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

    Pending-task capacity.

  • backend (Symbol) (defaults to: Internal::FROZEN_CONFIG.io_backend) —

    IO backend for the built-in scheduler.

Yield Returns:

  • (Object) —

    Fiber scheduler constructed in each worker.



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

def initialize(min_size: 0, max_size: 4, max_inflight: 64,
               grow_after: 0.005, shrink_after: 30, capacity: 1024,
               backend: Internal::FROZEN_CONFIG.io_backend, &constructor)
  constructor ||= Internal::FROZEN_CONFIG.fiber_scheduler_constructor
  initialize_configuration(min_size:, max_size:, max_inflight:, grow_after:, shrink_after:, backend:)
  @constructor  = prepare_constructor(constructor)
  @queue        = new_queue(capacity: capacity && Integer(capacity))
  @queue.interrupt(settings, settings.generation)
  @state        = Internal::Atom.new(:running)
  @error        = Internal::Atom.new
  @worker_count = Internal::Atom.new(0)
  @pressure     = Internal::Atom.new
  @admission    = Internal::Signal.new
  super()
  publish_pool
  begin
    settings.min_size.times { start_worker }
  rescue Exception # rubocop:disable Lint/RescueException
    close
    raise
  end
end

Instance Method Details

#at_max_size? ⇒ Boolean

Return whether all configured workers have been started.

Returns:

  • (Boolean) 


186
 
# File 'lib/farce/abstract/pool.rb', line 186

def at_max_size?(settings = self.settings) = size >= settings.max_size

#capacity ⇒ Integer?

Pending-task capacity. A smaller capacity preserves already queued tasks.

Returns:

  • (Integer, nil) 


73
 
# File 'lib/farce/abstract/pool.rb', line 73

def capacity = @queue.capacity

#capacity=(value) ⇒ BasicObject

Resize the pending-task queue. Nil makes the queue unbounded on CRuby. JRuby and TruffleRuby retain their initial bounded or unbounded choice.



77
78
79
 
# File 'lib/farce/abstract/pool.rb', line 77

def capacity=(value)
  @queue.capacity = value.nil? ? nil : Integer(value)
end

#close ⇒ Pool

Stop accepting tasks and drain queued work.

Returns:



150
151
152
153
154
155
156
157
158
159
 
# File 'lib/farce/abstract/pool.rb', line 150

def close
  return self if closed?
  @state.compare_and_set(:running, :closing)
  @queue.seal
  admission_changed
  if size.zero?
    @queue.closed? ? finish_close : start_worker(drain: true)
  end
  self
end

#closed? ⇒ Boolean

Return whether the pool rejects new tasks.

Returns:

  • (Boolean) 


174
 
# File 'lib/farce/abstract/pool.rb', line 174

def closed? = state != :running

#closing? ⇒ Boolean

Return whether the pool is draining queued work.

Returns:

  • (Boolean) 


177
 
# File 'lib/farce/abstract/pool.rb', line 177

def closing? = state == :closing

#error ⇒ BasicObject

Return the first worker error, if any.



162
163
164
165
 
# File 'lib/farce/abstract/pool.rb', line 162

def error
  value = @error.value
  value.is_a?(Envelope) ? value.value : value
end

#grow_after ⇒ Float

Returns Maximum queue wait before adding a worker.

Returns:

  • (Float) —

    Maximum queue wait before adding a worker.



54
 
# File 'lib/farce/abstract/pool.rb', line 54

def grow_after = settings.grow_after

#grow_after=(value) ⇒ BasicObject

Change the queue wait before adding a worker. Rechecks existing queued work.



104
105
106
107
108
 
# File 'lib/farce/abstract/pool.rb', line 104

def grow_after=(value)
  value = duration(value, :grow_after)
  update_settings(grow_after: value)
  value
end

#max_inflight ⇒ Integer?

Returns Soft task limit for each worker.

Returns:

  • (Integer, nil) —

    Soft task limit for each worker.



57
 
# File 'lib/farce/abstract/pool.rb', line 57

def max_inflight = settings.max_inflight

#max_inflight=(value) ⇒ BasicObject

Set the soft per-worker task limit. Nil disables the limit.



97
98
99
100
101
 
# File 'lib/farce/abstract/pool.rb', line 97

def max_inflight=(value)
  value &&= Integer(value)
  update_settings(max_inflight: value)
  value
end

#max_size ⇒ Integer

Returns Maximum worker count.

Returns:

  • (Integer) —

    Maximum worker count.



60
 
# File 'lib/farce/abstract/pool.rb', line 60

def max_size = settings.max_size

#max_size=(value) ⇒ BasicObject

Limit worker count. Excess workers retire when their running tasks leave them idle. Must be positive and at least min_size.



90
91
92
93
94
 
# File 'lib/farce/abstract/pool.rb', line 90

def max_size=(value)
  value = Integer(value)
  update_settings(max_size: value)
  value
end

#min_size ⇒ Integer

Returns Minimum worker count.

Returns:

  • (Integer) —

    Minimum worker count.



63
 
# File 'lib/farce/abstract/pool.rb', line 63

def min_size = settings.min_size

#min_size=(value) ⇒ BasicObject

Start and retain at least this many workers. Must not exceed max_size.



82
83
84
85
86
 
# File 'lib/farce/abstract/pool.rb', line 82

def min_size=(value)
  value = Integer(value)
  update_settings(min_size: value)
  value
end

#schedule(*args, mode: :smart, auto_local: true, timeout: nil) {|*args| ... } ⇒ self?

Enqueue a task for any worker in the pool.

Ractor pools reject local transfer. Thread pools preserve the block and arguments. With auto_local: false, they accept only local or smart modes. A timeout limits waiting for queue space. Expiry returns nil without enqueuing the task. Accepted tasks have no execution timeout.

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:

  • args (Array<Object>) —

    Positional arguments passed to the task block.

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

    Argument transfer mode.

  • auto_local (Boolean) (defaults to: true) —

    Whether to use local transfer in a thread pool.

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

    Maximum scheduling wait in seconds. Nil waits indefinitely, zero polls.

Yields:

  • (*args) —

    Task to execute.

Returns:

  • (self, nil) —

    self when submitted, or nil when the scheduling timeout expires.

Raises:



133
 
# File 'lib/farce/abstract/pool.rb', line 133

def schedule(...) = enqueue_task(false, ...)

#shrink_after ⇒ Float?

Returns Idle time before an extra worker retires.

Returns:

  • (Float, nil) —

    Idle time before an extra worker retires.



66
 
# File 'lib/farce/abstract/pool.rb', line 66

def shrink_after = settings.shrink_after

#shrink_after=(value) ⇒ BasicObject

Change the idle timeout. Nil disables idle retirement. Restarts idle waits.



111
112
113
114
115
 
# File 'lib/farce/abstract/pool.rb', line 111

def shrink_after=(value)
  value = duration(value, :shrink_after, nil_allowed: true)
  update_settings(shrink_after: value)
  value
end

#size ⇒ BasicObject

Return the current number of running and starting workers.



168
 
# File 'lib/farce/abstract/pool.rb', line 168

def size = @worker_count.value

#state ⇒ BasicObject

Return the current pool state.



171
 
# File 'lib/farce/abstract/pool.rb', line 171

def state = @state.value

#try_schedule(*args, mode: :smart, auto_local: true) ⇒ self?

Enqueues a task without waiting for queue space. Arguments and transfer modes follow #schedule.

Returns self when submitted, or nil when the queue is full.

Parameters:

  • args (Array) —

    Arguments to pass to the scheduled block. May be handled according to mode.

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

    The mode in which to schedule the task. Defaults to :smart. May be ignored by some executors.

  • auto_local (Boolean) (defaults to: true) —

    Whether to automatically use a local mode when possible. Defaults to true. May be ignored by some executors.

Returns:

  • (self, nil) —

    self when submitted, or nil when the queue is full.



143
144
145
 
# File 'lib/farce/abstract/pool.rb', line 143

def try_schedule(*, mode: :smart, auto_local: true, &)
  enqueue_task(true, *, mode:, auto_local:, &)
end