added proper selector/monitor, replaced where necessary

lib/httpx/channel/tcp.rb

@@ -78,6 +78,11 @@ module HTTPX::Channel
       @processor.send(request)
     end
 
+    def drain
+      dread
+      dwrite
+    end
+
     if RUBY_VERSION < "2.3"
       def dread(size = BUFFER_SIZE)
         begin

lib/httpx/connection.rb

@@ -3,6 +3,7 @@
 require "socket"
 require "timeout"
 
+require "httpx/selector"
 require "httpx/channel"
 
 module HTTPX
@@ -12,6 +13,7 @@ module HTTPX
       @timeout = options.timeout
       @channels = []
       @responses = {}
+      @selector = Selector.new
     end
 
     # opens a channel to the IP reachable through +uri+.
@@ -27,7 +29,9 @@ module HTTPX
         uri.scheme == channel.uri.scheme
       end || begin
         channel = Channel.by(uri, @options)
-        @channels << channel 
+        @channels << channel
+        monitor = @selector.register(channel, :rw)
+        monitor.value = -> { channel.drain }
         channel
       end
     end
@@ -46,27 +50,25 @@ module HTTPX
     end
 
     def process_events(timeout: @timeout.timeout)
-      rmonitors = @channels
-      wmonitors = rmonitors.reject(&:empty?)
-      readers, writers = IO.select(rmonitors, wmonitors, nil, timeout)
-      raise TimeoutError, "timed out waiting for data" if readers.nil? && writers.nil?
-      readers.each do |reader|
-        channel = catch(:close) { reader.dread }
-        close(channel) if channel 
-      end if readers
-      writers.each do |writer|
-        channel = catch(:close) { writer.dwrite }
-        close(channel) if channel 
-      end if writers
+      @selector.select(timeout) do |monitor|
+        if task = monitor.value
+          channel = catch(:close) { task.call } 
+          close(channel) if channel 
+        end
+      end
     end
 
     def close(channel = nil)
       if channel
         channel.close
-        @channels.delete(channel) if channel.closed?
+        if channel.closed?
+          @channels.delete(channel)
+          @selector.deregister(channel)
+        end
       else
         while ch = @channels.shift
           ch.close
+          @selector.deregister(ch)
         end 
       end
     end

lib/httpx/selector.rb

@@ -0,0 +1,149 @@
+# frozen_string_literal: true
+
+module HTTPX 
+  class Selector
+    #
+    # I/O monitor
+    #
+    class Monitor
+      attr_accessor :value, :interests, :readiness
+
+      def initialize(io, interests, reactor)
+        @io = io
+        @interests = interests
+        @reactor = reactor
+        @closed = false
+      end
+
+      def readable?
+        @interests == :rw || @interests == :r
+      end
+
+      def writable?
+        @interests == :rw || @interests == :w
+      end
+
+      # closes +@io+, deregisters from reactor (unless +deregister+ is false)
+      def close(deregister = true)
+        return if @closed
+        @closed = true
+        @reactor.deregister(@io) if deregister
+      end
+
+      def closed?
+        @closed
+      end
+
+      # :nocov:
+      def to_s
+        "#<#{self.class}: #{@io}(closed:#{@closed}) #{@interests} #{object_id.to_s(16)}>"
+      end
+      # :nocov:
+    end
+
+    def initialize
+      @readers = {}
+      @writers = {}
+      @lock = Mutex.new
+      @__r__, @__w__ = IO.pipe
+      @closed = false
+    end
+
+    # deregisters +io+ from selectables.
+    def deregister(io)
+      @lock.synchronize do
+        monitor = @readers.delete(io)
+        monitor ||= @writers.delete(io)
+        monitor.close(false) if monitor
+      end
+    end
+
+    # register +io+ for +interests+ events.
+    def register(io, interests)
+      readable = interests == :r || interests == :rw
+      writable = interests == :w || interests == :rw
+      @lock.synchronize do
+        if readable
+          monitor = @readers[io]
+          if monitor
+            monitor.interests = interests
+          else
+            monitor = Monitor.new(io, interests, self)
+          end
+          @readers[io] = monitor
+        end
+        if writable
+          monitor = @writers[io]
+          if monitor
+            monitor.interests = interests
+          else
+            # reuse object
+            monitor = readable ? @readers[io] : Monitor.new(io, interests, self)
+          end
+          @writers[io] = monitor
+        end
+        monitor
+      end
+    end
+
+    # waits for read/write events for +interval+. Yields for monitors of
+    # selected IO objects.
+    #
+    def select(interval)
+      begin
+        r = nil
+        w = nil
+        @lock.synchronize do
+          r = @readers.keys
+          w = @writers.keys
+        end
+        r.unshift(@__r__)
+
+        readers, writers = IO.select(r, w, nil, interval)
+      rescue IOError, SystemCallError
+        @lock.synchronize do
+          @readers.reject! { |io, _| io.closed? }
+          @writers.reject! { |io, _| io.closed? }
+        end
+        retry
+      end
+
+      readers.each do |io|
+        if io == @__r__
+          # clean up wakeups
+          @__r__.read(@__r__.stat.size)
+        else
+          monitor = io.closed? ? @readers.delete(io) : @readers[io]
+          next unless monitor
+          monitor.readiness = writers.delete(io) ? :rw : :r
+          yield monitor
+        end
+      end if readers
+
+      writers.each do |io|
+        monitor = io.closed? ? @writers.delete(io) : @writers[io]
+        next unless monitor
+        # don't double run this, the last iteration might have run this task already
+        monitor.readiness = :w
+        yield monitor
+      end if writers
+    end
+
+    # Closes the selector.
+    #
+    def close
+      return if @closed
+      @__r__.close
+      @__w__.close
+    rescue IOError
+    ensure
+      @closed = true
+    end
+
+    # interrupts the select call.
+    def wakeup
+      @__w__.write_nonblock("\0", exception: false)
+    end
+  end
+end
+