🎓 A1Learner
🔍
0 of 46 lessons completed

Elixir Project: Job Queue

A job queue is a GenServer that holds jobs in order and runs them one after another. A supervisor watches the queue and restarts it if its process stops.

Jobs run in the order they were added

Each job is a tuple that names its kind and its values. Running the queue maps the jobs to results in order.

defmodule Queue do
  use GenServer
  def start_link(_arg) do
    GenServer.start_link(__MODULE__, %{pending: [], done: []}, name: __MODULE__)
  end
  def init(state), do: {:ok, state}
  def handle_call({:enqueue, job}, _from, state) do
    {:reply, :ok, %{state | pending: state.pending ++ [job]}}
  end
  def handle_call(:run, _from, state) do
    results = Enum.map(state.pending, &process/1)
    {:reply, results, %{state | pending: [], done: results}}
  end
  def handle_call(:results, _from, state), do: {:reply, state.done, state}
  def handle_call(:pending, _from, state), do: {:reply, state.pending, state}
  defp process({:double, n}), do: n * 2
  defp process({:add, a, b}), do: a + b
  defp process({:upcase, text}), do: String.upcase(text)
end
{:ok, _pid} = Queue.start_link([])
GenServer.call(Queue, {:enqueue, {:double, 21}})
GenServer.call(Queue, {:enqueue, {:add, 2, 3}})
GenServer.call(Queue, {:enqueue, {:upcase, "ana"}})
IO.inspect(GenServer.call(Queue, :run))
GenServer.stop(Queue)

A supervisor owns the queue

The queue runs as a child of a supervisor. The supervisor reports one child.

defmodule Queue do
  use GenServer
  def start_link(_arg) do
    GenServer.start_link(__MODULE__, %{pending: [], done: []}, name: __MODULE__)
  end
  def init(state), do: {:ok, state}
  def handle_call({:enqueue, job}, _from, state) do
    {:reply, :ok, %{state | pending: state.pending ++ [job]}}
  end
  def handle_call(:run, _from, state) do
    results = Enum.map(state.pending, &process/1)
    {:reply, results, %{state | pending: [], done: results}}
  end
  def handle_call(:results, _from, state), do: {:reply, state.done, state}
  def handle_call(:pending, _from, state), do: {:reply, state.pending, state}
  defp process({:double, n}), do: n * 2
  defp process({:add, a, b}), do: a + b
  defp process({:upcase, text}), do: String.upcase(text)
end
{:ok, sup} = Supervisor.start_link([Queue], strategy: :one_for_one)
GenServer.call(Queue, {:enqueue, {:double, 21}})
GenServer.call(Queue, {:enqueue, {:add, 2, 3}})
IO.inspect(GenServer.call(Queue, :run))
IO.inspect(length(Supervisor.which_children(sup)))
Supervisor.stop(sup)

A stopped queue restarts empty

Stopping the queue process makes the supervisor start it again. The new queue has no pending jobs.

defmodule Queue do
  use GenServer
  def start_link(_arg) do
    GenServer.start_link(__MODULE__, %{pending: [], done: []}, name: __MODULE__)
  end
  def init(state), do: {:ok, state}
  def handle_call({:enqueue, job}, _from, state) do
    {:reply, :ok, %{state | pending: state.pending ++ [job]}}
  end
  def handle_call(:run, _from, state) do
    results = Enum.map(state.pending, &process/1)
    {:reply, results, %{state | pending: [], done: results}}
  end
  def handle_call(:results, _from, state), do: {:reply, state.done, state}
  def handle_call(:pending, _from, state), do: {:reply, state.pending, state}
  defp process({:double, n}), do: n * 2
  defp process({:add, a, b}), do: a + b
  defp process({:upcase, text}), do: String.upcase(text)
end
{:ok, sup} = Supervisor.start_link([Queue], strategy: :one_for_one)
GenServer.call(Queue, {:enqueue, {:double, 10}})
old = GenServer.whereis(Queue)
Process.exit(old, :kill)
wait = fn wait, old_pid, tries ->
  current = GenServer.whereis(Queue)
  if current != nil and current != old_pid do
    :ready
  else
    if tries >= 200 do
      :timeout
    else
      Process.sleep(10)
      wait.(wait, old_pid, tries + 1)
    end
  end
end
_ready = wait.(wait, old, 0)
IO.inspect(GenServer.call(Queue, :pending))
GenServer.call(Queue, {:enqueue, {:add, 3, 4}})
GenServer.call(Queue, {:enqueue, {:upcase, "ana"}})
IO.inspect(GenServer.call(Queue, :run))
Supervisor.stop(sup)

Example

defmodule Queue do
  use GenServer
  def start_link(_arg) do
    GenServer.start_link(__MODULE__, %{pending: [], done: []}, name: __MODULE__)
  end
  def init(state), do: {:ok, state}
  def handle_call({:enqueue, job}, _from, state) do
    {:reply, :ok, %{state | pending: state.pending ++ [job]}}
  end
  def handle_call(:run, _from, state) do
    results = Enum.map(state.pending, &process/1)
    {:reply, results, %{state | pending: [], done: results}}
  end
  def handle_call(:results, _from, state), do: {:reply, state.done, state}
  def handle_call(:pending, _from, state), do: {:reply, state.pending, state}
  defp process({:double, n}), do: n * 2
  defp process({:add, a, b}), do: a + b
  defp process({:upcase, text}), do: String.upcase(text)
end
{:ok, sup} = Supervisor.start_link([Queue], strategy: :one_for_one)
GenServer.call(Queue, {:enqueue, {:double, 21}})
GenServer.call(Queue, {:enqueue, {:add, 2, 3}})
GenServer.call(Queue, {:enqueue, {:upcase, "ana"}})
GenServer.call(Queue, {:enqueue, {:double, 3}})
IO.inspect(GenServer.call(Queue, :run))
IO.inspect(GenServer.call(Queue, :pending))
Supervisor.stop(sup)

Key points

  • Enqueue: a job is added to the end of the pending list.
  • Run: running the queue processes jobs in the order they were added.
  • Results: the run returns results in the same order as the jobs.
  • Supervisor: the supervisor restarts the queue if its process stops.
  • State: a restarted queue starts with an empty pending list.

In this example, four jobs are added to the queue under its supervisor. Running the queue returns one result per job in job order, and the pending list is empty afterwards. The supervisor is stopped at the end.

Note

jobs here are plain tuples, and each kind of job has its own rule.

Try it Yourself

Edit the code, then press Run.

Running is simulated

Exercise

The queue process is stopped and its supervisor restarts it. What is the pending list of the restarted queue?