Processes in Elixir

Keeping it Lightweight

Why Elixir and Erlang?

I’ve been experimenting with Elixir for a few years now but have never done anything production-worthy. I love the Elixir language. It combines the beauty and expressiveness of Ruby with a true functional language. Pus, when you combine it with the Phoenix Framework, you can design web applications in a Rails-like way, but with much more performance and explicitness. This is all great stuff, but I always had the nagging feeling that I was still not using the best part if the Elixir/Erlang ecosystem.

Erlang has a component called OTP (Open Telecom Protocol) that comes along for free when we use Elixir. OTP provides a rich set of tools to support concurrency and error handling. OTP can easily support huge numbers of lightweight concurrent processes, along with monitoring processes to ensure that these processes are automatically restarted if they die. These processes can also communicate through a simple but robust message handling system.

I’ve decided to take some time to build a simple distributed system to teach myself the ins and outs of OTP. This post will outline my experience setting up some simple processes with basic communication capabilities.

A Simple Problem

The problem I’m trying to solve here will start out very simple. I want to create a dispatcher process that can distribute and monitor tasks sent to an arbitrary number of servers. Each of these servers will also be a single Elixir process. For the first pass, we will create the following capabilities.

1. Create a dispatcher process
2. Create server processes and register them with the dispatcher
3. Dispatcher can catalog and report server summary

We will start small and build these capabilities in a step-by-step fashion. We will start by building a new mix project. We will create a project named ComputeFarm with this command.

% mix new compute_form

This creates a skeleton mix project that we will use for our experiments.

A Simple Server Using OTP Processes

The mix project contains an empty module in lib/compute_farm.ex. We will create a very simple server process in this file.

defmodule ComputeFarm do
  def server(name) do
    receive do
      message -> IO.puts "Server #{name} received #{message}"      
    end
    ComputeFarm.server(name)
  end
end

This simple function uses the receive function to wait for a message to be sent to it. When a message is received, we simply print information about the server and message. For this simple example, we assume the message is a string. Note that after the message is received, we recursively re-invoke the server function so we can wait another message.

In order to use this function as an OTP process, we need to invoke it with the spawn function. We can try this out at the Elixir command line.

% iex -S mix

This command compiles the mix project and gives us an interactive prompt where we can try out the function we just wrote. At the iex command line, we can then spawn a process running the function we just wrote.

iex(1)> server_1 = spawn(fn -> ComputeFarm.server("server_1") end)

Here we spawn an anonymous function that invokes the function we just wrote. The spawn function returns a PID that we can then use as a handle to send messages.

Now that we have a handle stored in the server_1 variable, we can send a message to that server like this.

iex(2)> send(server_1, "my_message")
Server server_1 received my_message
"my_message"

You’ll note that the server responded with the expected string. If we send another message, the server responds appropriately again.

iex(3)> send(server_1, "my_second_message")
Server server_1 received my_second_message
"my_second_message"

We can start any number of servers and route messages to those servers using the PID returned from each spawn invocation.

Now that we have a simple server, let’s work on an equally simple dispatcher.

Building a Dispatcher

Next, we need another process that serves as a dispatcher. This process will receive a message when a new server is created and will also support another message type that can be used to list available, registered servers.

Let’s start with a simple function that responds to these two message types.

def dispatcher do
  receive do
    :servers ->
       IO.puts("Available server list:")
    {:register, server_info} -> 
      IO.puts("Registered new server: #{server_info.name}")
    _ -> IO.puts("Unknown command")
  end
  ComputeFarm.dispatcher
end

Note that inside the receive block, we use Elixir’s pattern matching to look for two possible commands. The raw :servers atom will return a string listing the registered servers (currently empty). If a two element tuple with the atom, :register is provided, we will print a string showing the server we are registering. Finally, if we don’t match either of those patterns, we print an error. As with the server, we re-invoke the function so we can wait for more messages.

We can test drive this function as shown below.

iex(1)> dispatcher = spawn(fn -> ComputeFarm.dispatcher end)
#PID<0.119.0>
iex(2)> send(dispatcher, {:register, %{name: "My server"}})
Registered new server: My server
iex(3)> send(dispatcher, {:register, %{name: "My server2"}})
Registered new server: My server2
iex(4)> send(dispatcher, :servers)
Available server list:
iex(5)> send(dispatcher, :bad)
Unknown command

We now have the ability to communicate with both a server and dispatcher. We will make the dispatcher and servers to communicate.

Registering a Server

Our next step is to modify the server so that when we start it, we send the :register command to the dispatcher. In order to do this, the server needs to know the PID of the dispatcher process. This is the resulting server function.

def server(dispatcher_pid, name) do
  send(dispatcher_pid, {:register, %{name: name}})
  receive do·
    message -> IO.puts("Server #{name} received #{message}")
  end
  ComputeFarm.server(dispatcher_pid, name)
end

Now when we start a server, we send a message to the dispatcher to register that server. We can try out our new functionality as follows.

iex(1)> dispatcher = spawn(fn -> ComputeFarm.dispatcher end)
iex(2)> server_1 =
   spawn(fn -> ComputeFarm.server(dispatcher, "server_1") end)
Registered new server: server_1
iex(3)> server_2 =
   spawn(fn -> ComputeFarm.server(dispatcher, "server_2") end)
Registered new server: server_2

So once we can see that the act of spawning a new server now sends a message to the dispatcher and we see the dispatcher printing the appropriate message.

With this basic communication infrastructure in place, we can now focus on adding the real functionality.

Storing the Server List in the Dispatcher

Strong a list of servers inside the dispatcher brings up a new question: how do we store state in Elixir? If you are used to non-functional languages, you might try to initialize an empty list outside the dispatcher receive block, and then append a new name each time we register a new server. This will not work as the state we set will be lost each time we recursively re-invoke the dispatcher function. We can fix this by passing the state into the dispatch function each time it is called. Our resulting dispatch function now takes this form.

def dispatcher(servers \\ []) do
 receive do
   :servers ->
     server_string = Enum.join(servers, ",")
     IO.puts("Available server list: #{server_string}")
    {:register, server_info} ->
      servers = [server_info.name | servers]
      IO.puts("Registered new server: #{server_info.name}")
    _ ->
      IO.puts("Unknown command")
  end
    ComputeFarm.dispatcher(servers)
  end

With this change, we use the default parameter to dispatcher to set the initial server list to an empty string. Then each time we register a server, we add the server name to that list. After the receive block is processed, we re-invoke the dispatch function passing in the updated server list. This allows us to maintain state through the recursive calls.

Cleaning Up Warnings

While the function above does work, you’ll note that we get a fairly nasty warning at compile time.

warning: the variable "servers" is unsafe as it has been set inside a case/cond/receive/if/&&/||.

This happens because the case statement expects us to return a value but we are mutating a variable within the case statement. We can fix this by making sure each branch of the case statement returns the new value of the status list. So our resulting function looks like this.

def dispatcher(servers \\ []) do
  servers = receive do
    :servers ->
      server_string = Enum.join(servers, ", ")
      IO.puts("Available server list: #{server_string}")
      servers
    {:register, server_info} ->
      IO.puts("Registered new server: #{server_info.name}")
      [server_info.name | servers]
    _ ->
      IO.puts("Unknown command")
      servers
  end
  ComputeFarm.dispatcher(servers)
end

Summary

At this point, we’ve used OTP to build the framework of a simple server farm. We’ve gotten a glimpse of the following Elixir/OTP concepts.

1. How to use spawn to invoke a function inside an OTP process
2. How to use send to send messages to a process
3. How to use pattern matching on messages to support multiple message types
4. How to make two processes communicate through messages
5. How to save state using recursive functions

In my next post we will take this baseline and expand the functionality a bit. Next time, we will allow the dispatcher able to control the message flow to a number of servers and report more detailed server status.