* lib/test/unit.rb: Add new options; --jobs,-j,--ruby,--jobs-status,

--no-retry.
  [Feature #4415] [ruby-dev:43226],[ruby-dev:43222],[ruby-core:35294]
* lib/test/unit/parallel.rb: Used at test/unit --jobs(-j) option.
* test/csv/test_serialization.rb: test/unit parallel running ready.
* test/rake/test_file_task.rb: test/unit parallel running ready.

git-svn-id: svn+ssh://ci.ruby-lang.org/ruby/trunk@30939 b2dd03c8-39d4-4d8f-98ff-823fe69b080e

ChangeLog

@@ -1,3 +1,12 @@
+Tue Feb 22 12:27:26 2011  Shota Fukumori (sora_h) <sorah@tubusu.net>
+
+	* lib/test/unit.rb: Add new options; --jobs,-j,--ruby,--jobs-status,
+	  --no-retry.
+		[Feature #4415] [ruby-dev:43226],[ruby-dev:43222],[ruby-core:35294]
+	* lib/test/unit/parallel.rb: Used at test/unit --jobs(-j) option.
+	* test/csv/test_serialization.rb: test/unit parallel running ready.
+	* test/rake/test_file_task.rb: test/unit parallel running ready.
+
 2011-02-21  Eric Hodel  <drbrain@segment7.net>
 
 	* ext/syslog/syslog.c: Apply documentation patch from mathew murphy.

NEWS

@@ -113,6 +113,13 @@ with all sufficient information, see the ChangeLog file.
   * extended method:
     * StringIO#set_encoding can get 2nd argument and optional hash.
 
+* test/unit
+  * New arguments:
+    * -j N, --jobs=N: Allow run N testcases at once.
+    * --jobs-status: Show status of jobs when parallel running.
+    * --no-retry: Don't retry testcases which failed when parallel running.
+    * --ruby=RUBY: path to ruby for job(worker) process. optional.
+
 * uri
   * new methods:
     * URI::Generic#hostname

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,35 @@ module Test
         opts.on '-n', '--name PATTERN', "Filter test names on pattern." do |a|
           options[:filter] = a
         end
+ 
+        opts.on '--jobs-status [TYPE]', "Show status of jobs every file; Disabled when --jobs isn't specified." do |type|
+          options[:job_status] = true
+          options[:job_status_type] = type.to_sym if type
+        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 '--no-retry', "Don't retry running testcase when --jobs specified" do
+          options[:no_retry] = true
+        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 +206,7 @@ module Test
             $: << d
           end
           begin
-            require path
+            require path unless options[:parallel]
             result = true
           rescue LoadError
             puts "#{f}: #{$!}"
@@ -186,32 +217,299 @@ module Test
     end
 
     class Runner < MiniTest::Unit
+      include Test::Unit::Options
+      include Test::Unit::RequireFiles
       include Test::Unit::GlobOption
       include Test::Unit::LoadPathOption
       include Test::Unit::GCStressOption
       include Test::Unit::RunCount
 
       class << self; undef autorun; end
+
+      alias orig_run_anything _run_anything
+      undef _run_anything
+
+      def _run_anything type
+        if @opts[:parallel] && @warnings
+          warn ""
+          ary = []
+          @warnings.reject! do |w|
+            r = ary.include?(w[1].message)
+            ary << w[1].message
+            r
+          end
+          @warnings.each do |w|
+            warn "#{w[0]}: #{w[1].message} (#{w[1].class})"
+          end
+          warn ""
+        end
+        orig_run_anything(type)
+      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
+        return unless @opts[:job_status]
+        puts "" unless @opts[:verbose]
+        if @opts[:job_status]
+          line2 = []
+          line1 = @workers.map { |worker|
+            a = "#{worker[:pid]}:#{worker[:status].to_s.ljust(7)}"
+            if worker[:file]
+              if @opts[:job_status_type] == :replace
+                a = "#{worker[:pid]}=#{worker[:file]}"
+              else
+                if a.size > worker[:file].size
+                  line2 << worker[:file].ljust(a.size)
+                else
+                  a << " "*(worker[:file].size-a.size)
+                  line2 << worker[:file]
+                end
+              end
+            else
+              line2 << " "*a.size
+            end
+            a
+          }.join(" ")
+          if @opts[:job_status_type] == :replace
+            @terminal_width ||= %x{stty size 2>/dev/null}.split[1].to_i.nonzero? \
+                            ||  %x{tput cols 2>/dev/null}.to_i.nonzero? \
+                            ||  80
+            @jstr_size ||= 0
+            del_jobs_status
+            STDOUT.flush
+            print line1[0...@terminal_width]
+            STDOUT.flush
+            @jstr_size = line1.size > @terminal_width ? @terminal_width : line1.size
+          else
+            puts line1
+            puts line2.join(" ")
+          end
+        end
+      end
+
+      def del_jobs_status
+        return unless @opts[:job_status_type] == :replace && @jstr_size
+        print "\r"+" "*@jstr_size+"\r"
+      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.
+            @warnings = []
+            shutting_down = false
+            errors = []
+            failures = []
+            skips = []
+            rep = []
+
+            # 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
+                worker = @workers_hash[io]
+                buf = ((worker[:status] == :quit) ? io.read : io.gets).chomp
+                case buf
+                when /^okay$/ # Worker will run task
+                  worker[:status] = :running
+                  jobs_status
+                when /^ready$/ # Worker is ready
+                  worker[:status] = :ready
+                  if @tasks.empty?
+                    break unless @workers.find{|x| x[:status] == :running }
+                  else
+                    task = @tasks.shift
+                    worker[:file] = File.basename(task).gsub(/\.rb/,"")
+                    worker[:real_file] = task
+                    begin
+                      worker[:loadpath] ||= []
+                      worker[:in].puts "loadpath #{[Marshal.dump($:-worker[:loadpath])].pack("m").gsub("\n","")}"
+                      worker[:loadpath] = $:.dup
+                      worker[:in].puts "run #{task} #{type}"
+                      worker[:status] = :prepare
+                    rescue Errno::EPIPE
+                      after_worker_down worker 
+                    rescue IOError
+                      raise unless ["stream closed","closed stream"].include? $!.message
+                      after_worker_down worker
+                    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]
+                  rep << {file: worker[:real_file], report: r[2], result: r[3],
+                          testcase: r[5]}
+                  errors << [worker[:real_file],r[5],r[3][0]]
+                  failures << [worker[:real_file],r[5],r[3][1]]
+                  skips << [worker[:real_file],r[5],r[3][2]]
+                  $:.push(*r[4]).uniq!
+                  worker[:status] = :done
+                  jobs_status if @opts[:job_status_type] == :replace
+                  worker[:status] = :running
+                when /^p (.+?)$/ # Worker wanna print to STDOUT
+                  del_jobs_status
+                  print $1.unpack("m")[0]
+                  jobs_status if @opts[:job_status_type] == :replace
+                when /^after (.+?)$/
+                  @warnings << Marshal.load($1.unpack("m")[0])
+                when /^bye (.+?)$/ # Worker will shutdown
+                  e = Marshal.load($1.unpack("m")[0])
+                  after_worker_down worker, e
+                when /^bye$/ # Worker will shutdown
+                  if shutting_down
+                    after_worker_dead worker
+                  else
+                    after_worker_down worker
+                  end
+                end
+                break if @need_quit
+              end
+            end
+
+            # Retry
+            # TODO: Interrupt?
           rescue Interrupt => e
             @interrupt = e
-            break
+            return result
+          ensure
+            shutting_down = true
+
+            watchdog.kill if watchdog
+            @workers.each do |worker|
+              begin
+                timeout(1) do
+                  worker[:in].puts "quit"
+                end
+              rescue Errno::EPIPE
+              rescue Timeout::Error
+              end
+              [:in,:out].each do |name|
+                worker[name].close
+              end
+            end
+            begin
+              timeout(0.2*@workers.size) do
+                Process.waitall
+              end
+            rescue Timeout::Error
+              @workers.each do |worker|
+                begin
+                  Process.kill(:KILL,worker[:pid])
+                rescue Errno::ESRCH; end
+              end
+            end
+
+            unless @need_quit
+              if @interrupt || @opts[:no_retry]
+                rep.each do |r|
+                  report.push(*r[:report])
+                end
+                @errors += errors.map(&:last).inject(:+)
+                @failures += failures.map(&:last).inject(:+)
+                @skips += skips.map(&:last).inject(:+)
+              else
+                puts ""
+                puts "Retrying..."
+                puts ""
+                @options = @opts
+                rep.each do |r|
+                  if r[:testcase] && r[:file] && !r[:report].empty?
+                    require r[:file]
+                    _run_suite(eval(r[:testcase]),type)
+                  else
+                    report.push(*r[:report])
+                    @errors += r[:result][0]
+                    @failures += r[:result][1]
+                    @skips += r[:result][1]
+                  end
+                end
+              end
+            end
           end
-        }
+        else
+          suites.each {|suite|
+            begin
+              result << _run_suite(suite, type)
+            rescue Interrupt => e
+              @interrupt = e
+              break
+            end
+          }
+        end
         result
       end
 
@@ -223,10 +521,6 @@ module Test
     end
 
     class AutoRunner
-      class Runner < Test::Unit::Runner
-        include Test::Unit::RequireFiles
-      end
-
       attr_accessor :to_run, :options
 
       def initialize(force_standalone = false, default_dir = nil, argv = ARGV)

lib/test/unit/parallel.rb

@@ -0,0 +1,139 @@
+require 'test/unit'
+
+module Test
+  module Unit
+    class Worker < Runner
+      class << self
+        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)
+        result << suite.name
+
+        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 if o && !o.closed?
+        i.close if i && !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
+          stdin = STDIN.dup
+          stdout = STDOUT.dup
+          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 (.+?) (.+?)$/
+              STDOUT.puts "okay"
+
+              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 = MiniTest::Unit::TestCase.test_suites
+
+              begin
+                require $1
+              rescue LoadError
+                th.kill
+                STDOUT.puts "after #{[Marshal.dump([$1, $!])].pack("m").gsub("\n","")}"
+                STDOUT.puts "ready"
+                next
+              end
+              _run_suites MiniTest::Unit::TestCase.test_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)

test/csv/test_serialization.rb

@@ -131,7 +131,7 @@ class TestCSV::Serialization < TestCSV
   def test_io
     test_class_dump
 
-    data_file = File.join(File.dirname(__FILE__), "temp_test_data.csv")
+    data_file = File.join(File.dirname(__FILE__), "serialization_test_data.csv")
     CSV.dump(@names, File.open(data_file, "wb"))
 
     assert(File.exist?(data_file))

test/rake/test_file_task.rb

@@ -29,7 +29,9 @@ class Rake::TestFileTask < Test::Unit::TestCase
   end
 
   def test_file_times_new_depends_on_old
-    create_timed_files(OLDFILE, NEWFILE)
+    until File.exist?(OLDFILE) && File.exist?(NEWFILE)
+      create_timed_files(OLDFILE, NEWFILE)
+    end
 
     t1 = Rake.application.intern(FileTask, NEWFILE).enhance([OLDFILE])
     t2 = Rake.application.intern(FileTask, OLDFILE)
@@ -38,7 +40,9 @@ class Rake::TestFileTask < Test::Unit::TestCase
   end
 
   def test_file_times_old_depends_on_new
-    create_timed_files(OLDFILE, NEWFILE)
+    until File.exist?(OLDFILE) && File.exist?(NEWFILE)
+      create_timed_files(OLDFILE, NEWFILE)
+    end
 
     t1 = Rake.application.intern(FileTask,OLDFILE).enhance([NEWFILE])
     t2 = Rake.application.intern(FileTask, NEWFILE)
@@ -93,46 +97,46 @@ class Rake::TestDirectoryTask < Test::Unit::TestCase
   include Rake
 
   def setup
-    rm_rf "testdata", :verbose=>false
+    rm_rf "testdata2", :verbose=>false
   end
 
   def teardown
-    rm_rf "testdata", :verbose=>false
+    rm_rf "testdata2", :verbose=>false
   end
 
   def test_directory
     desc "DESC"
-    directory "testdata/a/b/c"
+    directory "testdata2/a/b/c"
-    assert_equal FileCreationTask, Task["testdata"].class
+    assert_equal FileCreationTask, Task["testdata2"].class
-    assert_equal FileCreationTask, Task["testdata/a"].class
+    assert_equal FileCreationTask, Task["testdata2/a"].class
-    assert_equal FileCreationTask, Task["testdata/a/b/c"].class
+    assert_equal FileCreationTask, Task["testdata2/a/b/c"].class
-    assert_nil             Task["testdata"].comment
+    assert_nil             Task["testdata2"].comment
-    assert_equal "DESC",   Task["testdata/a/b/c"].comment
+    assert_equal "DESC",   Task["testdata2/a/b/c"].comment
-    assert_nil             Task["testdata/a/b"].comment
+    assert_nil             Task["testdata2/a/b"].comment
     verbose(false) {
-      Task['testdata/a/b'].invoke
+      Task['testdata2/a/b'].invoke
     }
-    assert File.exist?("testdata/a/b")
+    assert File.exist?("testdata2/a/b")
-    assert ! File.exist?("testdata/a/b/c")
+    assert ! File.exist?("testdata2/a/b/c")
   end
 
   if Rake::Win32.windows?
     def test_directory_win32
       desc "WIN32 DESC"
-      FileUtils.mkdir_p("testdata")
+      FileUtils.mkdir_p("testdata2")
-      Dir.chdir("testdata") do
+      Dir.chdir("testdata2") do
-        directory 'c:/testdata/a/b/c'
+        directory 'c:/testdata2/a/b/c'
-        assert_equal FileCreationTask, Task['c:/testdata'].class
+        assert_equal FileCreationTask, Task['c:/testdata2'].class
-        assert_equal FileCreationTask, Task['c:/testdata/a'].class
+        assert_equal FileCreationTask, Task['c:/testdata2/a'].class
-        assert_equal FileCreationTask, Task['c:/testdata/a/b/c'].class
+        assert_equal FileCreationTask, Task['c:/testdata2/a/b/c'].class
-        assert_nil             Task['c:/testdata'].comment
+        assert_nil             Task['c:/testdata2'].comment
-        assert_equal "WIN32 DESC",   Task['c:/testdata/a/b/c'].comment
+        assert_equal "WIN32 DESC",   Task['c:/testdata2/a/b/c'].comment
-        assert_nil             Task['c:/testdata/a/b'].comment
+        assert_nil             Task['c:/testdata2/a/b'].comment
         verbose(false) {
-          Task['c:/testdata/a/b'].invoke
+          Task['c:/testdata2/a/b'].invoke
         }
-        assert File.exist?('c:/testdata/a/b')
+        assert File.exist?('c:/testdata2/a/b')
-        assert ! File.exist?('c:/testdata/a/b/c')
+        assert ! File.exist?('c:/testdata2/a/b/c')
       end
     end
   end