Module: Clacky::ThreadRegistry

Defined in:
lib/clacky/thread_registry.rb

Overview

Process-wide registry of managed threads.

Every long-lived worker thread should be created through ThreadRegistry.spawn so shutdown can wait for it, force-stop it, and report any thread that refuses to die — including the backtrace of its creation site for tracing stray Thread.new calls.

Threads created with daemon: true are exempt from force_stop!; they are terminated automatically when the process exits.

Constant Summary collapse

MUTEX = 
Mutex.new
ENTRIES =

Thread => { name:, daemon:, killable:, created_at:, backtrace: }

{}

Class Method Summary collapse

Class Method Details

.force_stop!(grace: 5.0) ⇒ Object

Wait up to grace seconds for all killable threads to exit cooperatively, then Thread#kill the stragglers. Returns the threads that were killed.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/clacky/thread_registry.rb', line 65

def force_stop!(grace: 5.0)
  snapshot = MUTEX.synchronize do
    ENTRIES.keys.each { |t| ENTRIES.delete(t) unless t.alive? }
    ENTRIES.map { |t, meta| [t, meta.dup] }
  end

  killable = snapshot.select { |_t, meta| meta[:killable] }
  deadline = Time.now + grace
  while Time.now < deadline && killable.any? { |t, _meta| t.alive? }
    sleep 0.05
  end

  stragglers = killable.select { |t, _meta| t.alive? }.map(&:first)
  stragglers.each(&:kill)
  # Thread#kill is asynchronous — give stragglers a moment to run their
  # ensure (unregister) so report_leaks! doesn't flag threads that did exit.
  stragglers.each { |t| t.join(0.5) rescue nil }
  stragglers
end

.register(thread, name:, daemon: false, killable: true, backtrace: nil) ⇒ Object

Register an already-created thread (for code that cannot switch to spawn).



49
50
51
52
53
54
55
56
57
 
# File 'lib/clacky/thread_registry.rb', line 49

def register(thread, name:, daemon: false, killable: true, backtrace: nil)
  MUTEX.synchronize do
    ENTRIES[thread] = {
      name: name, daemon: daemon, killable: killable,
      created_at: Time.now, backtrace: backtrace
    }
  end
  thread
end

.report_leaks! ⇒ Object

Log WARN for every managed thread still alive (with its creation site).



86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/clacky/thread_registry.rb', line 86

def report_leaks!
  snapshot = MUTEX.synchronize do
    ENTRIES.keys.each { |t| ENTRIES.delete(t) unless t.alive? }
    ENTRIES.map { |t, meta| [t, meta.dup] }
  end

  snapshot.each do |_thread, meta|
    Clacky::Logger.warn(
      "thread leaked after shutdown",
      thread: meta[:name] || "unnamed",
      created_from: meta[:backtrace] ? meta[:backtrace].join(" | ") : "unknown"
    )
  end
end

.spawn(name:, daemon: false, killable: true, &block) ⇒ Thread

Create a managed thread.

Parameters:

  • name (String) —

    human-readable label for logs

  • daemon (Boolean) (defaults to: false) —

    true → exempt from force_stop! (dies with process)

  • killable (Boolean) (defaults to: true) —

    false → never Thread#kill'd by force_stop!

  • block (Proc) —

    thread body

Returns:

  • (Thread)

Raises:

  • (ArgumentError) 


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

def spawn(name:, daemon: false, killable: true, &block)
  raise ArgumentError, "block required" unless block

  creation_backtrace = caller(1, 5)
  thread = Thread.new do
    begin
      Thread.current.name = name if Thread.current.respond_to?(:name=)
      block.call
    rescue Clacky::AgentInterrupted
      # Cooperative shutdown signal (Ctrl+C / server drain). A managed
      # thread reaching its top with this exception is exiting by design,
      # not failing — swallow it so the thread terminates cleanly and
      # Ruby doesn't print a "terminated with exception" backtrace.
    ensure
      unregister(Thread.current)
    end
  end

  register(thread, name: name, daemon: daemon, killable: killable, backtrace: creation_backtrace)
  thread
end

.unregister(thread) ⇒ Object 



59
60
61
 
# File 'lib/clacky/thread_registry.rb', line 59

def unregister(thread)
  MUTEX.synchronize { ENTRIES.delete(thread) }
end