lib/unicorn/app/inetd.rb

   1 # Copyright (c) 2009 Eric Wong
   2 # You can redistribute it and/or modify it under the same terms as Ruby.
   3
   4 # this class *must* be used with Rack::Chunked
   5
   6 module Unicorn::App
   7   class Inetd < Struct.new(:cmd)
   8
   9     class CatBody < Struct.new(:errors, :err_rd, :out_rd, :pid_map)
  10       def initialize(env, cmd)
  11         self.errors = env['rack.errors']
  12         in_rd, in_wr = IO.pipe
  13         self.err_rd, err_wr = IO.pipe
  14         self.out_rd, out_wr = IO.pipe
  15
  16         cmd_pid = fork {
  17           inp, out, err = (0..2).map { |i| IO.new(i) }
  18           inp.reopen(in_rd)
  19           out.reopen(out_wr)
  20           err.reopen(err_wr)
  21           [ in_rd, in_wr, err_rd, err_wr, out_rd, out_wr ].each { |i| i.close }
  22           exec(*cmd)
  23         }
  24         [ in_rd, err_wr, out_wr ].each { |io| io.close }
  25         [ in_wr, err_rd, out_rd ].each { |io| io.binmode }
  26         in_wr.sync = true
  27
  28         # Unfortunately, input here must be processed inside a seperate
  29         # thread/process using blocking I/O since env['rack.input'] is not
  30         # IO.select-able and attempting to make it so would trip Rack::Lint
  31         inp_pid = fork {
  32           input = env['rack.input']
  33           [ err_rd, out_rd ].each { |io| io.close }
  34           buf = Unicorn::Z.dup
  35
  36           # this is dependent on input.read having readpartial semantics:
  37           while input.read(16384, buf)
  38             in_wr.write(buf)
  39           end
  40           in_wr.close
  41         }
  42         in_wr.close
  43         self.pid_map = {
  44           inp_pid => 'input streamer',
  45           cmd_pid => cmd.inspect,
  46         }
  47       end
  48
  49       def each(&block)
  50         begin
  51           rd, = IO.select([err_rd, out_rd])
  52           rd && rd.first or next
  53
  54           if rd.include?(err_rd)
  55             begin
  56               errors.write(err_rd.read_nonblock(16384))
  57             rescue Errno::EINTR
  58             rescue Errno::EAGAIN
  59               break
  60             end while true
  61           end
  62
  63           rd.include?(out_rd) or next
  64
  65           begin
  66             yield out_rd.read_nonblock(16384)
  67           rescue Errno::EINTR
  68           rescue Errno::EAGAIN
  69             break
  70           end while true
  71         rescue EOFError,Errno::EPIPE,Errno::EBADF,Errno::EINVAL
  72           break
  73         end while true
  74
  75         self
  76       end
  77
  78       def close
  79         pid_map.each { |pid, str|
  80           begin
  81             pid, status = Process.waitpid2(pid)
  82             status.success? or
  83               errors.write("#{str}: #{status.inspect} (PID:#{pid})\n")
  84           rescue Errno::ECHILD
  85             errors.write("Failed to reap #{str} (PID:#{pid})\n")
  86           end
  87         }
  88       end
  89
  90     end
  91
  92     def initialize(*cmd)
  93       self.cmd = cmd
  94     end
  95
  96     def call(env)
  97       /\A100-continue\z/i =~ env[Unicorn::Const::HTTP_EXPECT] and
  98           return [ 100, {} , [] ]
  99
 100       [ 200, { 'Content-Type' => 'application/octet-stream' },
 101        CatBody.new(env, cmd) ]
 102     end
 103
 104   end
 105
 106 end