refact: Use Async

Author: alexrails

Gemfile

@@ -4,4 +4,3 @@ gem "async"
 gem "falcon", "0.44.0" # use different version on your own risk; 0.44.0 worked
 gem "sinatra"
 gem "faraday"
-gem "typhoeus"

Gemfile.lock

@@ -30,8 +30,6 @@ GEM
       fiber-annotation
       fiber-local (~> 1.1)
       json
-    ethon (0.16.0)
-      ffi (>= 1.15.0)
     falcon (0.44.0)
       async
       async-container (~> 0.17)
@@ -49,8 +47,6 @@ GEM
       faraday-net_http (>= 2.0, < 3.2)
     faraday-net_http (3.1.0)
       net-http
-    ffi (1.17.1)
-    ffi (1.17.1-x86_64-darwin)
     fiber-annotation (0.2.0)
     fiber-local (1.1.0)
       fiber-storage
@@ -96,8 +92,6 @@ GEM
     tilt (2.3.0)
     timers (4.3.5)
     traces (0.11.1)
-    typhoeus (1.4.1)
-      ethon (>= 0.9.0)
     uri (0.13.0)
 
 PLATFORMS
@@ -109,7 +103,6 @@ DEPENDENCIES
   falcon (= 0.44.0)
   faraday
   sinatra
-  typhoeus
 
 BUNDLED WITH
    2.5.11

client.rb

@@ -1,5 +1,8 @@
 require 'openssl'
-require 'typhoeus'
+require 'faraday'
+require 'async'
+require 'async/barrier'
+require 'async/semaphore'
 
 OpenSSL::SSL::VERIFY_PEER = OpenSSL::SSL::VERIFY_NONE
 
@@ -14,99 +17,81 @@
 #   - работает 1 секунду
 #   - одновременно можно запускать не более одного
 #
-def make_request(endpoint, value)
-  Typhoeus::Request.new(
-    "https://localhost:9292/#{endpoint}",
-    method: :get,
-    params: { value: value },
-    ssl_verifypeer: false,
-    timeout: 10
-  )
+def a(value)
+  puts "https://localhost:9292/a?value=#{value}"
+  Faraday.get("https://localhost:9292/a?value=#{value}").body
 end
 
+def b(value)
+  puts "https://localhost:9292/b?value=#{value}"
+  Faraday.get("https://localhost:9292/b?value=#{value}").body
+end
+
+def c(value)
+  puts "https://localhost:9292/c?value=#{value}"
+  Faraday.get("https://localhost:9292/c?value=#{value}").body
+end
+
+# Референсное решение, приведённое ниже работает правильно, занимает ~19.5 секунд
+# Надо сделать в пределах 7 секунд
+
 def collect_sorted(arr)
-  arr.sort.join('-')
+  arr.compact.sort.join('-')
 end
 
 start = Time.now
 
-hydra = Typhoeus::Hydra.new(max_concurrency: 6)
-responses = {}
+result = Async do
+  barrier = Async::Barrier.new
+  semaphore_a = Async::Semaphore.new(3, parent: barrier)
+  semaphore_b = Async::Semaphore.new(2, parent: barrier)
+  semaphore_c = Async::Semaphore.new(1, parent: barrier)
+
+  a_results = {
+    batch1: Array.new(3),
+    batch2: Array.new(3),
+    batch3: Array.new(3)
+  }
+  b_results = Array.new(3)
+  c_results = Array.new(3)
+
+  3.times do |i|
+    semaphore_a.async do
+      a_results[:batch1][i] = a(11 + i)
+      a_results[:batch2][i] = a(21 + i)
+    end
+  end
 
-a_batches = {
-  1 => [11, 12, 13].map { |v| make_request('a', v) },
-  2 => [21, 22, 23].map { |v| make_request('a', v) },
-  3 => [31, 32, 33].map { |v| make_request('a', v) }
-}
-b_requests = [1, 2, 3].map { |v| make_request('b', v) }
+  2.times do |i|
+    semaphore_b.async { b_results[i] = b(i + 1) }
+  end
 
-a_batches.each do |batch, reqs|
-  reqs.each do |req|
-    req.on_complete do |response|
-      responses["a#{batch}_#{req.options[:params][:value]}".to_sym] = response.body
-    end
+  barrier.wait
+
+  ab1 = "#{collect_sorted(a_results[:batch1])}-#{b_results[0]}"
+  ab2 = "#{collect_sorted(a_results[:batch2])}-#{b_results[1]}"
+
+  semaphore_c.async do
+    c_results[0] = c(ab1)
+    c_results[1] = c(ab2)
   end
-end
 
-b_requests.each_with_index do |req, idx|
-  req.on_complete do |response|
-    responses["b_#{idx + 1}".to_sym] = response.body
+  3.times do |i|
+    semaphore_a.async { a_results[:batch3][i] = a(31 + i) }
   end
-end
-# Queue first batch
-a_batches[1].each { |req| hydra.queue(req) }
-[b_requests[0], b_requests[1]].each { |req| hydra.queue(req) }
-# Run first batch
-hydra.run
-
-# Queue second batch
-ab1 = "#{collect_sorted(a_batches[1].map! { |req| responses["a1_#{req.options[:params][:value]}".to_sym] })}-#{responses[:b_1]}"
-puts "AB1 = #{ab1}"
-c1_req = make_request('c', ab1)
-c1_req.on_complete { |response| responses[:c1] = response.body }
-hydra.queue(c1_req)
-
-a_batches[2].each { |req| hydra.queue(req) }
-hydra.queue(b_requests[2])
-# Run second batch
-hydra.run
-
-ab2 = "#{collect_sorted(a_batches[2].map! { |req| responses["a2_#{req.options[:params][:value]}".to_sym] })}-#{responses[:b_2]}"
-puts "C1 = #{responses[:c1]}"
-puts "AB2 = #{ab2}"
-
-# Queue third batch
-c2_req = make_request('c', ab2)
-c2_req.on_complete { |response| responses[:c2] = response.body }
-hydra.queue(c2_req)
-a_batches[3].each { |req| hydra.queue(req) }
-
-# Run third batch
-hydra.run
-
-puts "C2 = #{responses[:c2]}"
-ab3 = "#{collect_sorted(a_batches[3].map! { |req| responses["a3_#{req.options[:params][:value]}".to_sym] })}-#{responses[:b_3]}"
-puts "AB3 = #{ab3}"
-
-# Queue fourth batch
-c3_req = make_request('c', ab3)
-c3_req.on_complete { |response| responses[:c3] = response.body }
-hydra.queue(c3_req)
-
-# Run fourth batch
-hydra.run
-
-puts "C3 = #{responses[:c3]}"
-c123 = collect_sorted([responses[:c1], responses[:c2], responses[:c3]])
-
-# Final processing
-final_req = make_request('a', c123)
-final_req.on_complete { |response| responses[:final] = response.body }
-hydra.queue(final_req)
-hydra.run
 
-puts "FINISHED in #{Time.now - start}s."
-puts "RESULT = #{responses[:final]}" # 0bbe9ecf251ef4131dd43e1600742cfb
+  semaphore_b.async { b_results[2] = b(3) }
+
+  barrier.wait
 
-# FINISHED in 7.064975s.
-# RESULT = 0bbe9ecf251ef4131dd43e1600742cfb
+  ab3 = "#{collect_sorted(a_results[:batch3])}-#{b_results[2]}"
+  semaphore_c.async { c_results[2] = c(ab3) }
+
+  barrier.wait
+
+  c123 = collect_sorted(c_results)
+  a(c123)
+end.wait
+
+puts "FINISHED in #{Time.now - start}s."
+puts "RESULT = #{result}" # 0bbe9ecf251ef4131dd43e1600742cfb