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.



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 26

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
  @watchdog = nil
end

Instance Attribute Details

#concurrency ⇒ Object

Returns the value of attribute concurrency.



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

def concurrency
  @concurrency
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



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

def config
  @config
end

#fetcher ⇒ Object

Returns the value of attribute fetcher.



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

def fetcher
  @fetcher
end

#mode ⇒ Object (readonly)

Returns the value of attribute mode.



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

def mode
  @mode
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



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

def name
  @name
end

#queues ⇒ Object

Returns the value of attribute queues.



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

def queues
  @queues
end

#watchdog ⇒ Object (readonly)

Returns the value of attribute watchdog.



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

def watchdog
  @watchdog
end

#weights ⇒ Object (readonly)

Returns the value of attribute weights.



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

def weights
  @weights
end

Instance Method Details

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

Yields:

  • (chain) 


111
112
113
114
115
 
# File 'lib/wurk/capsule.rb', line 111

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.



175
176
177
 
# File 'lib/wurk/capsule.rb', line 175

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.



165
166
167
 
# File 'lib/wurk/capsule.rb', line 165

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.



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

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

#logger ⇒ Object 



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

def logger
  @config.logger
end

#lookup(name) ⇒ Object 



169
170
171
 
# File 'lib/wurk/capsule.rb', line 169

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.



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

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.



105
106
107
108
109
 
# File 'lib/wurk/capsule.rb', line 105

def prepare_shared!
  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.



70
71
72
 
# File 'lib/wurk/capsule.rb', line 70

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

#redis(idempotent: false) ⇒ Object 



159
160
161
 
# File 'lib/wurk/capsule.rb', line 159

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

#redis_pool ⇒ Object 



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

def redis_pool
  @pools[:main] ||= 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), 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.



154
155
156
157
 
# File 'lib/wurk/capsule.rb', line 154

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

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

Yields:

  • (chain) 


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

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.



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

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).



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

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