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.



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

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.



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

def concurrency
  @concurrency
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



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

def config
  @config
end

#fetcher ⇒ Object

Returns the value of attribute fetcher.



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

def fetcher
  @fetcher
end

#mode ⇒ Object (readonly)

Returns the value of attribute mode.



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

def mode
  @mode
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



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

def name
  @name
end

#queues ⇒ Object

Returns the value of attribute queues.



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

def queues
  @queues
end

#weights ⇒ Object (readonly)

Returns the value of attribute weights.



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

def weights
  @weights
end

Instance Method Details

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

Yields:

  • (chain) 


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

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.



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

def fetch_poll_interval
  @config[:fetch_poll_interval]
end

#fetch_redis(idempotent: false) ⇒ 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.



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

def fetch_redis(idempotent: false, &)
  PoolCheckout.with(fetch_redis_pool, idempotent, &)
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.



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

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

#logger ⇒ Object 



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

def logger
  @config.logger
end

#lookup(name) ⇒ Object 



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

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.



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

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.



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

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

#redis(idempotent: false) ⇒ Object 



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

def redis(idempotent: false, &)
  PoolCheckout.with(redis_pool, idempotent, &)
end

#redis_pool ⇒ Object 



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

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.



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

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) 


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

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.



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

def stop; end

#to_h ⇒ Object 



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

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