GenServer.reply/2

README.md

GenServer.reply/2

Check the other two files in this gist.

In the common.exs we show a standard approach (by the book) to handle incoming requests via handle_call/3 and {:reply, ...} return values. It blocks processing of all the coming requests until the one that is handled finished.

In contrast, in the reply.exs we leverage power of Task.async/1, GenServer.reply/2, and {:noreply, ...} return values. It allows our GenServer process incoming requests in parallel. Then the processing finished, we send results back to the original client (from pid).

Warning

Be careful, as this approach can lead to uncontrollable load to our GenServer.

Check Flow or other techiques (like pooling) in order to provide back pressure.

Compare timestamps:

> elixir common.exs
2023-08-10 09:37:15Z: Sleeping for 2 seconds...
2023-08-10 09:37:17Z: Sleeping for 2 seconds...
2023-08-10 09:37:19Z: Sleeping for 2 seconds...
2023-08-10 09:37:21Z: Sleeping for 2 seconds...
2023-08-10 09:37:23Z: Sleeping for 2 seconds...
2023-08-10 09:37:25Z: [1000, 2000, 3000, 4000, 5000]

> elixir reply.exs
2023-08-10 09:37:31Z: Sleeping for 2 seconds...
2023-08-10 09:37:31Z: Sleeping for 2 seconds...
2023-08-10 09:37:31Z: Sleeping for 2 seconds...
2023-08-10 09:37:31Z: Sleeping for 2 seconds...
2023-08-10 09:37:31Z: Sleeping for 2 seconds...
2023-08-10 09:37:33Z: [1000, 2000, 3000, 4000, 5000]

common.exs

defmodule ReplyExample.Server do
  use GenServer

  @timeout :timer.seconds(30)

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def do_the_thing(pid \\ __MODULE__, n) do
    GenServer.call(pid, {:do_the_thing, n}, @timeout)
  end

  @impl GenServer
  def init(_opts) do
    {:ok, %{}}
  end

  @impl GenServer
  def handle_call({:do_the_thing, n}, _from, state) do
    log("Sleeping for 2 seconds...")
    Process.sleep(2000)

    {:reply, n * 1000, state}
  end

  defp log(message) do
    now = DateTime.utc_now() |> DateTime.truncate(:second)
    IO.puts("#{now}: #{message}")
  end

  def test do
    {:ok, pid} = start_link()

    try do
      1..5
      |> Enum.map(fn n -> Task.async(fn -> do_the_thing(n) end) end)
      |> Task.await_many(@timeout)
      |> inspect()
      |> log()
    after
      GenServer.stop(pid)
    end
  end
end

ReplyExample.Server.test()

reply.exs

defmodule ReplyExample.Server do
  use GenServer

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def do_the_thing(pid \\ __MODULE__, n) do
    GenServer.call(pid, {:do_the_thing, n})
  end

  @impl GenServer
  def init(_opts) do
    {:ok, %{}}
  end

  @impl GenServer
  def handle_call({:do_the_thing, n}, from, state) do
    Task.async(fn ->
      log("Sleeping for 2 seconds...")
      Process.sleep(2000)

      GenServer.reply(from, n * 1000)
    end)

    {:noreply, state}
  end

  @impl GenServer
  def handle_info(_msg, state) do
    {:noreply, state}
  end

  defp log(message) do
    now = DateTime.utc_now() |> DateTime.truncate(:second)
    IO.puts("#{now}: #{message}")
  end

  def test do
    {:ok, pid} = start_link()

    try do
      1..5
      |> Enum.map(fn n -> Task.async(fn -> do_the_thing(n) end) end)
      |> Task.await_many()
      |> inspect()
      |> log()
    after
      GenServer.stop(pid)
    end
  end
end

ReplyExample.Server.test()