Elixir Message Costs

Message costs are the mailbox and size effects of sending messages between processes. Every unread message waits in the receiver mailbox until it is read.

MailboxCost reports a waiting queue

A waiting worker receives four job messages. MailboxCost reports four waiting messages.

defmodule MailboxCost do
  def queue_len(pid) do
    {:message_queue_len, len} = Process.info(pid, :message_queue_len)
    len
  end

  def send_all(pid, messages) do
    Enum.each(messages, fn message -> send(pid, message) end)
    length(messages)
  end

  def message_bytes(messages) do
    Enum.reduce(messages, 0, fn message, total -> total + byte_size(message) end)
  end
end
parent = self()
worker = spawn(fn ->
  send(parent, :ready)
  receive do
    :stop -> :ok
  end
end)
receive do
  :ready -> :ok
end
IO.inspect(MailboxCost.send_all(worker, [{:job, 1}, {:job, 2}, {:job, 3}, {:job, 4}]))
IO.inspect(MailboxCost.queue_len(worker))
ref = Process.monitor(worker)
send(worker, :stop)
receive do
  {:DOWN, ^ref, :process, ^worker, :normal} -> :ok
end

Longer messages carry more bytes

MailboxCost adds the byte sizes of a message list. A short word and a longer word differ by their bytes.

defmodule MailboxCost do
  def queue_len(pid) do
    {:message_queue_len, len} = Process.info(pid, :message_queue_len)
    len
  end

  def send_all(pid, messages) do
    Enum.each(messages, fn message -> send(pid, message) end)
    length(messages)
  end

  def message_bytes(messages) do
    Enum.reduce(messages, 0, fn message, total -> total + byte_size(message) end)
  end
end
IO.inspect(MailboxCost.message_bytes(["hi", "hello"]))
IO.inspect(MailboxCost.message_bytes(["a"]))

Reading drains the queue to zero

The current process receives two messages of its own. Reading both leaves an empty mailbox.

defmodule MailboxCost do
  def queue_len(pid) do
    {:message_queue_len, len} = Process.info(pid, :message_queue_len)
    len
  end

  def send_all(pid, messages) do
    Enum.each(messages, fn message -> send(pid, message) end)
    length(messages)
  end

  def message_bytes(messages) do
    Enum.reduce(messages, 0, fn message, total -> total + byte_size(message) end)
  end
end
send(self(), :one)
send(self(), :two)
IO.inspect(MailboxCost.queue_len(self()))
receive do
  :one -> :ok
end
receive do
  :two -> :ok
end
IO.inspect(MailboxCost.queue_len(self()))

Example

defmodule MailboxCost do
  def queue_len(pid) do
    {:message_queue_len, len} = Process.info(pid, :message_queue_len)
    len
  end

  def send_all(pid, messages) do
    Enum.each(messages, fn message -> send(pid, message) end)
    length(messages)
  end

  def message_bytes(messages) do
    Enum.reduce(messages, 0, fn message, total -> total + byte_size(message) end)
  end
end
parent = self()
worker = spawn(fn ->
  send(parent, :ready)
  receive do
    :stop -> :ok
  end
end)
receive do
  :ready -> :ok
end
messages = ["hi", "hello", "hey"]
IO.inspect(MailboxCost.send_all(worker, messages))
IO.inspect(MailboxCost.queue_len(worker))
IO.inspect(MailboxCost.message_bytes(messages))
ref = Process.monitor(worker)
send(worker, :stop)
receive do
  {:DOWN, ^ref, :process, ^worker, :normal} -> :ok
end

Key points

Mailbox cost: each unread message adds one item to the receiver mailbox.
Queue growth: sending more messages grows the waiting queue until the receiver reads them.
Message size: a longer message carries more bytes than a short one.
MailboxCost: the MailboxCost module in this lesson measures queue length and message bytes.
Drain: reading messages removes them from the mailbox and shrinks the queue.
Use MailboxCost.queue_len to check how many messages wait for a process.

In this example, the MailboxCost module is defined, then a waiting worker is started, then three messages are defined, then they are sent to the worker, then the queue length is printed, then their total byte size is printed, and finally the worker is stopped.

Note

a process that never reads its mailbox lets its queue grow without bound.

Try it Yourself

Edit the code, then press Run.

Running is simulated

defmodule MailboxCost do
  def queue_len(pid) do
    {:message_queue_len, len} = Process.info(pid, :message_queue_len)
    len
  end

  def send_all(pid, messages) do
    Enum.each(messages, fn message -> send(pid, message) end)
    length(messages)
  end

  def message_bytes(messages) do
    Enum.reduce(messages, 0, fn message, total -> total + byte_size(message) end)
  end
end
parent = self()
worker = spawn(fn ->
  send(parent, :ready)
  receive do
    :stop -> :ok
  end
end)
receive do
  :ready -> :ok
end
messages = ["hi", "hello", "hey"]
IO.inspect(MailboxCost.send_all(worker, messages))
IO.inspect(MailboxCost.queue_len(worker))
IO.inspect(MailboxCost.message_bytes(messages))
ref = Process.monitor(worker)
send(worker, :stop)
receive do
  {:DOWN, ^ref, :process, ^worker, :normal} -> :ok
end

Exercise

A waiting worker receives five messages and reads none. What does MailboxCost.queue_len return?

5 0 1 3