Message queue length of a GenServer

Thanks a lot @msimonborg it makes more sense now.

You have 16 processes at work, each one with a job to do and each one can do its job asynchronously.

Your first process is the “caller” of your script or iex terminal, the one that is executing your code. It starts your 2nd process, the gen_server, which can then immediately go about its work asynchronously. Its job is to process messages in its queue and put the queue length to the terminal with each message it receives.

Once your gen_server is already doing its job waiting for messages, the caller starts up 14 new processes one by one, then waits until it receives messages from all of them. All 14 of those processes can do asynchronous work but they start their work in order one at a time. Their job is to immediately start sending messages to your gen_server and wait for its reply.

By the time your gen_server is processing its first message there are already 5 in the queue, but not 14, because it can start doing its job at the same time the caller is doing its job. Messages are piling up faster than the gen_server can process them, but they are finite, and once it catches up the length peaks and declines by 1 with each new message it processes.

Thanks for replying @Nicd, the desired output is the following:

queue length: 13
queue length: 12
queue length: 11
queue length: 10
queue length: 9
queue length: 8
queue length: 7
queue length: 6
queue length: 5
queue length: 4
queue length: 3
queue length: 2
queue length: 1
queue length: 0

I was trying to do a rate limit on an application with the message queue length, so I needed to keep track of the number of requests, now I know that the message queue length is not the best approach but I don’t understand why if I don’t block on the init function then the count of request will not be as I would expect.

What do you mean by “as expected”? How do you want it to work?

Hello @msimonborg thank you very much for replying to my message. After inspecting the length I was wrong is not actually 0, the code that I’m using is the following:

defmodule Demo do

  defp do_it(state) do
    {_, num} = Process.info(self(), :message_queue_len)
    IO.puts "queue length: #{inspect num}"

    {:reply, :hello, state}
  end

  def init(_args) do
    {:ok, []}
  end

  def handle_call(:hi, _from, state),
    do: do_it(state)

end

{:ok, pid} = GenServer.start_link(Demo,[])
f = fn ->
  GenServer.call(pid, :hi)
  :ok
end

(for _ <- 1..14, do: Task.async(f))
|> Enum.map(&Task.await(&1))

with output

queue length: 5
queue length: 12
queue length: 11
queue length: 10
queue length: 9
queue length: 8
queue length: 7
queue length: 6
queue length: 5
queue length: 4
queue length: 3
queue length: 2
queue length: 1
queue length: 0

But if I block as following:

defmodule Demo do

  defp do_it(state) do
    {_, num} = Process.info(self(), :message_queue_len)
    IO.puts "queue length: #{inspect num}"

    {:reply, :hello, state}
  end

  def init(_args) do
    {:ok, [], {:continue, :block}}
  end

  def handle_call(:hi, _from, state),
    do: do_it(state)

  def handle_continue(:block, state) do
    Process.sleep(1000) # block for one second
    {:noreply, state}
  end

end

{:ok, pid} = GenServer.start_link(Demo,[])
f = fn ->
  GenServer.call(pid, :hi)
  :ok
end

(for _ <- 1..14, do: Task.async(f))
|> Enum.map(&Task.await(&1))

Then it works as expected. So my question is why does blocking the process on initialization make it work as expected?

Thanks!

Hi @boilercoding , it would be helpful if you paste your code and your outputs. Are you copy and pasting the exact code above, or is it altered? Are you inspecting the length to verify that it’s 0?

@jeffweiss @peerreynders or anybody that can know the answer, why if I don’t block the init function with at least Process.sleep(50) the message queue length will always be 0?

Thanks @jeffweiss

If you proceed down this path and wish to return an {:error, :queue_full} you probably don’t want to do it from the handle_call. I’m assuming that if there are lots of messages that you want to discard the newest messages: the ones at the end of the queue. By putting the condition in handle_call if will affect the response to the message at the head of the queue.

In sandwich shop terms (because I’m hungry right now), putting the check and error response in handle_call is equivalent to getting in line at a sandwich shop before the lunch rush, waiting a little bit, lunch rush hits, getting to the counter to order and being told, “I’m sorry. I can’t make you a sandwich; the line is too long.” When I think what you want is a bouncer at the door that says, “I’m sorry. The line can’t extend outside.”

If you must implement this yourself rather than using some of the suggestions from other folks, you may wish to put the check in the client function rather than handle_call.

Here’s a quick proof of concept

defmodule OverloadedGenServer do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [])
  end

  def init(_) do
    {:ok, 0}
  end

  def add_num(pid, num) do
    case Process.info(pid, :message_queue_len) do
      {_, len} when len > 5 ->
        {:error, :queue_full}
      _ ->
        GenServer.call(pid, {:add_num, num})
    end
  end

  def handle_call({:add_num, num}, _from, prior_num) do
    new_num = prior_num + num
    Process.sleep(500)
    {:reply, new_num, new_num}
  end
end

{:ok, pid} = OverloadedGenServer.start_link()

for n <- 1..10 do
  Task.async(fn -> OverloadedGenServer.add_num(pid, n) end)
end
|> Enum.map(&Task.await/1)
|> IO.inspect

Process.exit(pid, :kill)

with output

$ elixir queue_test.exs
[
  1,
  3,
  6,
  10,
  15,
  21,
  {:error, :queue_full},
  {:error, :queue_full},
  {:error, :queue_full},
  {:error, :queue_full}
]
** (EXIT from #PID<0.73.0>) killed

@peerreynders thanks once again