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

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.



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/wurk/capsule.rb', line 24

def initialize(name, config)
  @name = name.to_s
  @config = config
  @concurrency = config[:concurrency] || 5
  @queues = ['default']
  @mode = :strict
  @weights = { 'default' => 0 }
  @fetcher = nil
  # One mutable Hash rather than two ivars: the pools are the only part of a
  # capsule that legitimately changes after Configuration#freeze! — fork
  # closes them, Launcher#stop releases them, an embedded host that boots
  # again rebuilds them. `Object#freeze` is shallow, so the Hash stays
  # writable and freezing a capsule keeps meaning "no more configuration"
  # instead of "these sockets are yours forever". Before this, a reset on a
  # frozen capsule disconnected the pool and then raised FrozenError on the
  # memo, leaving `redis_pool` answering with a shut-down pool for good.
  @pools = {}
  @client_chain = nil
  @server_chain = nil
  @server_middleware_seen = []
  @prepared = false
  @watchdog = 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

#watchdog ⇒ Object (readonly)

Returns the value of attribute watchdog.



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

def watchdog
  @watchdog
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) 


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

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.



219
220
221
 
# File 'lib/wurk/capsule.rb', line 219

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.



209
210
211
 
# File 'lib/wurk/capsule.rb', line 209

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.



179
180
181
 
# File 'lib/wurk/capsule.rb', line 179

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

#logger ⇒ Object 



223
224
225
 
# File 'lib/wurk/capsule.rb', line 223

def logger
  @config.logger
end

#lookup(name) ⇒ Object 



213
214
215
 
# File 'lib/wurk/capsule.rb', line 213

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.



82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/wurk/capsule.rb', line 82

def prepare!
  prepare_shared!
  @fetcher ||= build_fetcher
  # Here rather than in the constructor so a client-only process — a web
  # dyno building the same Configuration — allocates nothing, and so the
  # scanner thread a bounded job spawns belongs to the child that runs it.
  # The object itself holds no thread until something is bounded.
  @watchdog ||= Watchdog.new(self)
  redis_pool
  fetch_redis_pool
  self
end

#prepare_shared! ⇒ Object

The half of prepare! a forking parent can run on every child's behalf: the chains are a pure function of this capsule's identity, not of the slot (queues + concurrency) a swarm child is assigned later, and copy_for opens nothing. Run before the fork, the entries are allocated once and inherited copy-on-write instead of rebuilt in every child.

The rest of prepare! deliberately stays post-fork: both pools are sized off the slot's concurrency and one built here would hand every child an inherited socket, and build_fetcher fires the host's config[:fetch_setup] hook, which is per-child — running it in the parent would let a custom fetcher snapshot the wrong queues, or leak whatever the hook opened across the fork. Idempotent.



107
108
109
110
111
112
113
114
 
# File 'lib/wurk/capsule.rb', line 107

def prepare_shared!
  client_middleware
  refresh_server_middleware!
  # A second boot (a launcher run twice) finds the capsule frozen — the
  # flag is already set, so it is not written again.
  @prepared ||= true
  self
end

#prepared? ⇒ Boolean

True once this capsule has been prepared to run jobs — prepare_shared! in the swarm parent before the fork, prepare! in the child at boot. Only then is the memoized chain the one the workers run, and only then is it final: a middleware registered on the Configuration afterwards can never reach it.

Returns:

  • (Boolean) 


121
122
123
 
# File 'lib/wurk/capsule.rb', line 121

def prepared?
  @prepared
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.



72
73
74
 
# File 'lib/wurk/capsule.rb', line 72

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

#redis(idempotent: false) ⇒ Object 



203
204
205
 
# File 'lib/wurk/capsule.rb', line 203

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

#redis_pool ⇒ Object 



171
172
173
 
# File 'lib/wurk/capsule.rb', line 171

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

#redis_pools_built? ⇒ Boolean

Returns:

  • (Boolean) 


199
200
201
 
# File 'lib/wurk/capsule.rb', line 199

def redis_pools_built?
  !@pools.empty?
end

#reload_redis_pools! ⇒ Object

The fork-path form: closes idle sockets but keeps every pool usable, for a parent whose own threads may be mid-checkout.



195
196
197
 
# File 'lib/wurk/capsule.rb', line 195

def reload_redis_pools!
  @pools.each_value(&:reload!)
end

#reset_redis_pools! ⇒ Object

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



188
189
190
191
 
# File 'lib/wurk/capsule.rb', line 188

def reset_redis_pools!
  @pools.each_value(&:disconnect!)
  @pools.clear
end

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

Yields:

  • (chain) 


147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/wurk/capsule.rb', line 147

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 ||= begin
    config_chain = @config.server_middleware
    @server_middleware_seen = config_chain.entries.map(&:klass)
    config_chain.copy_for(self)
  end)
  yield chain if block_given?
  chain
end

#server_middleware_missing?(klass) ⇒ Boolean

True when this capsule's server middleware chain has already been snapshotted and that snapshot does not carry klass.

The snapshot is the copy server_middleware memoized; a middleware registered on the Configuration after that copy was taken can never reach this capsule — which is what Wurk::Encryption.enable has to detect before it seals arguments the workers could not open.

Pure read: it never materializes the chain and has no side effects.

Returns:

  • (Boolean) 


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

def server_middleware_missing?(klass)
  chain = @server_chain
  return false unless chain

  !chain.exists?(klass)
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.



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

def stop; end

#thread_priority ⇒ Object

Capsule-hosted components (Manager, Processor, Fetcher) hand their capsule to Component as config, and safe_thread reads the priority off it. Sidekiq delegates the same accessor (capsule.rb:30).



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

def thread_priority = @config.thread_priority

#to_h ⇒ Object 



48
49
50
 
# File 'lib/wurk/capsule.rb', line 48

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