Use native I/O when the scheduler is supported.

Author: ioquatix

.github/workflows/async-v1.yaml

@@ -12,7 +12,7 @@ jobs:
           - ubuntu
 
         ruby:
-          - 2.7
+          - "2.7"
 
     env:
       BUNDLE_GEMFILE: gems/async-v1.rb

.github/workflows/async-v2.yaml

@@ -0,0 +1,29 @@
+name: Async v2
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ${{matrix.os}}-latest
+    
+    strategy:
+      matrix:
+        os:
+          - ubuntu
+        
+        ruby:
+          - "3.1.1"
+    
+    env:
+      BUNDLE_GEMFILE: gems/async-v2.rb
+    
+    steps:
+    - uses: actions/checkout@v2
+    - uses: ruby/setup-ruby@v1
+      with:
+        ruby-version: ${{matrix.ruby}}
+        bundler-cache: true
+    
+    - name: Run tests
+      timeout-minutes: 5
+      run: bundle exec rspec

gems/async-v2.rb

@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2023, by Samuel Williams.
+
+source 'https://rubygems.org'
+
+gemspec path: "../"
+
+gem 'async', '~> 2.0'

lib/async/io/generic.rb

@@ -84,71 +84,75 @@ def wrap(*args)
 
 			wraps ::IO, :external_encoding, :internal_encoding, :autoclose?, :autoclose=, :pid, :stat, :binmode, :flush, :set_encoding, :set_encoding_by_bom, :to_path, :to_io, :to_i, :reopen, :fileno, :fsync, :fdatasync, :sync, :sync=, :tell, :seek, :rewind, :path, :pos, :pos=, :eof, :eof?, :close_on_exec?, :close_on_exec=, :closed?, :close_read, :close_write, :isatty, :tty?, :binmode?, :sysseek, :advise, :ioctl, :fcntl, :nread, :ready?, :pread, :pwrite, :pathconf
 
-			# Read the specified number of bytes from the input stream. This is fast path.
-			# @example
-			#   data = io.sysread(512)
-			wrap_blocking_method :sysread, :read_nonblock
-			
-			alias readpartial read_nonblock
-			
-			# Read `length` bytes of data from the underlying I/O. If length is unspecified, read everything.
-			def read(length = nil, buffer = nil)
-				if buffer
-					buffer.clear
-				else
-					buffer = String.new
-				end
+			if Async::Scheduler.supported?
+				def_delegators :@io, :read, :write, :sysread, :syswrite, :read_nonblock, :write_nonblock, :readpartial
+			else
+				# Read the specified number of bytes from the input stream. This is fast path.
+				# @example
+				#   data = io.sysread(512)
+				wrap_blocking_method :sysread, :read_nonblock
+				
+				alias readpartial read_nonblock
 
-				if length
-					return String.new(encoding: Encoding::BINARY) if length <= 0
+				# Read `length` bytes of data from the underlying I/O. If length is unspecified, read everything.
+				def read(length = nil, buffer = nil)
+					if buffer
+						buffer.clear
+					else
+						buffer = String.new
+					end
 
-					# Fast path:
-					if buffer = self.sysread(length, buffer)
+					if length
+						return "" if length <= 0
 
-						# Slow path:
-						while buffer.bytesize < length
+						# Fast path:
+						if buffer = self.sysread(length, buffer)
+							
 							# Slow path:
-							if chunk = self.sysread(length - buffer.bytesize)
-								buffer << chunk
-							else
-								break
+							while buffer.bytesize < length
+								# Slow path:
+								if chunk = self.sysread(length - buffer.bytesize)
+									buffer << chunk
+								else
+									break
+								end
 							end
+							
+							return buffer
+						else
+							return nil
+						end
+					else
+						buffer = self.sysread(BLOCK_SIZE, buffer)
+						
+						while chunk = self.sysread(BLOCK_SIZE)
+							buffer << chunk
 						end
 
 						return buffer
-					else
-						return nil
 					end
-				else
-					buffer = self.sysread(BLOCK_SIZE, buffer)
-					
-					while chunk = self.sysread(BLOCK_SIZE)
-						buffer << chunk
-					end
-					
-					return buffer
 				end
-			end
-			
-			# Write entire buffer to output stream. This is fast path.
-			# @example
-			#   io.syswrite("Hello World")
-			wrap_blocking_method :syswrite, :write_nonblock
-			
-			def write(buffer)
-				# Fast path:
-				written = self.syswrite(buffer)
-				remaining = buffer.bytesize - written
 
-				while remaining > 0
-					# Slow path:
-					length = self.syswrite(buffer.byteslice(written, remaining))
+				# Write entire buffer to output stream. This is fast path.
+				# @example
+				#   io.syswrite("Hello World")
+				wrap_blocking_method :syswrite, :write_nonblock
+				
+				def write(buffer)
+					# Fast path:
+					written = self.syswrite(buffer)
+					remaining = buffer.bytesize - written
 
-					remaining -= length
-					written += length
+					while remaining > 0
+						# Slow path:
+						length = self.syswrite(buffer.byteslice(written, remaining))
+						
+						remaining -= length
+						written += length
+					end
+					
+					return written
 				end
-				
-				return written
 			end
 
 			def << buffer

lib/async/io/stream.rb

@@ -234,6 +234,20 @@ def eof!
 
 			private
 
+			def sysread(size, buffer)
+				while true
+					result = @io.read_nonblock(size, buffer, exception: false)
+					
+					if result == :wait_readable
+						@io.wait_readable
+					elsif result == :wait_writable
+						@io.wait_writable
+					else
+						return result
+					end
+				end
+			end
+			
 			# Fills the buffer from the underlying stream.
 			def fill_read_buffer(size = @block_size)
 				# We impose a limit because the underlying `read` system call can fail if we request too much data in one go.
@@ -245,19 +259,23 @@ def fill_read_buffer(size = @block_size)
 				flush
 
 				if @read_buffer.empty?
-					if @io.read_nonblock(size, @read_buffer, exception: false)
+					if sysread(size, @read_buffer)
 						# Console.logger.debug(self, name: "read") {@read_buffer.inspect}
 						return true
 					end
 				else
-					if chunk = @io.read_nonblock(size, @input_buffer, exception: false)
+					if chunk = sysread(size, @input_buffer)
 						@read_buffer << chunk
 						# Console.logger.debug(self, name: "read") {@read_buffer.inspect}
 
 						return true
 					end
 				end
 
+				@eof = true
+				return false
+			
+			rescue EOFError
 				# else for both cases above:
 				@eof = true
 				return false

lib/async/io/tcp_socket.rb

@@ -12,55 +12,21 @@ module Async
 	module IO
 		# Asynchronous TCP socket wrapper.
 		class TCPSocket < IPSocket
-			wraps ::TCPSocket
+			wraps ::TCPSocket, :gets, :puts
 
 			def initialize(remote_host, remote_port = nil, local_host = nil, local_port = nil)
 				if remote_host.is_a? ::TCPSocket
 					super(remote_host)
 				else
-					remote_address = Addrinfo.tcp(remote_host, remote_port)
-					local_address = Addrinfo.tcp(local_host, local_port) if local_host
-					
-					# We do this unusual dance to avoid leaking an "open" socket instance.
-					socket = Socket.connect(remote_address, local_address: local_address)
-					fd = socket.fcntl(Fcntl::F_DUPFD)
-					Console.logger.debug(self) {"Connected to #{remote_address.inspect}: #{fd}"}
-					socket.close
-					
-					super(::TCPSocket.for_fd(fd))
-					
-					# The equivalent blocking operation. Unfortunately there is no trivial way to make this non-blocking.
-					# super(::TCPSocket.new(remote_host, remote_port, local_host, local_port))
+					super(::TCPSocket.new(remote_host, remote_port, local_host, local_port))
 				end
-				
-				@stream = Stream.new(self)
 			end
 
 			class << self
 				alias open new
 			end
 
-			def close
-				@stream.flush
-				super
-			end
-			
 			include Peer
-			
-			attr :stream
-			
-			# The way this buffering works is pretty atrocious.
-			def_delegators :@stream, :gets, :puts
-			
-			def sysread(size, buffer = nil)
-				data = @stream.read_partial(size)
-				
-				if buffer
-					buffer.replace(data)
-				end
-				
-				return data
-			end
 		end
 
 		# Asynchronous TCP server wrappper.

spec/async/io/tcp_socket_spec.rb

@@ -14,7 +14,7 @@
 
 	# Shared port for localhost network tests.
 	let(:server_address) {Async::IO::Address.tcp("localhost", 6788)}
-	let(:data) {"The quick brown fox jumped over the lazy dog."}
+	let(:data) {"The quick brown fox jumped over the lazy dog.\n"}
 
 	describe Async::IO::TCPServer do
 		it_should_behave_like Async::IO::Generic