merge revision(s) 46108,46209,46223,46297,48223,48224: [Backport #12168]

	* test/openssl: Join threads.

	* ext/openssl/ossl_ssl.c (ossl_ssl_close): Fix sync_close to work
	  when SSL is not started.
	  This fix the fd leak by test_https_proxy_authentication in
	  test/net/http/test_https_proxy.rb.

	* ext/openssl/lib/openssl/ssl.rb (SSLServer#accept): Close a socket
	  if any exception occur. 

	* test/ruby/envutil.rb (assert_join_threads): New assertion to
	  join multiple threads without exceptions.

	* test/openssl/utils.rb (start_server, server_loop): Use a
	  pipe to stop server instead of shutdown/close a listening socket.


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

5 files changed, 186 insertions, 87 deletions

diff --git a/test/openssl/test_pair.rb b/test/openssl/test_pair.rb
index e6af9b8c68..725d26a40f 100644
--- a/test/openssl/test_pair.rb
+++ b/test/openssl/test_pair.rb
@@ -83,16 +83,20 @@ end
 module OpenSSL::TestEOF1M
   def open_file(content)
     s1, s2 = ssl_pair
-    Thread.new { s2 << content; s2.close }
+    th = Thread.new { s2 << content; s2.close }
     yield s1
+  ensure
+    th.join
   end
 end
 
 module OpenSSL::TestEOF2M
   def open_file(content)
     s1, s2 = ssl_pair
-    Thread.new { s1 << content; s1.close }
+    th = Thread.new { s1 << content; s1.close }
     yield s2
+  ensure
+    th.join
   end
 end
 
@@ -312,6 +316,7 @@ module OpenSSL::TestPairM
     s1.print "a\ndef"
     assert_equal("a\n", s2.gets)
   ensure
+    th.join
     s1.close if s1 && !s1.closed?
     s2.close if s2 && !s2.closed?
     serv.close if serv && !serv.closed?
diff --git a/test/openssl/test_ssl.rb b/test/openssl/test_ssl.rb
index 5589a0f173..db139eb1c7 100644
--- a/test/openssl/test_ssl.rb
+++ b/test/openssl/test_ssl.rb
@@ -121,7 +121,12 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
       assert_raise(OpenSSL::SSL::SSLError, Errno::ECONNRESET){
         sock = TCPSocket.new("127.0.0.1", port)
         ssl = OpenSSL::SSL::SSLSocket.new(sock)
-        ssl.connect
+        ssl.sync_close = true
+        begin
+          ssl.connect
+        ensure
+          ssl.close
+        end
       }
 
       ctx = OpenSSL::SSL::SSLContext.new
@@ -166,27 +171,27 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
   end
 
   def test_starttls
-    start_server(PORT, OpenSSL::SSL::VERIFY_NONE, false){|server, port|
-      sock = TCPSocket.new("127.0.0.1", port)
-      ssl = OpenSSL::SSL::SSLSocket.new(sock)
-      ssl.sync_close = true
-      str = "x" * 1000 + "\n"
+    OpenSSL::TestUtils.silent do
+      start_server(PORT, OpenSSL::SSL::VERIFY_NONE, false){|server, port|
+        sock = TCPSocket.new("127.0.0.1", port)
+        ssl = OpenSSL::SSL::SSLSocket.new(sock)
+        ssl.sync_close = true
+        str = "x" * 1000 + "\n"
 
-      OpenSSL::TestUtils.silent do
         ITERATIONS.times{
           ssl.puts(str)
           assert_equal(str, ssl.gets)
         }
         starttls(ssl)
-      end
 
-      ITERATIONS.times{
-        ssl.puts(str)
-        assert_equal(str, ssl.gets)
-      }
+        ITERATIONS.times{
+          ssl.puts(str)
+          assert_equal(str, ssl.gets)
+        }
 
-      ssl.close
-    }
+        ssl.close
+      }
+    end
   end
 
   def test_parallel
@@ -217,9 +222,16 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
       ctx = OpenSSL::SSL::SSLContext.new
       ctx.set_params
       ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-      assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
-      assert_equal(OpenSSL::X509::V_ERR_SELF_SIGNED_CERT_IN_CHAIN, ssl.verify_result)
+      ssl.sync_close = true
+      begin
+        assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
+        assert_equal(OpenSSL::X509::V_ERR_SELF_SIGNED_CERT_IN_CHAIN, ssl.verify_result)
+      ensure
+        ssl.close
+      end
+    }
 
+    start_server(PORT, OpenSSL::SSL::VERIFY_NONE, true){|server, port|
       sock = TCPSocket.new("127.0.0.1", port)
       ctx = OpenSSL::SSL::SSLContext.new
       ctx.set_params(
@@ -229,9 +241,16 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
         end
       )
       ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-      ssl.connect
-      assert_equal(OpenSSL::X509::V_OK, ssl.verify_result)
+      ssl.sync_close = true
+      begin
+        ssl.connect
+        assert_equal(OpenSSL::X509::V_OK, ssl.verify_result)
+      ensure
+        ssl.close
+      end
+    }
 
+    start_server(PORT, OpenSSL::SSL::VERIFY_NONE, true){|server, port|
       sock = TCPSocket.new("127.0.0.1", port)
       ctx = OpenSSL::SSL::SSLContext.new
       ctx.set_params(
@@ -241,8 +260,13 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
         end
       )
       ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-      assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
-      assert_equal(OpenSSL::X509::V_ERR_APPLICATION_VERIFICATION, ssl.verify_result)
+      ssl.sync_close = true
+      begin
+        assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
+        assert_equal(OpenSSL::X509::V_ERR_APPLICATION_VERIFICATION, ssl.verify_result)
+      ensure
+        ssl.close
+      end
     }
   end
 
@@ -257,12 +281,16 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
         end
       )
       ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-      OpenSSL::TestUtils.silent do
-        # SSLError, not RuntimeError
-        assert_raise(OpenSSL::SSL::SSLError) { ssl.connect }
+      ssl.sync_close = true
+      begin
+        OpenSSL::TestUtils.silent do
+          # SSLError, not RuntimeError
+          assert_raise(OpenSSL::SSL::SSLError) { ssl.connect }
+        end
+        assert_equal(OpenSSL::X509::V_ERR_CERT_REJECTED, ssl.verify_result)
+      ensure
+        ssl.close
       end
-      assert_equal(OpenSSL::X509::V_ERR_CERT_REJECTED, ssl.verify_result)
-      ssl.close
     }
   end
 
@@ -279,8 +307,13 @@ class OpenSSL::TestSSL < OpenSSL::SSLTestCase
       assert(ciphers_names.all?{|v| /ADH/ !~ v })
       assert(ciphers_versions.all?{|v| /SSLv2/ !~ v })
       ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-      assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
-      assert_equal(OpenSSL::X509::V_ERR_SELF_SIGNED_CERT_IN_CHAIN, ssl.verify_result)
+      ssl.sync_close = true
+      begin
+        assert_raise(OpenSSL::SSL::SSLError){ ssl.connect }
+        assert_equal(OpenSSL::X509::V_ERR_SELF_SIGNED_CERT_IN_CHAIN, ssl.verify_result)
+      ensure
+        ssl.close
+      end
     }
   end
 
@@ -822,6 +855,15 @@ end
     }
   end
 
+  def test_sync_close_without_connect
+    Socket.open(:INET, :STREAM) {|s|
+      ssl = OpenSSL::SSL::SSLSocket.new(s)
+      ssl.sync_close = true
+      ssl.close
+      assert(s.closed?)
+    }
+  end
+
   private
 
   def start_server_version(version, ctx_proc=nil, server_proc=nil, &blk)
diff --git a/test/openssl/utils.rb b/test/openssl/utils.rb
index da281610b6..655cd0b6d2 100644
--- a/test/openssl/utils.rb
+++ b/test/openssl/utils.rb
@@ -240,19 +240,24 @@ AQjjxMXhwULlmuR/K+WwlaZPiLIBYalLAZQ7ZbOPeVkJ8ePao0eLAgEC
       ssl.close rescue nil
     end
 
-    def server_loop(ctx, ssls, server_proc)
+    def server_loop(ctx, ssls, stop_pipe_r, server_proc, threads)
       loop do
         ssl = nil
         begin
+          readable, = IO.select([ssls, stop_pipe_r])
+          if readable.include? stop_pipe_r
+            return
+          end
           ssl = ssls.accept
         rescue OpenSSL::SSL::SSLError
           retry
         end
 
-        Thread.start do
+        th = Thread.start do
           Thread.current.abort_on_exception = true
           server_proc.call(ctx, ssl)
         end
+        threads << th
       end
     rescue Errno::EBADF, IOError, Errno::EINVAL, Errno::ECONNABORTED, Errno::ENOTSOCK, Errno::ECONNRESET
     end
@@ -262,6 +267,7 @@ AQjjxMXhwULlmuR/K+WwlaZPiLIBYalLAZQ7ZbOPeVkJ8ePao0eLAgEC
       use_anon_cipher = args.fetch(:use_anon_cipher, false)
       server_proc = args[:server_proc]
       server_proc ||= method(:readwrite_loop)
+      threads = []
 
       store = OpenSSL::X509::Store.new
       store.add_cert(@ca_cert)
@@ -286,13 +292,15 @@ AQjjxMXhwULlmuR/K+WwlaZPiLIBYalLAZQ7ZbOPeVkJ8ePao0eLAgEC
         retry
       end
 
+      stop_pipe_r, stop_pipe_w = IO.pipe
+
       ssls = OpenSSL::SSL::SSLServer.new(tcps, ctx)
       ssls.start_immediately = start_immediately
 
       begin
         server = Thread.new do
           Thread.current.abort_on_exception = true
-          server_loop(ctx, ssls, server_proc)
+          server_loop(ctx, ssls, stop_pipe_r, server_proc, threads)
         end
 
         $stderr.printf("%s started: pid=%d port=%d\n", SSL_SERVER, $$, port) if $DEBUG
@@ -300,18 +308,10 @@ AQjjxMXhwULlmuR/K+WwlaZPiLIBYalLAZQ7ZbOPeVkJ8ePao0eLAgEC
         block.call(server, port.to_i)
       ensure
         begin
-          begin
-            tcps.shutdown
-          rescue Errno::ENOTCONN
-            # when `Errno::ENOTCONN: Socket is not connected' on some platforms,
-            # call #close instead of #shutdown.
-            tcps.close
-            tcps = nil
-          end if (tcps)
+          stop_pipe_w.close
           if (server)
             server.join(5)
             if server.alive?
-              server.kill
               server.join
               flunk("TCPServer was closed and SSLServer is still alive") unless $!
             end
@@ -320,6 +320,10 @@ AQjjxMXhwULlmuR/K+WwlaZPiLIBYalLAZQ7ZbOPeVkJ8ePao0eLAgEC
           tcps.close if (tcps)
         end
       end
+    ensure
+      stop_pipe_r.close if !stop_pipe_r.closed?
+      stop_pipe_w.close if !stop_pipe_w.closed?
+      assert_join_threads(threads)
     end
 
     def starttls(ssl)
diff --git a/test/ruby/envutil.rb b/test/ruby/envutil.rb
index a6a9f70820..1fe706df49 100644
--- a/test/ruby/envutil.rb
+++ b/test/ruby/envutil.rb
@@ -434,6 +434,36 @@ eom
         AssertFile
       end
 
+      # threads should respond to shift method.
+      # Array and Queue can be used.
+      def assert_join_threads(threads, message = nil)
+        errs = []
+        values = []
+        while th = threads.shift
+          begin
+            values << th.value
+          rescue Exception
+            errs << $!
+          end
+        end
+        if !errs.empty?
+          msg = errs.map {|err|
+            err.backtrace.map.with_index {|line, i|
+              if i == 0
+                "#{line}: #{err.message} (#{err.class})"
+              else
+                "\tfrom #{line}"
+              end
+            }.join("\n")
+          }.join("\n---\n")
+          if message
+            msg = "#{message}\n#{msg}"
+          end
+          raise MiniTest::Assertion, msg
+        end
+        values
+      end
+
       class << (AssertFile = Struct.new(:failure_message).new)
         include Assertions
         def assert_file_predicate(predicate, *args)
diff --git a/test/ruby/test_io.rb b/test/ruby/test_io.rb
index d3ac0c5f01..2283221d9c 100644
--- a/test/ruby/test_io.rb
+++ b/test/ruby/test_io.rb
@@ -584,11 +584,13 @@ class TestIO < Test::Unit::TestCase
             end
             t1 = Thread.new { w1 << megacontent; w1.close }
             t2 = Thread.new { r2.read }
-            ret = IO.copy_stream(r1, w2)
-            assert_equal(megacontent.bytesize, ret)
-            w2.close
-            t1.join
-            assert_equal(megacontent, t2.value)
+            t3 = Thread.new {
+              ret = IO.copy_stream(r1, w2)
+              assert_equal(megacontent.bytesize, ret)
+              w2.close
+            }
+            _, t2_value, _ = assert_join_threads([t1, t2, t3])
+            assert_equal(megacontent, t2_value)
           }
         }
       }
@@ -601,11 +603,13 @@ class TestIO < Test::Unit::TestCase
         with_pipe {|r2, w2|
           t1 = Thread.new { w1 << megacontent; w1.close }
           t2 = Thread.new { r2.read }
-          ret = IO.copy_stream(r1, w2)
-          assert_equal(megacontent.bytesize, ret)
-          w2.close
-          t1.join
-          assert_equal(megacontent, t2.value)
+          t3 = Thread.new {
+            ret = IO.copy_stream(r1, w2)
+            assert_equal(megacontent.bytesize, ret)
+            w2.close
+          }
+          _, t2_value, _ = assert_join_threads([t1, t2, t3])
+          assert_equal(megacontent, t2_value)
         }
       }
     }
@@ -614,11 +618,14 @@ class TestIO < Test::Unit::TestCase
   def test_copy_stream_megacontent_file_to_pipe
     with_megasrc {|megasrc, megacontent|
       with_pipe {|r, w|
-        t = Thread.new { r.read }
-        ret = IO.copy_stream(megasrc, w)
-        assert_equal(megacontent.bytesize, ret)
-        w.close
-        assert_equal(megacontent, t.value)
+        t1 = Thread.new { r.read }
+        t2 = Thread.new {
+          ret = IO.copy_stream(megasrc, w)
+          assert_equal(megacontent.bytesize, ret)
+          w.close
+        }
+        t1_value, _ = assert_join_threads([t1, t2])
+        assert_equal(megacontent, t1_value)
       }
     }
   end
@@ -666,11 +673,13 @@ class TestIO < Test::Unit::TestCase
   def test_copy_stream_socket2
     with_bigsrc {|bigsrc, bigcontent|
       with_socketpair {|s1, s2|
-        t = Thread.new { s2.read }
-        ret = IO.copy_stream(bigsrc, s1)
-        assert_equal(bigcontent.bytesize, ret)
-        s1.close
-        result = t.value
+        t1 = Thread.new { s2.read }
+        t2 = Thread.new {
+          ret = IO.copy_stream(bigsrc, s1)
+          assert_equal(bigcontent.bytesize, ret)
+          s1.close
+        }
+        result, _ = assert_join_threads([t1, t2])
         assert_equal(bigcontent, result)
       }
     }
@@ -679,11 +688,13 @@ class TestIO < Test::Unit::TestCase
   def test_copy_stream_socket3
     with_bigsrc {|bigsrc, bigcontent|
       with_socketpair {|s1, s2|
-        t = Thread.new { s2.read }
-        ret = IO.copy_stream(bigsrc, s1, 10000)
-        assert_equal(10000, ret)
-        s1.close
-        result = t.value
+        t1 = Thread.new { s2.read }
+        t2 = Thread.new {
+          ret = IO.copy_stream(bigsrc, s1, 10000)
+          assert_equal(10000, ret)
+          s1.close
+        }
+        result, _ = assert_join_threads([t1, t2])
         assert_equal(bigcontent[0,10000], result)
       }
     }
@@ -694,12 +705,14 @@ class TestIO < Test::Unit::TestCase
       File.open(bigsrc) {|f|
         assert_equal(0, f.pos)
         with_socketpair {|s1, s2|
-          t = Thread.new { s2.read }
-          ret = IO.copy_stream(f, s1, nil, 100)
-          assert_equal(bigcontent.bytesize-100, ret)
-          assert_equal(0, f.pos)
-          s1.close
-          result = t.value
+          t1 = Thread.new { s2.read }
+          t2 = Thread.new {
+            ret = IO.copy_stream(f, s1, nil, 100)
+            assert_equal(bigcontent.bytesize-100, ret)
+            assert_equal(0, f.pos)
+            s1.close
+          }
+          result, _ = assert_join_threads([t1, t2])
           assert_equal(bigcontent[100..-1], result)
         }
       }
@@ -712,12 +725,14 @@ class TestIO < Test::Unit::TestCase
         assert_equal(bigcontent[0,100], f.read(100))
         assert_equal(100, f.pos)
         with_socketpair {|s1, s2|
-          t = Thread.new { s2.read }
-          ret = IO.copy_stream(f, s1)
-          assert_equal(bigcontent.bytesize-100, ret)
-          assert_equal(bigcontent.length, f.pos)
-          s1.close
-          result = t.value
+          t1 = Thread.new { s2.read }
+          t2 = Thread.new {
+            ret = IO.copy_stream(f, s1)
+            assert_equal(bigcontent.bytesize-100, ret)
+            assert_equal(bigcontent.length, f.pos)
+            s1.close
+          }
+          result, _ = assert_join_threads([t1, t2])
           assert_equal(bigcontent[100..-1], result)
         }
       }
@@ -735,11 +750,13 @@ class TestIO < Test::Unit::TestCase
         rescue Errno::EBADF
           skip "nonblocking IO for pipe is not implemented"
         end
-        t = Thread.new { s2.read }
-        ret = IO.copy_stream("megasrc", s1)
-        assert_equal(megacontent.bytesize, ret)
-        s1.close
-        result = t.value
+        t1 = Thread.new { s2.read }
+        t2 = Thread.new {
+          ret = IO.copy_stream("megasrc", s1)
+          assert_equal(megacontent.bytesize, ret)
+          s1.close
+        }
+        result, _ = assert_join_threads([t1, t2])
         assert_equal(megacontent, result)
       }
     }
@@ -967,11 +984,12 @@ class TestIO < Test::Unit::TestCase
       w.write "zz"
       src = StringIO.new("abcd")
       IO.copy_stream(src, w)
-      t = Thread.new {
+      t1 = Thread.new {
         w.close
       }
-      assert_equal("zzabcd", r.read)
-      t.join
+      t2 = Thread.new { r.read }
+      _, result = assert_join_threads([t1, t2])
+      assert_equal("zzabcd", result)
     }
   end
 
@@ -2633,7 +2651,7 @@ End
           threads << Thread.new {write_file.print(i)}
           threads << Thread.new {read_file.read}
         end
-        threads.each {|t| t.join}
+        assert_join_threads(threads)
         assert(true, "[ruby-core:37197]")
       ensure
         read_file.close