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

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(capsule) ⇒ Manager

Returns a new instance of Manager.

Raises:

  • (ArgumentError) 


27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/wurk/manager.rb', line 27

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

  @done = false
  @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

#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

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



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/wurk/manager.rb', line 118

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

#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, crash the child (the swarm respawns it) rather than silently dropping concurrency. Snapshot under @plock; start the replacement — a side effect — outside the lock.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/wurk/manager.rb', line 92

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; instead report and crash the child on the main
  # thread so the swarm respawns it at full concurrency (plan 02 §6).
  @capsule.config.handle_exception(e, { context: 'Manager could not replace a dead Processor' })
  main_thread.raise(e)
end

#quiet ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/wurk/manager.rb', line 44

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

#start ⇒ Object 



40
41
42
 
# File 'lib/wurk/manager.rb', line 40

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.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/wurk/manager.rb', line 65

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) 


81
82
83
 
# File 'lib/wurk/manager.rb', line 81

def stopped?
  @done
end