Faraday parallelism support

[iMacTia]

Introduction

One of the objectives of our Faraday 3.0 Wishlist is to "Revisit pipelining or parallel requests".

With the advent of Async, Ruby 3.0 has really pressed on the accelerator when it comes to parallelism and concurrency, and the community is expecting more and more libraries to make this a big focus in the coming months and years.

Faraday has always supported requests parallelism for compatible adapters via a "parallel manager", but this was written years ago and does not really reflect the current state of the Ruby world.
We should take this opportunity to revisit how parallelism is managed in faraday, how middleware/adapters can opt-in to it and how developers can easily use it in their projects.

Async and a new concurrency style

Async has really shaken the Ruby world, so much so that it gained official support in Ruby 3.0.
The promises are really high: allow Ruby engineers to run concurrent tasks in Ruby without having to worry about callbacks or making distinctions between sync and async methods (JS developers will know!).
Interestingly enough, Faraday's default adapter net_http already supports Async, as the following snippet shows:

require 'faraday'
require 'async'
require 'benchmark'

def call(conn, path: '/', params: {})
  Async do
    res = conn.get(path, params)
  end
end

conn = Faraday.new('https://httpbingo.org')

b1 = Benchmark.measure do
  100.times { |i| call(conn, params: { q: i }) }
end.real

b2 = Benchmark.measure do
  Async do
    100.times { |i| call(conn, params: { q: i }) }
  end
end.real

puts "#{(b1/b2).round(2)}x faster!"

This opens up to the possibility of supporting parallel requests without even the need to use complex "parallel managers" or similar hacks. On the flip side, this relies on the underlying library support for Async, which may not be offered by all HTTP libraries.

Issues with the current parallel interface

The current interface looks simple and powerful at first glance:

conn = Faraday.new(:url => "http://coolness.com") do |faraday|
  faraday.adapter :typhoeus
end

conn.in_parallel do
  response1 = conn.get('/one')
  response2 = conn.get('/two')
end

However it does have some setbacks:

• It only works with some adapters: there's no easy way to tell if an adapter support parallel request, except by calling the support_parallel? method or reading it's documentation.
• It's connection-based, so it does not currently support cross-domain parallel requests
• Some middleware (e.g. instrumentation, retry) have not been built specifically for it. The current dual interface of call or on_request/on_complete is cause for confusion.

An alternative implementation

Ideally a better, easier implementation would bring the following benefits:

• Universal: it works with all, or most, adapters
• Easy and transparent: it doesn't require to instantiate "parallel managers" or other complex flows
• Transparent and Callback-free: it relies on blocks and doesn't require developers to work with callbacks-madness

cc @avmnu-sng, @BuonOmo, @ioquatix

[ioquatix]

https://github.com/socketry/async-http-faraday already solves this problem with no changes to the current interface. The only complexity is the per-thread mapping of connections since we don't support multi-thread reactor in Async as a matter of design.

I would personally just remove any kind of concurrency/parallelism "manager"/in_parallel and let the implementation take care of it on a case by case basis.

[iMacTia]

I feel a bit confused now and would be great if you could clarify @ioquatix.
The snippet I shared above already allows parallel requests using the standard net_http adapter.
What is the difference using async-http and when would that make more sense?

[ioquatix]

It supports HTTP/2 and persistent connections.

[ioquatix]

Async and a new concurrency style

Regarding this, the non-Async specific way to do this is Fiber.schedule{} which is very similar to Async{}. So, this is the Ruby official way to do concurrency and if we are forward looking, it would make sense for Faraday to adopt this model since it's part of Ruby's standard interface now.

[iMacTia]

That was my thought as well, but I wonder if this will be compatible with "outsourcing" libraries like Patron and Typhoeus that rely on external processes running outside of ruby-land? Or do they use fibers/async as well?

[ioquatix]

The Fiber scheduler is safe to use with threading primitives so as long as those libraries use Ruby threading primitives it should be okay.

[avmnu-sng]

@ioquatix Isn't async-http-faraday is another adapter? We should look into supporting concurrent requests but not get into the internals of each adapter (the parallel manager). The async-http has features like Async::Barrier and Async::Semaphore which can be utilized which gives a nice control over parallel requests and no. of simultaneous requests.

[iMacTia]

Having the option to configure concurrency via a Faraday API would totally make sense, I think this is also missing from the current implementation (hence you need to explicitly provide a Hydra instance when using typhoeus in order to configure it).
But would the snippet above work with other adapters? Does typhoeus support running parallel transactions using Async instead of Hydra?

[iMacTia]

unless that in_parallel would need to be defined on the adapter? In which case it would be the return to the parallel_manager or some other type of abstraction that would allow adapters to provide the implementation

[ioquatix]

Async::HTTP uses the concurrency as recommended by the server connection, i.e. HTTP/1 is hard coded I believe and HTTP/2 is advertised by the stream limit, so there is usually no need for users to be explicit about it, except if you are dealing with rate limiting etc - of which the standard rate limiting could be handled internally.

I guess if you wanted to explicitly handle it, an interface would be required to be generic.

[iMacTia]

I see, that makes sense. Parallel stuff is a bit outside of my comfort area, so I'm not sure why the existing libraries would normally allow you to set a concurrency limit. But I'd expect there are some cases where you need to do that to ensure you don't saturate your server/application resources (e.g. open files limit).

I'd assume this might be a requirement though considering it is possible today, and would rather design an interface that allows this type of configuration.

[ioquatix]

I'd assume this might be a requirement though considering it is possible today, and would rather design an interface that allows this type of configuration.'

Most users won't know how to set this and at worst get it totally wrong.

[BuonOmo]

A note on implementation. Some times ago, I started working on making the retry middleware work in parallel. However, the major issue I encountered was when I was starting to add multiple middlewares, especially the Raise Error Middleware is a pain since it shortcuts any other one and forces another middleware implementation to be aware of its existence.

I think this tight coupling makes developers less likely to contribute and errors more likely to happen due to the improved complexity. And of course, maintainance is also harder.

IMHO, this is mostly due to:

• call or on_request/on_complete vs call mentioned by @iMacTia
• middlewares that leak their behaviours on others (such as RaiseError)

The point being: nowaday, Faraday is a bit too complete to support parallelism AND keep its simplicity. I think you mentioned at some point that you wanted to refactor the middleware chain. This would be a better time to introduce parallelism IMHO.

[BuonOmo]

For example I'd be curious to understand why the RaiseError middleware was an issue,

It was for me because its presence forced my parallel aware retry middleware implementation to either be aware of it (and rescue everywhere) or just fail to retry correctly.

Here's a patch I had been working on (feel free to re-use) on which you can see the retry_block method is just a pain (IMHO):

diff --git a/Gemfile b/Gemfile
index 80777df..876c88f 100644
--- a/Gemfile
+++ b/Gemfile
@@ -16,6 +16,7 @@ group :development, :test do
   gem 'rspec_junit_formatter', '~> 0.4'
   gem 'simplecov'
   gem 'webmock', '~> 3.4'
+  gem "typhoeus"
 end
 
 group :development, :lint do
diff --git a/docs/middleware/request/retry.md b/docs/middleware/request/retry.md
index f111504..1582d2e 100644
--- a/docs/middleware/request/retry.md
+++ b/docs/middleware/request/retry.md
@@ -67,6 +67,19 @@ retry_options = {
 }
 BLOB```
 
+⚠️ Order matters when using this middleware combined with the [RaiseError][raise_error] middleware:
+
+```ruby
+# This won't retry if an error is raised.
+faraday.request :retry
+faraday.response :raise_error
+
+# This will retry and then raise.
+faraday.response :raise_error
+faraday.request :retry
+```
+
+
 #### Specify on which response statuses to retry
 
 By default the `Retry` middleware will only retry the request if one of the expected exceptions arise.
diff --git a/lib/faraday/adapter/test.rb b/lib/faraday/adapter/test.rb
index 313b141..c900799 100644
--- a/lib/faraday/adapter/test.rb
+++ b/lib/faraday/adapter/test.rb
@@ -237,6 +237,33 @@ module Faraday
       def call(env)
         super
 
+        if env.parallel?
+          env.parallel_manager.queue self, env
+        else
+          perform_request(env)
+        end
+
+        @app.call(env)
+      end
+
+      self.supports_parallel = true
+      def self.setup_parallel_manager(_options = nil)
+        Class.new do
+          def initialize
+            @envs = []
+          end
+          def queue(instance, env)
+            @envs << [instance, env]
+          end
+          def run
+            while (instance, env = @envs.shift)
+              instance.perform_request(env)
+            end
+          end
+        end.new
+      end
+
+      def perform_request(env)
         env.request.params_encoder ||= Faraday::Utils.default_params_encoder
         env[:params] = env.params_encoder.decode(env[:url].query) || {}
         stub, meta = stubs.match(env)
@@ -254,8 +281,7 @@ module Faraday
             stub.block.call(env, meta)
           end
         save_response(env, status, body, headers)
-
-        @app.call(env)
+        env.response.finish(env) if env.parallel?
       end
     end
   end
diff --git a/lib/faraday/request/instrumentation.rb b/lib/faraday/request/instrumentation.rb
index c03ba1e..4652718 100644
--- a/lib/faraday/request/instrumentation.rb
+++ b/lib/faraday/request/instrumentation.rb
@@ -13,7 +13,7 @@ module Faraday
 
         # @return [Class]
         def instrumenter
-          self[:instrumenter] ||= ActiveSupport::Notifications
+          self[:instrumenter] ||= ActiveSupport::Notifications.instrumenter
         end
       end
 
diff --git a/lib/faraday/request/retry.rb b/lib/faraday/request/retry.rb
index 3ba50b7..29f6260 100644
--- a/lib/faraday/request/retry.rb
+++ b/lib/faraday/request/retry.rb
@@ -87,6 +87,22 @@ module Faraday
         end
       end
 
+      # Fifo is a simple FIFO queue, that will be used to stack our retries
+      # in parallel mode.
+      class Fifo
+        extend Forwardable
+
+        def initialize
+          @ary = []
+        end
+
+        def_delegator :@ary, :first, :peek
+        def_delegator :@ary, :shift, :deq
+        def_delegator :@ary, :<<, :enq
+
+        def_delegators :@ary, :size, :empty?
+      end
+
       # @param app [#call]
       # @param options [Hash]
       # @option options [Integer] :max (2) Maximum number of retries
@@ -123,6 +139,8 @@ module Faraday
         super(app)
         @options = Options.from(options)
         @errmatch = build_exception_matcher(@options.exceptions)
+        @finished_count = 0
+        @retries_queue = Fifo.new
       end
 
       def calculate_sleep_amount(retries, env)
@@ -139,32 +157,39 @@ module Faraday
       end
 
       # @param env [Faraday::Env]
-      def call(env)
-        retries = @options.max
-        request_body = env[:body]
-        begin
-          # after failure env[:body] is set to the response body
-          env[:body] = request_body
-          @app.call(env).tap do |resp|
-            if @options.retry_statuses.include?(resp.status)
-              raise Faraday::RetriableResponse.new(nil, resp)
+      def call(env, retries = @options.max, request_body = nil)
+        # We must keep track of the original request body to make sure it is
+        # retried correctly.
+        request_body ||= env[:body]
+        # after failure env[:body] is set to the response body.
+        env[:body] = request_body
+        retry_block(env, retries, request_body) do
+          @app.call(env).on_complete do |resp|
+            # We must repeat the retry logic here for usage with #in_parallel.
+            retry_block(env, retries, request_body) do
+              if @options.retry_statuses.include?(resp.status)
+                raise Faraday::RetriableResponse.new(nil, resp)
+              end
             end
           end
-        rescue @errmatch => e
-          if retries.positive? && retry_request?(env, e)
-            retries -= 1
-            rewind_files(request_body)
-            @options.retry_block.call(env, @options, retries, e)
-            if (sleep_amount = calculate_sleep_amount(retries + 1, env))
-              sleep sleep_amount
-              retry
-            end
+        end
+      end
+
+      def retry_block(env, retries, request_body)
+        yield
+      rescue @errmatch => e
+        if retries.positive? && retry_request?(env, e)
+          rewind_files(request_body)
+          @options.retry_block.call(env, @options, retries - 1, e)
+          if (sleep_amount = calculate_sleep_amount(retries, env))
+            sleep sleep_amount
+            call(env, retries - 1, request_body)
           end
+        end
 
-          raise unless e.is_a?(Faraday::RetriableResponse)
+        raise unless e.is_a?(Faraday::RetriableResponse)
 
-          e.response
-        end
+        e.response
       end
 
       # An exception matcher for the rescue clause can usually be any object
diff --git a/lib/faraday/response.rb b/lib/faraday/response.rb
index 505e4ea..4385112 100644
--- a/lib/faraday/response.rb
+++ b/lib/faraday/response.rb
@@ -33,6 +33,10 @@ module Faraday
       finished? ? env.body : nil
     end
 
+    def time
+      finished? ? env.time : nil
+    end
+
     def finished?
       !!env
     end
diff --git a/spec/faraday/request/retry_spec.rb b/spec/faraday/request/retry_spec.rb
index 5891b73..ece878b 100644
--- a/spec/faraday/request/retry_spec.rb
+++ b/spec/faraday/request/retry_spec.rb
@@ -239,4 +239,24 @@ RSpec.describe Faraday::Request::Retry do
       it { expect(times_called).to eq(1) }
     end
   end
+
+  context 'when used in parallel' do
+    context 'when no exception raised' do
+      let(:options) { [{ max: 1, retry_statuses: 429 }] }
+
+      before { conn.in_parallel { conn.get('/unstable') } }
+
+      context 'and response code is in retry_statuses' do
+        let(:callback) { -> { [429, {}, ''] } }
+
+        it { expect(times_called).to eq(2) }
+      end
+
+      context 'and response code is not in retry_statuses' do
+        let(:callback) { -> { [503, {}, ''] } }
+
+        it { expect(times_called).to eq(1) }
+      end
+    end
+  end
 end

Two quick notes:

• the BLOB in the patch is just to avoid markdown syntax error in github, remove to apply
• the design idea is fairly simple: whenever a request is failed, we add it to a retry queue, if we have workers available and backoff duration is expired, we retry.

[ioquatix]

I don't think retry logic should be middleware, because it's behaviour is too specific to the underlying request structure (can you replay request body for example).

[iMacTia]

Thanks for sharing this @BuonOmo. I had a quick look and I'm struggling to see where the Fifo queue is being used. I can see it's initialised into @retries_queue = Fifo.new, but I don't see any usage of that variable aftwerwards 🤔.

Apart from that, the implementation makes sense, and I also see now what you meant with the raise_error middleware conflicts 👍

[BuonOmo]

@ioquatix that would make things way simpler, and I've worked on parallel requests enough to make it a simple separate gem (that takes advantage of typhoeus' hydra for instance). However, the only point you mentioned (can the request body be rewinded) is quite easy to solve IMHO : fail when using #rewind

@iMacTia the patch is as I left my work in progress, I think I added that FIFO before taking into account the raise middleware

[ioquatix]

The faraday adaptor for async-http has complete support for retry as an example data point, without requiring any extra overhead/interface.

[BuonOmo]

Found back the thread were I was weighting creating a gem dedicated to parallel requests: typhoeus/typhoeus#182 (comment)