Debounce producer dispatch for throughput (oban-bg#136)

Producers may become overloaded under high insert/execute load when
processes complete quickly. This is due to a flood of `DOWN` messages
from completed jobs, which triggers an immediate dispatch. Under high
load the "demand" (concurrency limit - running jobs) hovers around 1, so
every dispatch only retrieves one job. This is wasteful, unnecessarily
hard on the database, and floods the producer's message queue.

This introduces dispatch cooldown for each producer. Rather than
fetching jobs every time dispatch is called it waits a small period of
time (5ms by default) to allow demand to accumulate. According to
benchmarks this has a huge impact on overall throughput: without
cooldown a producer maxes out around ~1000 jobs/sec, with a 5ms
debounce producers can handle ~5000 jobs/sec and have an empty message
queue.

Author: sorentwo

CHANGELOG.md

@@ -13,6 +13,19 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   than "Etc/UTC". Using a custom timezone requires a timezone database such as
   tzdata.
 
+- [Oban] Add `dispatch_cooldown` option to configure the minimum time between
+  a producer fetching more jobs to execute.
+
+### Changed
+
+- [Oban.Queue.Producer] Introduce "dispatch cooldown" as a way to debounce
+  repeatedly fetching new jobs. Repeated fetching floods the producer's message
+  queue and forces the producer to repeatedly fetch one job at a time, which is
+  not especially efficient. Debounced fetching is much more efficient for the
+  producer and the database, increasing maximum jobs/sec throughput so that it
+  scales linearly with a queue's concurrency settings (up to what the database
+  can handle).
+
 ## [0.12.1] — 2019-12-13
 
 ### Fixed

README.md

@@ -216,7 +216,7 @@ and job dispatching altogether when testing:
 
 ```elixir
 # config/test.exs
-config :my_app, Oban, queues: false, prune: :disabled
+config :my_app, Oban, crontab: false, queues: false, prune: :disabled
 ```
 
 Without dispatch and pruning disabled Ecto will raise constant ownership errors

lib/oban.ex

@@ -71,7 +71,7 @@ defmodule Oban do
   defmodule MyApp.Business do
     use Oban.Worker, queue: "events", max_attempts: 10
 
-    @impl Worker
+    @impl Oban.Worker
     def perform(%{"id" => id}, _job) do
       model = MyApp.Repo.get(MyApp.Business.Man, id)
 
@@ -513,7 +513,10 @@ defmodule Oban do
   alias Oban.Queue.Supervisor, as: QueueSupervisor
 
   @type option ::
-          {:name, module()}
+          {:circuit_backoff, timeout()}
+          | {:crontab, [Config.cronjob()]}
+          | {:dispatch_cooldown, pos_integer()}
+          | {:name, module()}
           | {:node, binary()}
           | {:poll_interval, pos_integer()}
           | {:prefix, binary()}
@@ -522,7 +525,10 @@ defmodule Oban do
           | {:prune_limit, pos_integer()}
           | {:queues, [{atom(), pos_integer()}]}
           | {:repo, module()}
+          | {:rescue_after, pos_integer()}
+          | {:rescue_interval, pos_integer()}
           | {:shutdown_grace_period, timeout()}
+          | {:timezone, Calendar.time_zone()}
           | {:verbose, false | Logger.level()}
 
   @type queue_name :: atom() | binary()
@@ -577,11 +583,20 @@ defmodule Oban do
   ### Twiddly Options
 
   Additional options used to tune system behaviour. These are primarily useful for testing or
-  troubleshooting and shouldn't be changed usually.
+  troubleshooting and don't usually need modification.
 
   * `:circuit_backoff` — the number of milliseconds until queries are attempted after a database
     error. All processes communicating with the database are equipped with circuit breakers and
     will use this for the backoff. Defaults to `30_000ms`.
+  * `:dispatch_cooldown` — the minimum number of milliseconds a producer will wait before fetching
+    and running more jobs. A slight cooldown period prevents a producer from flooding with
+    messages and thrashing the database. The cooldown period _directly impacts_ a producer's
+    throughput: jobs per second for a single queue is calculated by `(1000 / cooldown) * limit`.
+    For example, with a `5ms` cooldown and a queue limit of `25` a single queue can run 2,500
+    jobs/sec.
+
+    The default is `5ms` and the minimum is `1ms`, which is likely faster than the database can
+    return new jobs to run.
   * `:poll_interval` - the number of milliseconds between polling for new jobs in a queue. This
     is directly tied to the resolution of _scheduled_ jobs. For example, with a `poll_interval` of
     `5_000ms`, scheduled jobs are checked every 5 seconds. The default is `1_000ms`.

lib/oban/config.ex

@@ -12,6 +12,7 @@ defmodule Oban.Config do
   @type t :: %__MODULE__{
           circuit_backoff: timeout(),
           crontab: [cronjob()],
+          dispatch_cooldown: pos_integer(),
           name: atom(),
           node: binary(),
           poll_interval: pos_integer(),
@@ -33,6 +34,7 @@ defmodule Oban.Config do
   @enforce_keys [:node, :repo]
   defstruct circuit_backoff: :timer.seconds(30),
             crontab: [],
+            dispatch_cooldown: 5,
             name: Oban,
             node: nil,
             poll_interval: :timer.seconds(1),
@@ -103,6 +105,12 @@ defmodule Oban.Config do
     end
   end
 
+  defp validate_opt!({:dispatch_cooldown, period}) do
+    unless is_integer(period) and period > 0 do
+      raise ArgumentError, "expected :dispatch_cooldown to be a positive integer"
+    end
+  end
+
   defp validate_opt!({:name, name}) do
     unless is_atom(name) do
       raise ArgumentError, "expected :name to be a module or atom"

lib/oban/queue/producer.ex

@@ -22,6 +22,8 @@ defmodule Oban.Queue.Producer do
     @enforce_keys [:conf, :foreman, :limit, :nonce, :queue]
     defstruct [
       :conf,
+      :cooldown_ref,
+      :dispatched_at,
       :foreman,
       :limit,
       :name,
@@ -102,29 +104,14 @@ defmodule Oban.Queue.Producer do
   end
 
   @impl GenServer
-  def handle_info(:poll, %State{conf: conf} = state) do
-    conf.repo.checkout(fn ->
-      state
-      |> deschedule()
-      |> pulse()
-      |> send_poll_after()
-      |> dispatch()
-    end)
-  end
-
-  def handle_info(:rescue, %State{conf: conf} = state) do
-    conf.repo.checkout(fn ->
-      state
-      |> rescue_orphans()
-      |> send_rescue_after()
-      |> dispatch()
-    end)
-  end
-
   def handle_info({:DOWN, ref, :process, _pid, _reason}, %State{running: running} = state) do
     dispatch(%{state | running: Map.delete(running, ref)})
   end
 
+  def handle_info(:dispatch, %State{} = state) do
+    dispatch(state)
+  end
+
   def handle_info({:notification, insert(), payload}, %State{queue: queue} = state) do
     case payload do
       %{"queue" => ^queue} -> dispatch(state)
@@ -162,6 +149,25 @@ defmodule Oban.Queue.Producer do
     dispatch(state)
   end
 
+  def handle_info(:poll, %State{conf: conf} = state) do
+    conf.repo.checkout(fn ->
+      state
+      |> deschedule()
+      |> pulse()
+      |> send_poll_after()
+      |> dispatch()
+    end)
+  end
+
+  def handle_info(:rescue, %State{conf: conf} = state) do
+    conf.repo.checkout(fn ->
+      state
+      |> rescue_orphans()
+      |> send_rescue_after()
+      |> dispatch()
+    end)
+  end
+
   def handle_info(:reset_circuit, state) do
     {:noreply, open_circuit(state)}
   end
@@ -230,12 +236,14 @@ defmodule Oban.Queue.Producer do
     state
   end
 
-  defp pulse(%State{conf: conf} = state) do
+  defp pulse(%State{conf: conf, running: running} = state) do
+    running_ids = for {_ref, {%_{id: id}, _pid}} <- running, do: id
+
     args =
       state
       |> Map.take([:limit, :nonce, :paused, :queue, :started_at])
       |> Map.put(:node, conf.node)
-      |> Map.put(:running, running_job_ids(state))
+      |> Map.put(:running, running_ids)
 
     Query.insert_beat(conf, args)
 
@@ -256,29 +264,56 @@ defmodule Oban.Queue.Producer do
     {:noreply, state}
   end
 
-  defp dispatch(%State{conf: conf, foreman: foreman} = state) do
-    %State{limit: limit, nonce: nonce, queue: queue, running: running} = state
+  defp dispatch(%State{} = state) do
+    cond do
+      dispatch_now?(state) ->
+        %State{conf: conf, limit: limit, nonce: nonce, queue: queue, running: running} = state
 
-    started_jobs =
-      for job <- fetch_jobs(conf, queue, nonce, limit - map_size(running)), into: %{} do
-        {:ok, pid} = DynamicSupervisor.start_child(foreman, Executor.child_spec(job, conf))
+        running =
+          conf
+          |> fetch_jobs(queue, nonce, limit - map_size(running))
+          |> start_jobs(state)
+          |> Map.merge(running)
 
-        {Process.monitor(pid), {job, pid}}
-      end
+        {:noreply, %{state | cooldown_ref: nil, dispatched_at: system_now(), running: running}}
+
+      cooldown_available?(state) ->
+        %State{conf: conf, dispatched_at: dispatched_at} = state
+
+        dispatch_after = system_now() - dispatched_at + conf.dispatch_cooldown
+        cooldown_ref = Process.send_after(self(), :dispatch, dispatch_after)
 
-    {:noreply, %{state | running: Map.merge(running, started_jobs)}}
+        {:noreply, %{state | cooldown_ref: cooldown_ref}}
+
+      true ->
+        {:noreply, state}
+    end
   rescue
     exception in trip_errors() -> {:noreply, trip_circuit(exception, state)}
   end
 
+  defp dispatch_now?(%State{dispatched_at: nil}), do: true
+
+  defp dispatch_now?(%State{conf: conf, dispatched_at: dispatched_at}) do
+    system_now() > dispatched_at + conf.dispatch_cooldown
+  end
+
+  defp cooldown_available?(%State{cooldown_ref: ref}), do: is_nil(ref)
+
   defp fetch_jobs(conf, queue, nonce, count) do
     case Query.fetch_available_jobs(conf, queue, nonce, count) do
       {0, nil} -> []
       {_count, jobs} -> jobs
     end
   end
 
-  defp running_job_ids(%State{running: running}) do
-    for {_ref, {%_{id: id}, _pid}} <- running, do: id
+  defp start_jobs(jobs, %State{conf: conf, foreman: foreman}) do
+    for job <- jobs, into: %{} do
+      {:ok, pid} = DynamicSupervisor.start_child(foreman, Executor.child_spec(job, conf))
+
+      {Process.monitor(pid), {job, pid}}
+    end
   end
+
+  defp system_now, do: System.monotonic_time(:millisecond)
 end

lib/oban/worker.ex

@@ -13,7 +13,7 @@ defmodule Oban.Worker do
       defmodule MyApp.Workers.Business do
         use Oban.Worker, queue: "events", max_attempts: 10, unique: [period: 30]
 
-        @impl Worker
+        @impl Oban.Worker
         def perform(_args, %Oban.Job{attempt: attempt}) when attempt > 3 do
           IO.inspect(attempt)
         end

test/oban/config_test.exs

@@ -23,6 +23,15 @@ defmodule Oban.ConfigTest do
       assert_valid(circuit_backoff: 10)
     end
 
+    test ":dispatch_cooldown is validated as a positive integer" do
+      assert_invalid(dispatch_cooldown: -1)
+      assert_invalid(dispatch_cooldown: 0)
+      assert_invalid(dispatch_cooldown: "5")
+      assert_invalid(dispatch_cooldown: 1.0)
+
+      assert_valid(dispatch_cooldown: 500)
+    end
+
     test ":crontab is validated as a list of cron job expressions" do
       assert_invalid(crontab: ["* * * * *"])
       assert_invalid(crontab: [["* * * * *", Fake]])