Use native I/O when the scheduler is supported.

Author: ioquatix

async-io.gemspec

@@ -15,7 +15,7 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 2.5"
 
-	spec.add_dependency "async", "~> 1.14"
+	spec.add_dependency "async", "~> 1.29"
 
 	spec.add_development_dependency "async-container", "~> 0.15"
 	spec.add_development_dependency "async-rspec", "~> 1.10"

lib/async/io/generic.rb

@@ -100,71 +100,75 @@ def wrap(*args)
 
 			wraps ::IO, :external_encoding, :internal_encoding, :autoclose?, :autoclose=, :pid, :stat, :binmode, :flush, :set_encoding, :set_encoding_by_bom, :to_io, :to_i, :reopen, :fileno, :fsync, :fdatasync, :sync, :sync=, :tell, :seek, :rewind, :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 "" 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

@@ -258,16 +258,30 @@ def fill_read_buffer(size = @block_size)
 				flush
 
 				if @read_buffer.empty?
-					if @io.read_nonblock(size, @read_buffer, exception: false)
-						# Async.logger.debug(self, name: "read") {@read_buffer.inspect}
-						return true
+					while true
+						case result = @io.read_nonblock(size, @read_buffer, exception: false)
+						when String
+							# Async.logger.debug(self, name: "read") {@read_buffer.inspect}
+							return true
+						when Symbol
+							@io.__send__(result)
+						else
+							break
+						end
 					end
 				else
-					if chunk = @io.read_nonblock(size, @input_buffer, exception: false)
-						@read_buffer << chunk
-						# Async.logger.debug(self, name: "read") {@read_buffer.inspect}
-						
-						return true
+					while true
+						case result = @io.read_nonblock(size, @input_buffer, exception: false)
+						when String
+							@read_buffer << result
+							# Async.logger.debug(self, name: "read") {@read_buffer.inspect}
+							
+							return true
+						when Symbol
+							@io.__send__(result)
+						else
+							break
+						end
 					end
 				end