lib/unicorn/app/inetd.rb

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