Class: Wurk::Capsule

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/capsule.rb

Overview

One processing unit: a set of threads + queues sharing a fetcher and a Redis pool. Configurations can hold many capsules; each maps to its own Manager and Processors.

Spec: docs/target/sidekiq-free.md §5 (Sidekiq::Capsule).

Constant Summary collapse

MODES = 
%i[strict weighted random].freeze
POOL_HEADROOM =

Headroom above concurrency for the main pool; the whole pool is then floored at MIN_POOL_SIZE. Blocking BLMOVE fetch has its own pool (#fetch_redis_pool), so the main pool serves only the background loops — heartbeat, scheduled poller, leader election, cron, the two metrics rollups, reaper, history, health probe — plus the host's own job-code checkouts. concurrency + 5 (floor 10) gives each an unstarvable slot; the old concurrency + 2 starved them once fetch also drew from here — the #101 0/N pool-exhaustion incident. Override via config.redis[:size].

5
MIN_POOL_SIZE = 
10

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, config) ⇒ Capsule

Returns a new instance of Capsule.



19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/wurk/capsule.rb', line 19

def initialize(name, config)
  @name = name.to_s
  @config = config
  @concurrency = config[:concurrency] || 5
  @queues = ['default']
  @mode = :strict
  @weights = { 'default' => 0 }
  @fetcher = nil
  @redis_pool = nil
  @fetch_redis_pool = nil
  @client_chain = nil
  @server_chain = nil
end

Instance Attribute Details

#concurrency ⇒ Object

Returns the value of attribute concurrency.



17
18
19
 
# File 'lib/wurk/capsule.rb', line 17

def concurrency
  @concurrency
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



16
17
18
 
# File 'lib/wurk/capsule.rb', line 16

def config
  @config
end

#fetcher ⇒ Object

Returns the value of attribute fetcher.



17
18
19
 
# File 'lib/wurk/capsule.rb', line 17

def fetcher
  @fetcher
end

#mode ⇒ Object (readonly)

Returns the value of attribute mode.



16
17
18
 
# File 'lib/wurk/capsule.rb', line 16

def mode
  @mode
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



16
17
18
 
# File 'lib/wurk/capsule.rb', line 16

def name
  @name
end

#queues ⇒ Object

Returns the value of attribute queues.



16
17
18
 
# File 'lib/wurk/capsule.rb', line 16

def queues
  @queues
end

#weights ⇒ Object (readonly)

Returns the value of attribute weights.



16
17
18
 
# File 'lib/wurk/capsule.rb', line 16

def weights
  @weights
end

Instance Method Details

#client_middleware {|chain| ... } ⇒ Object

Yields:

  • (chain) 


74
75
76
77
78
 
# File 'lib/wurk/capsule.rb', line 74

def client_middleware
  chain = (@client_chain ||= @config.client_middleware.copy_for(self))
  yield chain if block_given?
  chain
end

#fetch_poll_interval ⇒ Object

Empty-poll BLMOVE backoff for this capsule's reliable fetcher (Pro super_fetch §3.3). nil → the fetcher falls back to its TIMEOUT default.



139
140
141
 
# File 'lib/wurk/capsule.rb', line 139

def fetch_poll_interval
  @config[:fetch_poll_interval]
end

#fetch_redis ⇒ Object

Checkout from the dedicated fetch pool. Only the reliable fetcher's blocking BLMOVE uses this, so a parked fetch never holds a main-pool slot.



129
130
131
 
# File 'lib/wurk/capsule.rb', line 129

def fetch_redis(&)
  fetch_redis_pool.with(&)
end

#fetch_redis_pool ⇒ Object

Dedicated pool for the reliable fetcher's blocking BLMOVE: one slot per processor thread (concurrency), since at most that many threads park in fetch at once. Keeping fetch off the main pool is what lets an idle worker hold zero main-pool connections again.



108
109
110
 
# File 'lib/wurk/capsule.rb', line 108

def fetch_redis_pool
  @fetch_redis_pool ||= build_pool(size: @concurrency, name: "#{@name}-fetch")
end

#logger ⇒ Object 



143
144
145
 
# File 'lib/wurk/capsule.rb', line 143

def logger
  @config.logger
end

#lookup(name) ⇒ Object 



133
134
135
 
# File 'lib/wurk/capsule.rb', line 133

def lookup(name)
  @config.lookup(name)
end

#prepare! ⇒ Object

Materialize everything that lazy-inits via ||= and default the fetcher, BEFORE Configuration#freeze! freezes the capsule — otherwise the first post-freeze access (a fetch tick, a middleware call) hits a nil fetcher or FrozenErrors building a pool. The swarm's ChildBoot used to do this by hand; centralizing it here covers the standalone CLI and embedded paths too (the bug behind a nil fetcher in exe/wurk). Idempotent.



65
66
67
68
69
70
71
72
 
# File 'lib/wurk/capsule.rb', line 65

def prepare!
  @fetcher ||= build_fetcher
  redis_pool
  fetch_redis_pool
  client_middleware
  server_middleware
  self
end

#queue_specs ⇒ Object

Lossless name[,weight] specs (unlike queues, which is the weight-expanded list). Lets Configuration#topology rebuild a slot that round-trips back through queues= without flattening weights.



55
56
57
 
# File 'lib/wurk/capsule.rb', line 55

def queue_specs
  @weights.map { |q, w| w.positive? ? "#{q},#{w}" : q }
end

#redis ⇒ Object 



123
124
125
 
# File 'lib/wurk/capsule.rb', line 123

def redis(&)
  redis_pool.with(&)
end

#redis_pool ⇒ Object 



100
101
102
 
# File 'lib/wurk/capsule.rb', line 100

def redis_pool
  @redis_pool ||= build_pool(size: main_pool_size, name: "#{@name}-main")
end

#reset_redis_pools! ⇒ Object

Disconnect and drop cached pools. Called by Wurk::Swarm just before fork (parent side: close inherited sockets) and just after fork (child side: rebuild lazily). Connection_pool#shutdown is terminal, so dropping the reference is required — redis_pool will rebuild.



116
117
118
119
120
121
 
# File 'lib/wurk/capsule.rb', line 116

def reset_redis_pools!
  @redis_pool&.disconnect!
  @redis_pool = nil
  @fetch_redis_pool&.disconnect!
  @fetch_redis_pool = nil
end

#server_middleware {|chain| ... } ⇒ Object

Yields:

  • (chain) 


80
81
82
83
84
85
86
87
 
# File 'lib/wurk/capsule.rb', line 80

def server_middleware
  # copy_for(self) — not dup — binds the chain's `@config` to this capsule,
  # so middleware that reach for `redis_pool`/`redis`/`logger` resolve them
  # instead of hitting `nil` (a plain dup leaves @config nil).
  chain = (@server_chain ||= @config.server_middleware.copy_for(self))
  yield chain if block_given?
  chain
end

#stop ⇒ Object

No-op in OSS. Reserved for Pro/Ent hooks that need to flush per-capsule state on shutdown. Manager#stop invokes this in ensure so the contract is honored regardless of how stop unwinds.

Spec: docs/target/sidekiq-free.md §5.



152
 
# File 'lib/wurk/capsule.rb', line 152

def stop; end

#to_h ⇒ Object 



33
34
35
 
# File 'lib/wurk/capsule.rb', line 33

def to_h
  { concurrency: @concurrency, mode: @mode, weights: @weights }
end