Class: Fate::ProcessManager

Inherits:
Object
  • Object
show all
Defined in:
lib/fate/process_manager.rb

Overview

A process management tool, concerned primarily with spawning child processes, tracking them by name, and handling unexpected exits and signals.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(service, options = {}) ⇒ ProcessManager

Returns a new instance of ProcessManager.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/fate/process_manager.rb', line 10

def initialize(service, options={})
  @directory = options[:directory]
  @mutex = Mutex.new
  @service = service
  @output_handlers = @service.output_handlers
  @logger = @service.logger["Fate Manager"]

  @threads = {}
  @commands_by_name = {}
  @names_by_pid = {}
  @pids_by_name = {}
  at_exit do
    stop_all
  end
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



9
10
11
 
# File 'lib/fate/process_manager.rb', line 9

def logger
  @logger
end

#output_handlers ⇒ Object (readonly)

Returns the value of attribute output_handlers.



9
10
11
 
# File 'lib/fate/process_manager.rb', line 9

def output_handlers
  @output_handlers
end

Instance Method Details

#copy_stream(src, dest) ⇒ Object

Replacement for IO.copy_stream, which is a native function. We’ve been seeing segfaults when using it for lots of data.



126
127
128
129
130
131
132
133
 
# File 'lib/fate/process_manager.rb', line 126

def copy_stream(src, dest)
  begin
    while s = src.readpartial(1024)
      dest.write(s)
    end
  rescue EOFError
  end
end

#running ⇒ Object

list currently running commands



136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/fate/process_manager.rb', line 136

def running
  names = []
  @names_by_pid.each do |pid, name|
    begin
      # Signal 0 checks for the process, but sends no signal.
      Process.kill(0, pid)
      names << name
    rescue
    end
  end
  names.sort
end

#spawn(name, command) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/fate/process_manager.rb', line 80

def spawn(name, command)
  return Thread.new do

    pipe = nil
    pid = nil
    @mutex.synchronize do
      unless @down_in_flames
        if @directory
          Dir.chdir @directory do
            pipe = IO.popen(command, "r", :err => :out)
          end
        else
          pipe = IO.popen(command, "r", :err => :out)
        end
        pid = pipe.pid
        logger.info "Starting '#{name}' (pid #{pid})"

        @names_by_pid[pid] = name
        @pids_by_name[name] = pid
      end
    end

    unless @down_in_flames
      # Obtain an IO-ish object
      handler = output_handlers[name]
      # First line written to STDOUT is assumed to be the service
      # signalling that it is ready.
      line = pipe.gets
      @mutex.synchronize do
        unless @down_in_flames
          logger.info "#{name} is running."
          handler.write(line)
        end
      end
      @threads[name] = Thread.current

      copy_stream(pipe, handler)
      pid, status = Process.wait2(pid)
      handle_child_termination(pid, status)
    end

  end
end

#start_command(name, command) ⇒ Object 



48
49
50
51
52
53
54
 
# File 'lib/fate/process_manager.rb', line 48

def start_command(name, command)
  if pid = @pids_by_name[name]
    logger.warn "'#{name}' is already running with pid #{pid}"
  else
    spawn(name, command)
  end
end

#start_group(hash) ⇒ Object 



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

def start_group(hash)
  hash.each do |name, command|
    @commands_by_name[name] = command
    start_command(name, command) unless @down_in_flames
  end

  until hash.keys.all? { |key| @threads[key] }
    return false if @down_in_flames
    sleep 0.1
  end
  return true
end

#stop_all ⇒ Object 



26
27
28
29
30
31
32
33
 
# File 'lib/fate/process_manager.rb', line 26

def stop_all
  @mutex.synchronize do
    ordered = @service.stop_order(running)
    ordered.each do |name|
      term(name)
    end
  end
end

#stop_command(name) ⇒ Object 



56
57
58
 
# File 'lib/fate/process_manager.rb', line 56

def stop_command(name)
  term(name)
end

#term(name) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/fate/process_manager.rb', line 60

def term(name)
  if pid = @pids_by_name[name]
    @names_by_pid.delete(pid)
    @pids_by_name.delete(name)
    @threads.delete(name)
    system "kill -s TERM #{pid}"
    logger.info "Sent a kill signal to '#{name}' running at #{pid}"
    begin
      # Signal 0 checks for the process, but sends no signal.
      Process.kill(0, pid)
    rescue
      # TODO: limit number of retries, possibly issue kill -9?
      sleep 0.01
      retry
    end
  else
    logger.error "Could not find pid for '#{name}'"
  end
end