|
# 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 |
This comment has been minimized.
elct9620 commentedAug 17, 2018