Class: Ductwork::Processes::ThreadSupervisor

Inherits:
Object
  • Object
show all
Defined in:
lib/ductwork/processes/thread_supervisor.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ThreadSupervisor

Returns a new instance of ThreadSupervisor.



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/ductwork/processes/thread_supervisor.rb', line 8

def initialize
  @running_context = Ductwork::RunningContext.new
  @workers = []

  adopt_or_create_process!
  run_hooks_for(:start)

  Signal.trap(:INT) { @running_context.shutdown! }
  Signal.trap(:TERM) { @running_context.shutdown! }
  Signal.trap(:TTIN) do
    Thread.list.each do |thread|
      puts thread.name
      if thread.backtrace
        puts thread.backtrace.join("\n")
      else
        puts "No backtrace to dump"
      end
      puts
    end
  end
end

Instance Attribute Details

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



6
7
8
 
# File 'lib/ductwork/processes/thread_supervisor.rb', line 6

def workers
  @workers
end

Instance Method Details

#add_worker(metadata: {}, &block) ⇒ Object

TODO: maybe change the whole supervisor interface because this is clunky



31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/ductwork/processes/thread_supervisor.rb', line 31

def add_worker(metadata: {}, &block)
  worker = block.call(metadata)
  workers << worker
  worker.start

  Ductwork.logger.debug(
    msg: "Started supervised thread",
    role: :thread_supervisor,
    thread: worker.name
  )
end

#run ⇒ Object 



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/ductwork/processes/thread_supervisor.rb', line 43

def run
  Ductwork.logger.debug(
    msg: "Entering main work loop",
    role: :thread_supervisor,
    pid: ::Process.pid
  )

  while running_context.running?
    sleep(Ductwork.configuration.supervisor_polling_timeout)
    check_workers_health
    report_heartbeat!
    reap_process_records
  end

  shutdown
end