Class: Wurk::Manager

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/manager.rb

Overview

One per Capsule. Lives inside each forked child and owns the Processor pool. Replaces dead processors on the fly (replace-on-die), forwards the quiet/stop signals received by the Swarm to its processors, and ensures in-flight UnitsOfWork are bulk_requeued before threads are killed.

Lifecycle:

* `start`         — spawn each processor thread.
* `quiet`         — stop fetching; in-flight jobs run to completion.
* `stop(deadline)`— quiet + wait for drain; hard_shutdown on timeout.

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

Constant Summary collapse

PAUSE_TIME =

0.1 in TTY mode so interactive shutdown feels snappy; 0.5 in production so the supervisor isn't spinning while threads drain.

$stdout.tty? ? 0.1 : 0.5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(capsule, shutdown: nil) ⇒ Manager

shutdown: is the owning Launcher's process-wide shutdown request, invoked when concurrency can no longer be held (see #processor_result). Injected rather than reached for: only the entry point knows how this process exits — a swarm child / standalone CLI re-delivers TERM to itself so the installed trap drives the normal drain, while embedded mode owns no traps and must stop the launcher in place. A Manager built without an owner (Sidekiq's Manager.new(capsule) arity, kept for the alias) has no route to take, so it only reports.

Raises:

  • (ArgumentError) 


35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/wurk/manager.rb', line 35

def initialize(capsule, shutdown: nil)
  @config = @capsule = capsule
  @count = capsule.concurrency
  raise ArgumentError, "Concurrency of #{@count} is not supported" if @count < 1

  @done = false
  @shutdown = shutdown
  @workers = Set.new
  @plock = ::Mutex.new
  @count.times do
    @workers << Processor.new(@capsule, &method(:processor_result))
  end
end

Instance Attribute Details

#capsule ⇒ Object (readonly)

Returns the value of attribute capsule.



25
26
27
 
# File 'lib/wurk/manager.rb', line 25

def capsule
  @capsule
end

#config ⇒ Object (readonly) Originally defined in module Component

Returns the value of attribute config.

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



25
26
27
 
# File 'lib/wurk/manager.rb', line 25

def workers
  @workers
end

Instance Method Details

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Component

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Component

#hard_shutdown ⇒ Object

Reached when the deadline expired with workers still busy. Atomically move their in-flight UoWs private→public (Reliable#bulk_requeue) BEFORE raising Wurk::Shutdown into the threads, so a job killed mid-perform is re-run once (Sidekiq's at-least-once contract). job is read off another thread, so a Processor can ACK between this map and the requeue — but bulk_requeue's LREM guard skips the RPUSH on a miss, so a job that finished in that window is not resurrected onto the public queue.



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/wurk/manager.rb', line 133

def hard_shutdown # rubocop:disable Metrics/AbcSize
  cleanup = workers_snapshot

  if cleanup.any?
    jobs = cleanup.map(&:job).compact

    logger.warn { "Terminating #{cleanup.size} busy threads" }
    logger.debug { "Jobs still in progress #{jobs.inspect}" }

    # `&.` like #quiet: a TERM in the pre-prepare! window (traps install
    # before launcher.run) reaches here with no fetcher built yet.
    capsule.fetcher&.bulk_requeue(jobs)
  end

  cleanup.each(&:kill)

  # The caller typically `exit`s immediately after we return; give
  # threads a brief window to run their `ensure` blocks.
  deadline = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + 3
  wait_for(deadline) { workers_empty? }
end

#hostname ⇒ Object Originally defined in module Component

#identity ⇒ Object Originally defined in module Component

#leader? ⇒ Boolean Originally defined in module Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

Spec: docs/target/sidekiq-ent.md §6.1.

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Component

#process_nonce ⇒ Object Originally defined in module Component

#processor_result(processor, _reason = nil) ⇒ Object

Processor#run callback: invoked when a Processor thread exits, whether cleanly or via raised exception. Removes the dead processor from the pool and (unless we're already stopping) spawns a replacement so the capsule's concurrency stays constant. If the replacement itself can't be spawned, ask the owner to shut this process down (the swarm respawns it at full concurrency) rather than silently dropping a worker for the life of the process. Snapshot under @plock; start the replacement — a side effect — outside the lock.



102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/wurk/manager.rb', line 102

def processor_result(processor, _reason = nil)
  replacement = @plock.synchronize do
    @workers.delete(processor)
    unless @done
      p = Processor.new(@capsule, &method(:processor_result))
      @workers << p
      p
    end
  end
  replacement&.start
rescue StandardError => e
  # Replacement spawn failed (e.g. ThreadError at the OS thread limit).
  # Silently running one Processor short for the life of the process is
  # invisible degradation, so take the process down (the swarm respawns it
  # at full concurrency) — but through the owner's TERM path, never
  # `Thread.main.raise`: that raise unwound straight past Manager#stop, so
  # the in-flight UnitsOfWork were never bulk_requeued and each one waited
  # out a full reaper interval, and in embedded mode it killed the host's
  # main thread instead of just the worker. Non-blocking by contract — we
  # are on the dying Processor's own thread, which the drain will kill.
  @capsule.config.handle_exception(e, { context: 'Manager could not replace a dead Processor' })
  @shutdown&.call
end

#quiet ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/wurk/manager.rb', line 53

def quiet
  return if @done

  snapshot = @plock.synchronize do
    @done = true
    @workers.dup
  end
  # Halt fetching for the whole capsule up front: the shared fetcher's drain
  # flag makes retrieve_work return nil immediately, so a processor can't pull
  # a fresh job between quiet and its own terminate taking effect. Safe-nav
  # covers a quiet that lands before Capsule#prepare! materializes the fetcher
  # (e.g. a signal-driven quiet on a partially-booted launcher) — nothing is
  # fetching yet, so there is nothing to halt.
  capsule.fetcher&.terminate
  logger.info { "Terminating quiet threads for #{capsule.name} capsule" }
  snapshot.each(&:terminate)
end

#real_ms ⇒ Object Originally defined in module Component

--- clocks ---------------------------------------------------------

#redis(idempotent: false) ⇒ Object Originally defined in module Component

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#start ⇒ Object 



49
50
51
 
# File 'lib/wurk/manager.rb', line 49

def start
  workers_snapshot.each(&:start)
end

#stop(deadline) ⇒ Object

Graceful shutdown: quiet first, then poll for workers to clear. If the deadline elapses with workers still alive we fall through to hard_shutdown, which bulk_requeues their UoWs before killing threads.



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/wurk/manager.rb', line 74

def stop(deadline)
  quiet
  # Lifecycle hooks (e.g. :quiet) can be async; give them a tick to settle
  # before we start polling. Matches Sidekiq's PAUSE_TIME behavior.
  sleep PAUSE_TIME
  return if workers_empty?

  logger.info { 'Pausing to allow jobs to finish...' }
  wait_for(deadline) { workers_empty? }
  return if workers_empty?

  hard_shutdown
ensure
  capsule.stop
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


90
91
92
 
# File 'lib/wurk/manager.rb', line 90

def stopped?
  @done
end

#tid ⇒ Object Originally defined in module Component

#watchdog(last_words) ⇒ Object Originally defined in module Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.