Message queue length of a GenServer

How are you sending messages to that process?

I notice that you’re doing this in a handle_call which is synchronous. So, if you only have one process doing a GenServer.call to this process it’ll block until it gets a response. That means, you’ll never have a queue length, because this handle_call is consuming the only message that has been sent.

In short, what you’re doing is correct, but the way you’re testing it isn’t.

Try something like this to test:

1..100
|> Task.async_stream(fn _ -> 
  # Thing that calls your process
end)
|> Stream.run()

Thanks @blatyo

If you want to observe the size and not assert on it, :observer is specifically for this. :observer.start() will open the Observer WX GUI, go to the processes tab and you can sort on the MsgQ column. Right-clicking will allow you to get a dump of the current messages, but the Processes tab will refresh itself every so often, so you can use it to monitor for message queues growing, which usually means you have a bug with an unmatched message. I have found at least 3 bugs in production using only the MsgQ column and clicking to sort it. It is super useful.

@KronicDeth thanks…but I want the GenServer to take actions based on the queue length.Imagine 3 GenServers spun up by a DynamicSupervisor and they are all sending messages to each other but they don’t accept new messages if the message count is over 50 in their respective queues and send a reply back to the caller ..something like {:error, :queue_full}

Sounds like you want to control the ingestion of data in your system. If that’s the case, rather than fiddling with the queue length, I’d suggest taking a look at GenStage.

That is irrelevant given that the process can only control removal of messages from the mailbox and nothing else - i.e. a process can’t stop accepting messages; it can only stop accepting work which would be tracked inside the process state.

something like {:error, :queue_full}

a response like that happens when a received message is requesting more work but your work backlog indicates that you already have enough - i.e. the response should have nothing to do with messages remaining in the process mailbox but everything with information you currently have in the process state.

Thanks @stefanchrobot @peerreynders
@peerreynders basically I want to know the total remaining messages in the mailbox that haven’t been processed

A better way is to use ETS to track the queue length, then you can tell that the queue is full before messaging the process. See “Bounded Queues” under Handling Overload

Edit: just realized you might be beginning with Elixir, so this is probably overkill. What kind of work do your processes do? There’s likely a way to rearrange the problem so you don’t need bounded queues. For instance, using a process per message / task and limiting concurrency via DynamicSupervisor’s max children instead.

You already know the answer - you just aren’t creating the right conditions to see the effects.

defmodule Demo do

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

    reply = {:reply, :hello, state}
    cond do
      num > 0 ->
        reply
      true ->
        send(self(), :done)
        reply
    end
  end

  def init(_args) do
    send(self(), :block) # get the first message in the mailbox
    {:ok, []}
  end

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

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

  def terminate(reason, state) do
    IO.puts "terminate: #{inspect reason} #{inspect state}"
  end

end

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

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

$ elixir demo.exs
queue length: 3
queue length: 2
queue length: 1
queue length: 0
terminate: :normal []
$