Merge pull request #501 from larskanis/racy-tcp-gate

Fix racy TcpGateSwitcher

spec/helpers/tcp_gate_switcher.rb

@@ -7,15 +7,15 @@
 # The TCP communication in a C extension can be verified in a (mostly) timing insensitive way.
 # If a call does IO but doesn't handle non-blocking state, the test will block and can be caught by an external timeout.
 #
-#   PG.connect
-#    port:5444                    TcpGateSwitcher                      DB
-#  -------------      ----------------------------------------      --------
-#  |    non-   |      | TCPServer                  TCPSocket |      |      |
-#  |  blocking |----->|  port 5444                  port 5432|----->|Server|
-#  |   specs   |      |                                      |      | port |
-#  '------|----'      |,--> stop_read :     <-send data--    |      | 5432 |
-#         '---------------> stop_write:     --send data->    |      '------'
-#                     '--------------------------------------'
+#
+#   PG.connect                    TcpGateSwitcher
+#    port:5444        .--------------------------------------.
+#        .--start/stop---------------> T                     |         DB
+#  .-----|-----.      |                | /                   |      .------.
+#  |    non-   |      |                |/                    |      |Server|
+#  |  blocking |      | TCPServer      /           TCPSocket |      | port |
+#  |   specs   |------->port 5444-----/   ---------port 5432------->| 5432 |
+#  '-----------'      '--------------------------------------'      '------'
 
 module Helpers
 class TcpGateSwitcher
@@ -32,10 +32,10 @@ def initialize(internal_io, external_host, external_port, debug: false)
 			@debug = debug
 			@wait = nil
 
-			Thread.new do
+			@th1 = Thread.new do
 				read
 			end
-			Thread.new do
+			@th2 = Thread.new do
 				write
 			end
 		end
@@ -117,8 +117,8 @@ def write_fds
 		# Make sure all data is transferred and both connections are closed.
 		def finish
 			puts "finish transfers #{write_fds} and #{read_fds}"
-			write
-			read
+			@th1.join
+			@th2.join
 		end
 
 		def start
@@ -143,12 +143,13 @@ def initialize(external_host:, external_port:, internal_host: 'localhost', inter
 		@debug = debug
 		puts "TcpGate server listening: #{@server_io.inspect}"
 
-		run
+		@th = run
 	end
 
 	def finish
 		@finish = true
 		TCPSocket.new('localhost', internal_port).close
+		@th.join
 	end
 
 	def internal_port
@@ -175,7 +176,7 @@ def run
 					puts "accept new int:#{conn.internal_io.inspect} from #{conn.internal_io.remote_address.inspect} server fd:#{@server_io.fileno}"
 					@connections << conn
 
-					# Handle the reading and writing in a separate thread
+					# Unblock read and write transfer
 					conn.start
 				end
 

spec/pg/connection_spec.rb

@@ -641,6 +641,9 @@
 			unless RUBY_PLATFORM =~ /i386-mingw|x86_64-darwin|x86_64-linux/
 				skip "this spec depends on out-of-memory condition in put_copy_data, which is not reliable on all platforms"
 			end
+			if RUBY_ENGINE == "truffleruby"
+				skip "TcpGateSwitcher transfers wrong data on Truffleruby"
+			end
 
 			run_with_gate(200) do |conn, gate|
 				conn.setnonblocking(true)
@@ -668,6 +671,10 @@
 		end
 
 		it "needs to flush data after send_query" do
+			if RUBY_ENGINE == "truffleruby"
+				skip "TcpGateSwitcher transfers wrong data on Truffleruby"
+			end
+
 			run_with_gate(200) do |conn, gate|
 				conn.setnonblocking(true)