Getting Started with Elixir’s GenServer

Async Processing Made Easy

Elixir and Parallel Processing

I think when most of us start working with Elixir, we tend to be drawn to the functional nature of the language and the unique pattern matching syntax that allows us to write applications in new expressive ways. While these represent the first layer of what makes Elixir great, there is another layer that usually takes us longer to grasp.

For me, the second layer of Elixir greatness is its ability to easily handle asynchronous and parallel processing in a smart elegant way. Elixir can spin up small, discrete processes with little system overhead. This capability allows us to use processes ubiquitously in our applications.

While there are a number of building blocks of asynchronous processing in the Elixir language (agents, processes, tasks, etc.), the real cornerstone of this part of Elixir is the GenServer protocol. GenServer is a framework that allows us to write servers that can hold state and receive and respond to messages asynchronously. GenServer is also a building block of other protocols like GenStage.

Introduction to GenServer

At it’s core, GenServer is pretty simple. It’s a protocol that allows you to spin up a stateful server and define an API to asynchronously send messages and receive responses.

The best way to learn GenServer is through an example. Let’s build a little service that allows us to send items to a queue, query the length and contents of that queue and fetch elements from that queue.

Defining Our Public API

Let’s start outside in and first define the API. One thing to keep in mind is that every GenServer instance has an associated PID that will be used to access it. These API functions will accept a PID as one of their parameters. Here is a first pass at the API.

defmodule BatchQueue do
  def add(pid, item) do  
  end
  def list(pid) do 
  end
  def length(pid) do  
  end
  def fetch(pid) do  
  end
end

You’ll note that there are two types of function calls here.

1. Functions .that receive a command and/or data and don’t expect a response. These are referred to as casts.
2. Functions that receive and command and/or data and return a response synchronously. These are referred to as calls.

Next, we will build out these functions and start to reveal the protocol and callback structure behind GenServer.

Building Out the Public API

These API calls are generally simple. We use the GenServer.cast and GenServer.call functions to send commands and data to our server. These functions are usually every simple. By convention, we generally use a single atom to send a command without data or a tuple to send data to the calls that require command and data. In our example, the only API call that sends data is our add function. In this case, the built-out API looks like this.

defmodule BatchQueue do
  use GenServer
  def add(pid, item) do  
    GenServer.cast(pid, {:add, item})
  end
  def list(pid) do 
    GenServer.call(pid, :list)
  end
  def length(pid) do  
    GenServer.call(pid, :length)
  end
  def fetch(pid) do  
    GenServer.call(pid, :fetch)
  end
end

You’ll note we’ve added the use GenServer statement. This gives us access to the GenServer API. If you try to compile the code, you’ll get a warning about a missing init statement. In order to start using this application, we will need a way to initialize and start the server.

Initializing and Starting the GenServer

In order to start our GenServer , we will need to add two additional functions.

The init/1 function is a required callback for the GenServer protocol. This function accepts one arbitrary parameter. In our case, we will be passing in the initial state of our system. While in a complex system, there may be a complex init/1 function, in our case, we simply return a tuple with the state that was passed in.

  def init(queue) do
    {:ok, queue}
  end

The start_link/1 is the public function used to start our server. It calls GenServer.start_link/3 with the name of our module and an argument that will get passed to init/1. In our case this is our initial state (a newly initialized and empty Erlang queue).

  def start_link() do
    GenServer.start_link(__MODULE__, :queue.new())
  end

At this point, we can start our application and call our start_link/0 function.

iex(1)> {:ok, pid} = BatchQueue.start_link()
{:ok, #PID<0.115.0>}

If we try to invoke one of our previously defined API functions, we get an error like this.

iex(2)> BatchQueue.length(pid)
** (RuntimeError) attempted to call GenServer #PID<0.124.0> but no handle_call/3 clause was provided

This is expected since we haven’t defined the callbacks needed to serve the cast or call functions. Let’s start to define these functions.

Creating GenServer Callbacks

For each of the cast or call functions we invoke from our public API, we need to add a handle_cast/2 or handle_call/3 callback to our GenServer module. Let’s start with the callback that handles returning the length of the queue. In this case, we send the atom :length to GenServer.call. The callback to handle this function looks like this.

  def handle_call(:length, _from, queue) do
    {:reply, :queue.len(queue), queue}
  end

The handle_call callback receives three parameters:

1. The request we sent (in this case a single atom)
2. A tuple indicating the PID of the sender and a unique identifier (we ignore this)
3. The GenServer system state (in this case the Erlang queue)

This function clause pattern matches on the single :length atom in the request parameter.

We return a tuple of the :reply atom, the response we wish to return (in this case, the queue length) and the new system state. Since the length operation does not modify the state of the queue, we simply return the existing state.

With this code in place, we can now make our first successful call to our GenServer.

iex(1)> {:ok, pid} = BatchQueue.start_link()
{:ok, #PID<0.125.0>}
iex(2)> BatchQueue.length(pid)
0

With this in place, we can build out the more interesting callback to handle our add command. Since we use a cast to invoke this function, we need to create a handle_cast/2 function clause that matches the pattern we defined for this function.

  def handle_cast({:add, item}, queue) do
    {:noreply, :queue.in(item, queue)}
  end

Note that this function accepts two parameters:

1. Our request (in this case a tuple with a command and data)
2. The current state of the syatem

We respond with a tuple that contains the :noreply atom and the new state of the system. This state is created by the :queue.in function which adds an element to the queue and returns the new queue.

With this callback in place, we can give it a test drive.

iex(1)> {:ok, pid} = BatchQueue.start_link()
{:ok, #PID<0.125.0>}
iex(2)> BatchQueue.add(pid, "item-1")
:ok
iex(3)> BatchQueue.length(pid)
1
iex(4)> BatchQueue.add(pid, "item-2")
:ok
iex(5)> BatchQueue.length(pid)
2

This seems to indicate that we have added the item to the queue.

Following this pattern, we can add the rest of our callbacks

  def handle_call(:list, _from, queue) do
    {:reply, :queue.to_list(queue), queue}
  end
  def handle_call(:fetch, _from, queue) do
    with {{:value, item}, new_queue} <- :queue.out(queue) do
      {:reply, item, new_queue}
    else
      {:empty, _} ->
        {:reply, :empty, queue}
    end
  end

These are relatively simple, bit note that the API surrounding the Erlang queue is a little weird and results in some extra logic to properly handle the empty queue condition.

With these last callbacks in place, we can then run through a more interesting exercise with our server.

iex(1)> {:ok, pid} = BatchQueue.start_link()
{:ok, #PID<0.124.0>}
iex(2)> BatchQueue.add(pid, "item-1")
:ok
iex(3)> BatchQueue.add(pid, "item-2")
:ok
iex(4)> BatchQueue.list(pid)
["item-1", "item-2"]
iex(5)> BatchQueue.fetch(pid)
"item-1"
iex(6)> BatchQueue.list(pid)
["item-2"]
iex(7)> BatchQueue.length(pid)
1

We now have a functional queue with some rudimentary operations that work as expected.

Summary

We now have a working version of a simple queue written using Elixir’s GenServer protocol. You can see the full source code for this example along with a simple test suite in this github repository.

GenServer is a very powerful way to build lightweight processes that allow us to execute code in a completely parallel and asynchronous fashion. It is also the first building block in creating even more powerful services that span multiple processes. In my next post, we will build upon this work to build a batch processing tool that uses data sourced from this queue withthe Elixir GenStage protocol.