Nonblock without exception

lib/celluloid/io/dns_resolver.rb

@@ -42,7 +42,10 @@ def resolve(hostname)
 
         query = build_query(hostname)
         @socket.send query.encode, 0, @server.to_s, DNS_PORT
-        data, _ = @socket.recvfrom(MAX_PACKET_SIZE)
+        data, _ = RUBY_VERSION >= "2.3" ?
+                  # Documenting the cryptic second parameter (flags): http://stackoverflow.com/questions/1527895/where-are-msg-options-defined-for-ruby-sockets
+                  @socket.recvfrom(MAX_PACKET_SIZE, 0, nil, exception: false) :
+                  @socket.recvfrom(MAX_PACKET_SIZE)
         response = Resolv::DNS::Message.decode(data)
 
         addrs = []

lib/celluloid/io/socket.rb

@@ -70,6 +70,32 @@ def self.try_convert(socket, convert_io = true)
         end
       end
 
+      private 
+
+      def perform_io
+        loop do
+          begin
+            result = yield
+
+            case result
+            when :wait_readable then wait_readable
+            when :wait_writable then wait_writable
+            when NilClass       then return :eof
+            else                return result
+            end
+          rescue ::IO::WaitReadable
+            wait_readable
+            retry 
+          rescue ::IO::WaitWritable,
+                 Errno::EAGAIN
+            wait_writable
+            retry
+          end
+        end
+      rescue EOFError
+        :eof
+      end
+
       class << self
         extend Forwardable
         def_delegators '::Socket', *(::Socket.methods - self.methods - [:try_convert])

lib/celluloid/io/stream.rb

@@ -38,12 +38,13 @@ def sysread(length = nil, buffer = nil)
         buffer ||= ''.force_encoding(Encoding::ASCII_8BIT)
 
         @read_latch.synchronize do
-          begin
-            read_nonblock(length, buffer)
-          rescue ::IO::WaitReadable
-            wait_readable
-            retry
+          op = perform_io do
+            # TODO: remove after ending ruby 2.0.0 support
+            RUBY_VERSION < "2.1" ?
+            read_nonblock(length, buffer) :
+            read_nonblock(length, buffer, exception: false) 
           end
+          raise EOFError if op == :eof
         end
 
         buffer
@@ -58,17 +59,13 @@ def syswrite(string)
 
         @write_latch.synchronize do
           while total_written < length
-            begin
-              written = write_nonblock(remaining)
-            rescue ::IO::WaitWritable
-              wait_writable
-              retry
-            rescue EOFError
-              return total_written
-            rescue Errno::EAGAIN
-              wait_writable
-              retry
-            end
+            written = perform_io do
+              # TODO: remove after ending ruby 2.0.0 support
+              RUBY_VERSION < "2.1" ?
+              write_nonblock(remaining) :
+              write_nonblock(remaining, exception: false)
+            end 
+            return total_written if written == :eof
 
             total_written += written
 

lib/celluloid/io/udp_socket.rb

@@ -31,18 +31,15 @@ def wait_readable; Celluloid::IO.wait_readable(self); end
       # MSG_ options. The first element of the results, mesg, is the data
       # received. The second element, sender_addrinfo, contains
       # protocol-specific address information of the sender.
-      def recvfrom(maxlen, flags = 0)
-        begin
+      def recvfrom(*args)
+        perform_io do
           socket = to_io
           if socket.respond_to? :recvfrom_nonblock
-            socket.recvfrom_nonblock(maxlen, flags)
+            socket.recvfrom_nonblock(*args)
           else
             # FIXME: hax for JRuby
-            socket.recvfrom(maxlen, flags)
+            socket.recvfrom(*args)
           end
-        rescue ::IO::WaitReadable
-          wait_readable
-          retry
         end
       end