gistfile1.diff

diff --git common.mk common.mk
index dbbbe93..d12e882 100644
--- common.mk
+++ common.mk
@@ -471,7 +471,7 @@ test: test-sample btest-ruby test-knownbug
 
 test-all: $(TEST_RUNNABLE)-test-all
 yes-test-all: PHONY
-	$(RUNRUBY) "$(srcdir)/test/runner.rb" $(TESTS)
+	$(RUNRUBY) "$(srcdir)/test/runner.rb" $(TESTS) --ruby="$(RUNRUBY)"
 TESTS_BUILD = mkmf
 no-test-all: PHONY
 	$(MINIRUBY) -I"$(srcdir)/lib" "$(srcdir)/test/runner.rb" $(TESTS_BUILD)
diff --git lib/test/unit.rb lib/test/unit.rb
index fd50d81..66fd139 100644
--- lib/test/unit.rb
+++ lib/test/unit.rb
@@ -51,6 +51,11 @@ module Test
         non_options(args, options)
         @help = orig_args.map { |s| s =~ /[\s|&<>$()]/ ? s.inspect : s }.join " "
         @options = options
+        @opts = @options = options
+        if @options[:parallel]
+          @files = args 
+          @args = orig_args
+        end
       end
 
       private
@@ -75,9 +80,30 @@ module Test
         opts.on '-n', '--name PATTERN', "Filter test names on pattern." do |a|
           options[:filter] = a
         end
+ 
+        opts.on '--jobs-status', "Show status of jobs every file; Disabled when --jobs isn't specified." do
+          options[:job_status] = true
+        end
+
+        opts.on '-j N', '--jobs N', "Allow run tests with N jobs at once" do |a|
+          options[:parallel] = a.to_i
+        end
+
+        opts.on '--ruby VAL', "Path to ruby; It'll have used at -j option" do |a|
+          options[:ruby] = a
+        end
       end
 
       def non_options(files, options)
+        begin
+          require "rbconfig"
+        rescue LoadError
+          warn "#{caller(1)[0]}: warning: Parallel running disabled because can't get path to ruby; run specify with --ruby argument"
+          options[:parallel] = nil
+        else
+          options[:ruby] = RbConfig.ruby
+        end
+
         true
       end
     end
@@ -175,7 +201,7 @@ module Test
             $: << d
           end
           begin
-            require path
+            require path unless options[:parallel]
             result = true
           rescue LoadError
             puts "#{f}: #{$!}"
@@ -194,26 +220,192 @@ module Test
       include Test::Unit::RunCount
 
       class << self; undef autorun; end
+      @@stop_auto_run = false
       def self.autorun
         at_exit {
           Test::Unit::RunCount.run_once {
             exit(Test::Unit::Runner.new.run(ARGV) || true)
-          }
+          } unless @@stop_auto_run
         } unless @@installed_at_exit
         @@installed_at_exit = true
       end
 
+      def after_worker_down(worker, e=nil, c=1)
+        return unless @opts[:parallel]
+        return if @interrupt
+        after_worker_dead worker
+        if e
+          b = e.backtrace
+          warn "#{b.shift}: #{e.message} (#{e.class})"
+          STDERR.print b.map{|s| "\tfrom #{s}"}.join("\n")
+        end
+        @need_quit = true
+        warn ""
+        warn "Some worker was crashed. It seems ruby interpreter's bug"
+        warn "or, a bug of test/unit/parallel.rb. try again without -j"
+        warn "option."
+        warn ""
+        STDERR.flush
+        exit c
+      end
+
+      def jobs_status
+        puts "" unless @opts[:verbose]
+        if @opts[:job_status]
+          b = []
+          puts @workers.map { |x|
+            a = "#{x[:pid]}:#{x[:status].to_s.ljust(7)}"
+            b << (x[:file] ? x[:file].ljust(a.size)[0...a.size] : " "*a.size)
+            a
+          }.join(" ")
+          puts b.join(" ")
+        end
+      end
+
+      def after_worker_dead(worker)
+        return unless @opts[:parallel]
+        return if @interrupt
+        worker[:status] = :quit
+        worker[:in].close
+        worker[:out].close
+        @workers.delete(worker)
+        @dead_workers << worker
+        @ios = @workers.map{|w| w[:out] }
+      end
+
       def _run_suites suites, type
         @interrupt = nil
         result = []
-        suites.each {|suite|
+        if @opts[:parallel]
           begin
-            result << _run_suite(suite, type)
+            # Require needed things for parallel running
+            require 'thread'
+            require 'timeout'
+            @tasks = @files.dup # Array of filenames.
+            @need_quit = false
+            @dead_workers = []  # Array of dead workers.
+            shutting_down = false
+
+            # Array of workers.
+            @workers = @opts[:parallel].times.map do
+              i,o = IO.pipe("ASCII-8BIT") # worker o>|i> master
+              j,k = IO.pipe("ASCII-8BIT") # worker <j|<k master
+              k.sync = true
+              pid = spawn(*@opts[:ruby].split(/ /),File.dirname(__FILE__) +
+                          "/unit/parallel.rb", *@args, out: o, in: j)
+              [o,j].each{|io| io.close }
+              {in: k, out: i, pid: pid, status: :waiting}
+            end
+
+            # Thread: watchdog
+            watchdog = Thread.new do
+              while stat = Process.wait2
+                break if @interrupt # Break when interrupt
+                w = (@workers + @dead_workers).find{|x| stat[0] == x[:pid] }.dup
+                next unless w
+                unless w[:status] == :quit
+                  # Worker down
+                  after_worker_down w, nil, stat[1].to_i
+                end
+              end
+            end
+            @workers_hash = Hash[@workers.map {|w| [w[:out],w] }] # out-IO => worker
+            @ios = @workers.map{|w| w[:out] } # Array of worker IOs
+
+            while _io = IO.select(@ios)[0]
+              break unless _io.each do |io|
+                break if @need_quit
+                a = @workers_hash[io]
+                buf = ((a[:status] == :quit) ? io.read : io.gets).chomp
+                case buf
+                when /^okay$/ # Worker will run task
+                  a[:status] = :running
+                  jobs_status
+                when /^ready$/ # Worker is ready
+                  a[:status] = :ready
+                  if @tasks.empty?
+                    break unless @workers.find{|x| x[:status] == :running }
+                  else
+                    task = @tasks.shift
+                    a[:file] = File.basename(task)
+                    begin
+                      a[:loadpath] ||= []
+                      a[:in].puts "loadpath #{[Marshal.dump($:-a[:loadpath])].pack("m").gsub("\n","")}"
+                      a[:loadpath] = $:.dup
+                      a[:in].puts "run #{task} #{type}"
+                    rescue Errno::EPIPE
+                      after_worker_down a
+                    rescue IOError
+                      raise unless ["stream closed","closed stream"].include? $!.message
+                      after_worker_down a
+                    end
+                  end
+
+                  jobs_status
+                when /^done (.+?)$/ # Worker ran a one of suites in a file
+                  r = Marshal.load($1.unpack("m")[0])
+                  # [result,result,report,$:]
+                  result << r[0..1]
+                  report.push(*r[2])
+                  @errors += r[3][0]
+                  @failures += r[3][1]
+                  @skips += r[3][2]
+                  $:.push(*r[4]).uniq!
+                when /^p (.+?)$/ # Worker wanna print to STDOUT
+                  print $1.unpack("m")[0]
+                when /^bye (.+?)$/ # Worker will shutdown
+                  e = Marshal.load($1.unpack("m")[0])
+                  after_worker_down a, e
+                when /^bye$/ # Worker will shutdown
+                  if shutting_down
+                    after_worker_dead a
+                  else
+                    after_worker_down a
+                  end
+                end
+                break if @need_quit
+              end
+            end
           rescue Interrupt => e
             @interrupt = e
-            break
+            return result
+          ensure
+            shutting_down = true
+            watchdog.kill if watchdog
+            @workers.each do |w|
+              begin
+                timeout(1) do
+                  w[:in].puts "quit"
+                end
+              rescue Errno::EPIPE
+              rescue Timeout::Error
+              end
+              [:in,:out].each do |x|
+                w[x].close
+              end
+            end
+            begin
+              timeout(0.2*@workers.size) do
+                Process.waitall
+              end
+            rescue Timeout::Error
+              @workers.each do |w|
+                begin
+                  Process.kill(:KILL,w[:pid])
+                rescue Errno::ESRCH; end
+              end
+            end
           end
-        }
+        else
+          suites.each {|suite|
+            begin
+              result << _run_suite(suite, type)
+            rescue Interrupt => e
+              @interrupt = e
+              break
+            end
+          }
+        end
         result
       end
 
diff --git lib/test/unit/parallel.rb lib/test/unit/parallel.rb
new file mode 100644
index 0000000..16a65b7
--- /dev/null
+++ lib/test/unit/parallel.rb
@@ -0,0 +1,145 @@
+require 'test/unit'
+
+module Test                # :nodoc:
+  module Unit              # :nodoc:
+    class TestCase         # :nodoc:
+      class << self; alias orig_inherited inherited; end
+      def self.inherited x # :nodoc:
+        orig_inherited x
+        Test::Unit::Worker.suites << x
+      end
+    end
+  end
+end
+
+module Test
+  module Unit
+    class Worker < Runner
+      @@suites = []
+
+      class << self
+        def suites; @@suites; end
+        undef autorun
+      end
+      
+      alias orig_run_suite _run_suite
+      undef _run_suite
+      undef _run_suites
+
+      def _run_suites suites, type
+        suites.map do |suite|
+          result = _run_suite(suite, type)
+        end
+      end
+
+      def _run_suite(suite, type)
+        r = report.dup
+        orig_stdout = MiniTest::Unit.output
+        i,o = IO.pipe
+        MiniTest::Unit.output = o
+
+        stdout = STDOUT.dup
+
+        th = Thread.new(i.dup) do |io|
+          begin
+            while buf = (self.verbose ? io.gets : io.read(5))
+              stdout.puts "p #{[buf].pack("m").gsub("\n","")}"
+            end
+          rescue IOError
+          rescue Errno::EPIPE
+          end
+        end
+
+        e, f, s = @errors, @failures, @skips
+
+        result = orig_run_suite(suite, type)
+
+        MiniTest::Unit.output = orig_stdout
+
+        o.close
+        i.close
+
+        begin
+          th.join
+        rescue IOError
+          raise unless ["stream closed","closed stream"].include? $!.message
+        end
+
+        result << (report - r)
+        result << [@errors-e,@failures-f,@skips-s]
+        result << ($: - @old_loadpath)
+
+        begin
+          STDOUT.puts "done #{[Marshal.dump(result)].pack("m").gsub("\n","")}"
+        rescue Errno::EPIPE; end
+        return result
+      ensure
+        MiniTest::Unit.output = orig_stdout
+        o.close unless o.closed?
+        i.close unless i.closed?
+      end
+
+      def run(args = [])
+        process_args args
+        @@stop_auto_run = true
+        @opts = @options.dup
+
+        STDOUT.sync = true
+        STDOUT.puts "ready"
+        Signal.trap(:INT,"IGNORE")
+
+
+        @old_loadpath = []
+        begin
+          while buf = STDIN.gets
+            case buf.chomp
+            when /^loadpath (.+?)$/
+              @old_loadpath = $:.dup
+              $:.push(*Marshal.load($1.unpack("m")[0].force_encoding("ASCII-8BIT"))).uniq!
+            when /^run (.+?) (.+?)$/
+              puts "okay"
+
+              stdin = STDIN.dup
+              stdout = STDOUT.dup
+              th = Thread.new do
+                while puf = stdin.gets
+                  if puf.chomp == "quit"
+                    begin
+                      stdout.puts "bye"
+                    rescue Errno::EPIPE; end
+                    exit 
+                  end
+                end
+              end
+
+              @options = @opts.dup
+              @@suites = []
+              require $1
+              _run_suites @@suites, $2.to_sym
+
+              STDIN.reopen(stdin)
+              STDOUT.reopen(stdout)
+
+              th.kill
+              STDOUT.puts "ready"
+            when /^quit$/
+              begin
+                STDOUT.puts "bye"
+              rescue Errno::EPIPE; end
+              exit
+            end
+          end
+        rescue Exception => e
+          begin
+            STDOUT.puts "bye #{[Marshal.dump(e)].pack("m").gsub("\n","")}"
+          rescue Errno::EPIPE;end
+          exit
+        ensure
+          stdin.close
+        end
+      end
+    end
+  end
+end
+
+Test::Unit::Worker.new.run(ARGV)