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_group(running)
  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

#service ⇒ Object (readonly)

Returns the value of attribute service.



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

def service
  @service
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.



164
165
166
167
168
169
170
171
 
# File 'lib/fate/process_manager.rb', line 164

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

#run(command_strings = []) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/fate/process_manager.rb', line 26

def run(command_strings=[])
  if command_strings.empty?
    run(@service.commands.keys)
  else
    commands = @service.resolve(command_strings)
    # don't need to start processes that are already running
    noop = commands & running
    to_start = commands - noop
    to_stop = running - commands

    stop_group(to_stop)
    start_group(to_start)
  end
end

#running ⇒ Object

list currently running commands



174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/fate/process_manager.rb', line 174

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 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/fate/process_manager.rb', line 110

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 = nil
      loop do
        line = pipe.gets
        if line =~ /debug/i
          handler.write(line)
        else
          break
        end
      end
      @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_strings = []) ⇒ Object 



41
42
43
44
45
46
47
48
 
# File 'lib/fate/process_manager.rb', line 41

def start(command_strings=[])
  if command_strings.empty?
    start_group(@service.commands.keys)
  else
    commands = @service.resolve(command_strings)
    start_group(commands)
  end
end

#start_command(name, command) ⇒ Object 



82
83
84
85
86
87
88
 
# File 'lib/fate/process_manager.rb', line 82

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(names) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/fate/process_manager.rb', line 59

def start_group(names)
  ordered = @service.start_order(names)
  ordered.each do |name|
    command = service.commands[name]
    @commands_by_name[name] = command
    start_command(name, command) unless @down_in_flames
  end
  until names.all? { |name| @threads[name] }
    return false if @down_in_flames
    sleep 0.1
  end
  return true
end

#stop(command_strings = []) ⇒ Object 



50
51
52
53
54
55
56
57
 
# File 'lib/fate/process_manager.rb', line 50

def stop(command_strings=[])
  if command_strings.empty?
    stop_group(running)
  else
    names = @service.resolve(command_strings)
    stop_group(names)
  end
end

#stop_group(names) ⇒ Object 



73
74
75
76
77
78
79
80
 
# File 'lib/fate/process_manager.rb', line 73

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

#term(name) ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/fate/process_manager.rb', line 90

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