Class: Farce::Local::Pool

Inherits:
Abstract::Pool show all
Includes:
Scoped, Shareable::Unfreezable, Internal::LocalExecutorMethods
Defined in:
lib/farce/local/pool.rb

Overview

A shareable handle with an independent thread pool in each scope.

Each pool uses the scaling settings and fiber scheduler configured at construction. Changing settings tunes the current scope's pool. New scopes use the constructor settings. Tasks always keep their block and argument references. Only local and smart modes are accepted when auto_local is false. A scheduler constructor must be convertible to a shareable proc for use in each Ractor. #close drains only the current scope's pool. #close_all drains every pool across all scopes and Ractors, and prevents new pools from starting.

Examples:

Keep task state within each Ractor

pool    = Farce::Local::Pool.new(max_size: 2)
results = Farce::Unshared::Queue.new
pool.schedule { results.push(:done) }
results.pop # => :done
pool.close_all

Instance Attribute Summary

Attributes included from Scoped

#scope

Instance Method Summary collapse

Methods included from Shareable

#ractor_shareable?

Methods inherited from Abstract::Pool

#at_max_size?, #closing?, #grow_after, #max_inflight, #max_size, #min_size, #schedule, #shrink_after, #try_schedule

Methods inherited from Abstract::Executor

#execute, #local?, #ractor_shareable?, #schedule, #try_schedule

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, scope: :ractor, backend: Internal::FROZEN_CONFIG.io_backend, &constructor) ⇒ BasicObject

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.

Valid scopes are:

  • :ractor - The value is shared by all threads within the same Ractor.
  • :thread_group - The value is shared by all threads within the same thread group.
  • :thread - The value is shared by all fibers within the same thread.
  • :fiber_storage - The value is shared by all fibers using the same storage. Unless explicitly specified, storage is inherited from the parent fiber for blocking fibers, but not for non-blocking fibers (such as those created by Scheduler#schedule).
  • :fiber - Each fiber has its own independent value.

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.

  • scope (Symbol) (defaults to: :ractor) —

    The scope of the pool. Defaults to :ractor.

Yield Returns:

  • (Object) —

    Fiber scheduler constructed in each worker.



31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/farce/local/pool.rb', line 31

def initialize(min_size: 0, max_size: 4, max_inflight: 64,
               grow_after: 0.005, shrink_after: 30, capacity: 1024, scope: :ractor,
               backend: Internal::FROZEN_CONFIG.io_backend, &constructor)
  initialize_configuration(min_size:, max_size:, max_inflight:, grow_after:, shrink_after:, backend:)
  @capacity           = capacity && Integer(capacity)
  @constructor        = prepare_constructor(constructor || Internal::FROZEN_CONFIG.fiber_scheduler_constructor)
  @pools              = Strict::Map.new
  @closed_all         = Internal::Flag.new(false)
  @initialization_key = Object.new.freeze
  super(scope:, min_size: @min_size, max_size: @max_size, max_inflight: @max_inflight,
        grow_after: @grow_after, shrink_after: @shrink_after, capacity:, backend:)
end

Instance Method Details

#capacity ⇒ BasicObject 



48
 
# File 'lib/farce/local/pool.rb', line 48

def capacity = (pool = current_pool) ? pool.capacity : @capacity

#capacity=(value) ⇒ BasicObject 



70
71
72
 
# File 'lib/farce/local/pool.rb', line 70

def capacity=(value)
  scoped_value.capacity = value
end

#close ⇒ self

Stop accepting work in this scope and drain its queued tasks.

Returns:

  • (self) 


76
77
78
79
80
 
# File 'lib/farce/local/pool.rb', line 76

def close
  pool = @closed_all.value ? Internal::Storage[self, scope:] : scoped_value
  pool&.close
  self
end

#close_all ⇒ self

Stop accepting work in every scope and Ractor and request draining. Future scopes also reject submissions. Does not wait for running tasks.

Returns:

  • (self) 


85
86
87
88
89
90
91
92
93
94
 
# File 'lib/farce/local/pool.rb', line 85

def close_all
  return self unless @closed_all.compare_and_set(false, true)
  @pools.each_pair do |pool, queue|
    queue.push([pool, :close].freeze, clock: 0)
  rescue ClosedQueueError
    # The owning Ractor may already have shut down its supervisor.
    nil
  end
  close
end

#closed? ⇒ Boolean

Returns Whether this scope rejects new tasks.

Returns:

  • (Boolean) —

    Whether this scope rejects new tasks.



106
 
# File 'lib/farce/local/pool.rb', line 106

def closed? = @closed_all.value || super

#error ⇒ Exception?

Returns The first worker error in this scope.

Returns:

  • (Exception, nil) —

    The first worker error in this scope.



103
 
# File 'lib/farce/local/pool.rb', line 103

def error = current_pool&.error

#grow_after=(value) ⇒ BasicObject 



62
63
64
 
# File 'lib/farce/local/pool.rb', line 62

def grow_after=(value)
  scoped_value.grow_after = value
end

#max_inflight=(value) ⇒ BasicObject 



58
59
60
 
# File 'lib/farce/local/pool.rb', line 58

def max_inflight=(value)
  scoped_value.max_inflight = value
end

#max_size=(value) ⇒ BasicObject 



54
55
56
 
# File 'lib/farce/local/pool.rb', line 54

def max_size=(value)
  scoped_value.max_size = value
end

#min_size=(value) ⇒ BasicObject 



50
51
52
 
# File 'lib/farce/local/pool.rb', line 50

def min_size=(value)
  scoped_value.min_size = value
end

#shrink_after=(value) ⇒ BasicObject 



66
67
68
 
# File 'lib/farce/local/pool.rb', line 66

def shrink_after=(value)
  scoped_value.shrink_after = value
end

#size ⇒ Integer

Returns Running and starting workers in this scope.

Returns:

  • (Integer) —

    Running and starting workers in this scope.



97
 
# File 'lib/farce/local/pool.rb', line 97

def size = current_pool&.size || 0

#state ⇒ Symbol

Returns The current scope's pool state.

Returns:

  • (Symbol) —

    The current scope's pool state.



100
 
# File 'lib/farce/local/pool.rb', line 100

def state = current_pool&.state || :closed