From c5548a345ed9df05996d11b7e4ce79e314b4833b Mon Sep 17 00:00:00 2001 From: David Dollar Date: Thu, 8 Dec 2011 16:19:19 -0800 Subject: [PATCH] wip --- lib/foreman/engine.rb | 141 ++++++++++++++++------------------ lib/foreman/process.rb | 46 +++++++++-- lib/foreman/procfile.rb | 17 ++-- lib/foreman/procfile_entry.rb | 22 ++++++ 4 files changed, 138 insertions(+), 88 deletions(-) create mode 100644 lib/foreman/procfile_entry.rb diff --git a/lib/foreman/engine.rb b/lib/foreman/engine.rb index 9d794ee..991e2cc 100644 --- a/lib/foreman/engine.rb +++ b/lib/foreman/engine.rb @@ -23,6 +23,7 @@ class Foreman::Engine @directory = File.expand_path(File.dirname(procfile)) @options = options @environment = read_environment_files(options[:env]) + @output_mutex = Mutex.new end def self.load_env!(env_file) @@ -32,34 +33,17 @@ class Foreman::Engine def start proctitle "ruby: foreman master" - termtitle "#{File.basename(@directory)} - foreman (#{processes.size} processes)" - - processes.each do |process| - process.color = next_color - fork process - end + termtitle "#{File.basename(@directory)} - foreman" trap("TERM") { puts "SIGTERM received"; terminate_gracefully } trap("INT") { puts "SIGINT received"; terminate_gracefully } + assign_colors + spawn_processes + watch_for_output watch_for_termination end - def execute(name) - error "no such process: #{name}" unless process = procfile[name] - process.color = next_color - fork process - - trap("TERM") { puts "SIGTERM received"; terminate_gracefully } - trap("INT") { puts "SIGINT received"; terminate_gracefully } - - watch_for_termination - end - - def processes - procfile.processes - end - def port_for(process, num, base_port=nil) base_port ||= 5000 offset = procfile.process_names.index(process.name) * 100 @@ -68,60 +52,22 @@ class Foreman::Engine private ###################################################################### - def fork(process) + def spawn_processes concurrency = Foreman::Utils.parse_concurrency(@options[:concurrency]) - 1.upto(concurrency[process.name]) do |num| - fork_individual(process, num, port_for(process, num, @options[:port])) - end - end - - def fork_individual(process, num, port) - apply_environment! - - ENV["PORT"] = port.to_s - ENV["PS"] = "#{process.name}.#{num}" - - pid = Process.fork do - run(process) - end - - info "started with pid #{pid}", process - running_processes[pid] = process - end - - def run(process) - proctitle "ruby: foreman #{process.name}" - trap("SIGINT", "IGNORE") - - begin - Dir.chdir directory do - IO.popen(process.command, "w+") do |pipe| - trap("SIGTERM") do - Thread.new do - begin - Process.kill("SIGTERM", pipe.pid) - Timeout.timeout(3) { Process.waitall } - rescue Timeout::Error - Process.kill("SIGKILL", pipe.pid) - end - end - end - until pipe.eof? - info pipe.gets, process - end - end - end - rescue Interrupt, Errno::EIO, Errno::ENOENT - begin - info "process exiting", process - rescue Interrupt + procfile.entries.each do |entry| + reader, writer = IO.pipe + entry.spawn(concurrency[entry.name], writer, @directory, @environment).each do |process| + running_processes[process.pid] = process + readers[process] = reader end end end def kill_all(signal="SIGTERM") + p [:ff] running_processes.each do |pid, process| + p [:pid, pid] Process.kill(signal, pid) rescue Errno::ESRCH end end @@ -135,23 +81,53 @@ private ###################################################################### kill_all "SIGKILL" end + def watch_for_output + Thread.new do + begin + loop do + rs, ws = IO.select(readers.values, [], [], 1) + (rs || []).each do |r| + ps, message = r.gets.split(",", 2) + color = colors[ps.split(".").first] + info message, ps, color + end + end + rescue Exception => ex + puts ex.message + puts ex.backtrace + end + end + end + def watch_for_termination pid, status = Process.wait2 process = running_processes.delete(pid) - info "process terminated", process + info "process terminated", process.name terminate_gracefully kill_all rescue Errno::ECHILD end - def info(message, process=nil) - print process.color if process - print "#{Time.now.strftime("%H:%M:%S")} #{pad_process_name(process)} | " + def info(message, name="system", color=Term::ANSIColor.white) + print color + print "#{Time.now.strftime("%H:%M:%S")} #{pad_process_name(name)} | " print Term::ANSIColor.reset print message.chomp puts end + def print(message=nil) + @output_mutex.synchronize do + $stdout.print message + end + end + + def puts(message=nil) + @output_mutex.synchronize do + $stdout.puts message + end + end + def error(message) puts "ERROR: #{message}" exit 1 @@ -165,9 +141,8 @@ private ###################################################################### end end - def pad_process_name(process) - name = process ? "#{ENV["PS"]}" : "system" - name.ljust(longest_process_name + 3) # add 3 for process number padding + def pad_process_name(name="system") + name.to_s.ljust(longest_process_name + 3) # add 3 for process number padding end def proctitle(title) @@ -182,6 +157,24 @@ private ###################################################################### @running_processes ||= {} end + def readers + @readers ||= {} + end + + def colors + @colors ||= {} + end + + def assign_colors + procfile.entries.each do |entry| + colors[entry.name] = next_color + end + end + + def process_by_reader(reader) + readers.invert[reader] + end + def next_color @current_color ||= -1 @current_color += 1 diff --git a/lib/foreman/process.rb b/lib/foreman/process.rb index 939924f..8a3f6ab 100644 --- a/lib/foreman/process.rb +++ b/lib/foreman/process.rb @@ -2,13 +2,47 @@ require "foreman" class Foreman::Process - attr_reader :name - attr_reader :command - attr_accessor :color + attr_reader :entry + attr_reader :num + attr_reader :pid - def initialize(name, command) - @name = name - @command = command + def initialize(entry, num) + @entry = entry + @num = num + end + + def run(pipe, basedir, environment) + Dir.chdir(basedir) do + with_environment(environment) do + io = IO.popen("#{entry.command} 2>&1", "w+") + @pid = io.pid + output pipe, "started with pid %d" + Thread.new do + until io.eof? + output pipe, io.gets + end + end + end + end + end + + def name + "%s.%s" % [ entry.name, num ] + end + +private + + def output(pipe, message) + pipe.puts "%s,%s" % [ name, message ] + end + + def with_environment(environment) + old_env = ENV.each_pair.inject({}) { |h,(k,v)| h.update(k => v) } + environment.each { |k,v| ENV[k] = v } + ret = yield + ENV.clear + old_env.each { |k,v| ENV[k] = v} + ret end end diff --git a/lib/foreman/procfile.rb b/lib/foreman/procfile.rb index f169e4a..e630275 100644 --- a/lib/foreman/procfile.rb +++ b/lib/foreman/procfile.rb @@ -1,4 +1,5 @@ require "foreman" +require "foreman/procfile_entry" # A valid Procfile entry is captured by this regex. # All other lines are ignored. @@ -10,18 +11,18 @@ require "foreman" # class Foreman::Procfile - attr_reader :processes + attr_reader :entries def initialize(filename) - @processes = parse_procfile(filename) - end - - def process_names - processes.map(&:name) + @entries = parse_procfile(filename) end def [](name) - processes.detect { |process| process.name == name } + entries.detect { |entry| entry.name == name } + end + + def process_names + entries.map(&:name) end private @@ -29,7 +30,7 @@ private def parse_procfile(filename) File.read(filename).split("\n").map do |line| if line =~ /^([A-Za-z0-9_]+):\s*(.+)$/ - Foreman::Process.new($1, $2) + Foreman::ProcfileEntry.new($1, $2) end end.compact end diff --git a/lib/foreman/procfile_entry.rb b/lib/foreman/procfile_entry.rb new file mode 100644 index 0000000..a7fd4f1 --- /dev/null +++ b/lib/foreman/procfile_entry.rb @@ -0,0 +1,22 @@ +require "foreman" + +class Foreman::ProcfileEntry + + attr_reader :name + attr_reader :command + attr_accessor :color + + def initialize(name, command) + @name = name + @command = command + end + + def spawn(num, pipe, basedir, environment) + (1..num).to_a.map do |n| + process = Foreman::Process.new(self, n) + process.run(pipe, basedir, environment) + process + end + end + +end