async-http-fiber.rb

## async-http-fiber.rb
# frozen_string_literal: true

require 'net/http'
require 'fiber'
require 'uri'
require 'openssl'
require 'socket'
require 'pp'

require 'benchmark/ips'
require 'benchmark/memory'

require 'memory_profiler'

#rubocop:disable Metrics/LineLength
# :nodoc:
class Selector
  def initialize
    @readable = {}
  end

  def wait_readable(io)
    Fiber.new do
      @readable[io] = Fiber.current
      Fiber.yield
      yield
    end.resume
  end

  def resume
    readable, = IO.select(@readable.keys)
    readable.each do |io|
      fiber = @readable.delete(io)
      fiber.resume
    end
  end

  def run
    resume until @readable.empty?
  end
end

# :nodoc:
class AsyncHTTP
  # :nodoc:
  class Connection
    BODY_SPLIT = "\r\n\r\n".freeze

    attr_reader :hostname, :port, :ssl, :ready, :selector

    def initialize(hostname, port, selector)
      @selector = selector
      @hostname = hostname
      @port = port

      @socket = TCPSocket.new hostname, port
      @ssl = OpenSSL::SSL::SSLSocket.new(@socket)

      @ready = false
      @buffer = +''

      @count = 0
      @requests = 0
      @chunked = false
    end

    def connect
      return if ready
      @ready = true
      ssl.sync_close = true
      ssl.connect
    end

    def start(request)
      @requests += 1
      ssl.write "GET #{request.request_uri} HTTP/1.1\r\nHost: #{request.hostname}\r\n\r\n"
    end

    def wait_response(&block)
      loop do
        break if @count == @requests
        @buffer ||= +''
        @buffer << ssl.read_nonblock(4096)
        if @buffer.include?(BODY_SPLIT) && @chunk == false
          header, @buffer = buf.split(BODY_SPLIT)
          @chunked = true if header.include?('chunked')
        end

        if @chunk == true
          if @buffer.include?("0#{BODY_SPLIT}")
            @chunk = false
            @count += 1
            response, @buffer = @buffer.split("0#{BODY_SPLIT}")
            # TODO: Combine Chunk
            yield response
          end
        else
          @count += 1
          yield @buffer
        end
      end
    rescue IO::WaitReadable
      return if @count == @requests
      selector.wait_readable(ssl) do
        wait_response(&block)
      end
    end

    def close
      return unless ready
      ssl.close
    end
  end

  include Enumerable

  attr_reader :selector

  class << self
    def start(&block)
      new.start(&block)
    end
  end

  def initialize
    @selector = Selector.new
    @pool = {}
  end

  def start(&block)
    instance_exec(self, &block)
    @pool.values.each(&:close)
  end

  def request(request)
    conn = @pool[request.hostname] ||= Connection.new(request.hostname, request.port, selector)
    conn.connect
    conn.start(request)
  end

  def each(&block)
    @pool.values.each do |conn|
      conn.wait_response(&block)
    end
    selector.run
  end
end
#rubocop:enable Metrics/LineLength

uris = [
  'https://api.bitfinex.com/v1/pubticker/BTCUSD',
  'https://api.bitfinex.com/v1/pubticker/ETHUSD',
  'https://api.hitbtc.com/api/2/public/ticker/BTCUSD',
  'https://api.hitbtc.com/api/2/public/ticker/ETHUSD',
  'https://bittrex.com/api/v1.1/public/getmarketsummary?market=USD-BTC',
  'https://bittrex.com/api/v1.1/public/getmarketsummary?market=USD-ETH',
  'https://api.binance.com/api/v1/ticker/24hr?symbol=BTCUSDT',
  'https://api.binance.com/api/v1/ticker/24hr?symbol=ETHUSDT'
]

Benchmark.ips do |x|
  x.report('Fiber') do |c|
    c.times do
      AsyncHTTP.start do |http|
        uris.each do |uri|
          http.request(URI(uri))
        end

        http.each do |body|
          # TODO
        end
      end

      sleep 5
    end
  end

  x.report('Net::HTTP') do |c|
    c.times do
      uris.each do |uri|
        Net::HTTP.get(URI(uri))
      end

      sleep 5
    end
  end

  x.report('[P] Net::HTTP') do |c|
    c.times do
      uris.map do |uri|
        Thread.new { Net::HTTP.get(URI(uri)) }
      end.each(&:join)

      sleep 5
    end
  end

  x.compare!
end

Benchmark.memory do |x|
  x.report('Fiber') do
    AsyncHTTP.start do |http|
      uris.each do |uri|
        http.request(URI(uri))
      end

      http.each do |body|
        # TODO
      end
    end
  end

  sleep 5

  x.report('Net::HTTP') do
    uris.each do |uri|
      Net::HTTP.get(URI(uri))
    end
  end

  sleep 5

  x.report('[P] Net::HTTP') do
    uris.map do |uri|
      Thread.new { Net::HTTP.get(URI(uri)) }
    end.each(&:join)
  end

  x.compare!
end

# MemoryProfiler.report do
#   AsyncHTTP.start do |http|
#     uris.each do |uri|
#       http.request(URI(uri))
#     end
#
#     http.each do |body|
#       # TODO
#     end
#   end
# end.pretty_print