[wwitek-whatnot]

Background

In our production system, when a lot of users disconnect at the same time from a single Phoenix channel, we run into very high queues that put pressure on GC and the results is that the queues never go down. We tested using off-heap message queue for the PubSub`s registry and it helped, so we'd like to introduce that as an option for PubSub, starting with an option for Registry.

Changes

Adds an :off_heap start option to Registry (default false) that sets message_queue_data: :off_heap on the registry's partition processes.

Adds spawn_opt that are passed to Registry partitions.

Benchmark

Benchmark simulates a crash storm: a single-partition registry with N linked registrants, the partition process suspended (via :sys.suspend/1) so :EXIT messages accumulate in its mailbox instead of being processed as they arrive, then all N registrants killed at once. Measures the time for an explicit :erlang.garbage_collect/1 and the time to resume and fully drain the mailbox.

defmodule OffHeapBench do
  def run(off_heap, queue_size) do
    name =
      String.to_atom("Bench_#{off_heap}_#{queue_size}_#{System.unique_integer([:positive])}")

    {:ok, _} =
      Registry.start_link(keys: :unique, name: name, partitions: 1, off_heap: off_heap)

    {_kind, _partitions, _key_ets, {partition, _pid_ets}, _listeners} =
      :ets.lookup_element(name, -1, 2)

    # Plain spawn (not spawn_link / Task.start_link): the only link created
    # is the one Registry.register/3 makes between the worker and the
    # partition process, so killing workers only affects the partition.
    workers =
      for i <- 1..queue_size do
        spawn(fn ->
          {:ok, _owner} = Registry.register(name, i, :value)
          receive do: (:stop -> :ok)
        end)
      end

    wait_until(fn -> Registry.count(name) == queue_size end)

    # Suspend the partition GenServer so the burst of :EXIT messages piles
    # up in its mailbox instead of being drained as it arrives, simulating
    # a crash storm hitting an already-busy partition.
    :sys.suspend(partition)

    for pid <- workers, do: Process.exit(pid, :kill)

    wait_until(fn ->
      Process.info(partition, :message_queue_len) == {:message_queue_len, queue_size}
    end)

    before = Process.info(partition, [:memory, :heap_size, :total_heap_size])

    gc_time_us = time_us(fn -> :erlang.garbage_collect(partition) end)

    after_gc = Process.info(partition, [:memory, :heap_size, :total_heap_size])

    drain_time_us =
      time_us(fn ->
        :sys.resume(partition)
        wait_until(fn -> Process.info(partition, :message_queue_len) == {:message_queue_len, 0} end)
      end)

    Supervisor.stop(name)

    %{
      queue_size: queue_size,
      off_heap: off_heap,
      before_gc: before,
      after_gc: after_gc,
      gc_time_us: gc_time_us,
      drain_time_us: drain_time_us
    }
  end

  defp wait_until(fun) do
    if fun.() do
      :ok
    else
      Process.sleep(5)
      wait_until(fun)
    end
  end

  defp time_us(fun) do
    t0 = System.monotonic_time(:microsecond)
    fun.()
    System.monotonic_time(:microsecond) - t0
  end
end

queue_sizes = [1_000, 10_000, 50_000, 100_000]

results =
  for queue_size <- queue_sizes, off_heap <- [false, true] do
    result = OffHeapBench.run(off_heap, queue_size)
    Process.sleep(50)
    result
  end

for r <- results do
  IO.inspect(r)
end

Results:

queue      off_heap    heap(before)  heap(after_gc)  mem(before)   gc_time_us    drain_us
--------------------------------------------------------------------------------------
1000       false            233             233         130880            37       7453
1000       true             233             233         130984             4       5883
10000      false            233             233        1282880           281      41452
10000      true             233             233        1282984             6      11763
50000      false            233             233        6402880          1031     884399
50000      true             233             233        6402984             8      35475
100000     false            233             233       12802880          1883    3281872
100000     true             233             233       12802984            12      65628

At 100k queued :EXIT messages, draining the backlog takes ~3.28s with the default :on_heap behavior vs. ~66ms with off_heap: true — roughly 50x faster.

Testing

Added tests covering: the :off_heap option is validated as a boolean, partitions run with :on_heap message queue data by default, and partitions run with :off_heap message queue data when the option is enabled.

[wwitek-whatnot]

This PR is a follow up to switch to order_set

[josevalim]

Sorry, just one other tiny change needed!