Distributed Elixir

Scaling systems can be tricky. When you add multiple instances of an application all kinds of problems rear up. For a simple CRUD apps that revolves around a database this doesn't require much thought.. There is no persisted state and everything just happily makes calls to an ACID complaint database.

For systems with real time drama, state outside a database, or distributed workloads, things are less straightforward. Elixir, by way of Erlang, has a much more elegant approach to this problem. Distributed processing is a first class citizen and its rather easy to setup.

A Simple Elixir Cluster

This is an example of setting up a simple distributed application from scratch. First we need an application scaffolded:

mix new elixir_cluster --sup
cd elixir_cluster

Now add libcluster to your mix.exs to give us a little help with the cluster configuration.

defp deps do
  [
    {:libcluster, "~> 3.3"}
  ]
end

Inside our application.ex we need setup the configuration for the cluster, just replace the scaffolded file completely.

defmodule ElixirCluster.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Cluster.Supervisor, 
      [topologies(), [name: ElixirCluster.ClusterSupervisor]]},
    ]

    opts = [strategy: :one_for_one, name: LibclusterCluster.Supervisor]
    Supervisor.start_link(children, opts)
  end

  defp topologies do
    [
      example: [
        strategy: Cluster.Strategy.Epmd,
        config: [
          hosts: [
            :"app@127.0.0.1",
            :"worker1@127.0.0.1",
            :"worker2@127.0.0.1"
          ]
        ]
      ]
    ]
  end
end

So whats happening here? First we changed the start function to include a new Cluster.Supervisor and gave it a name: ElxirCluster.ClusterSupervisor. This name is arbitrary. Then we setup the basic strategy and started our supervisor.

The topologies function is where we define the cluster configuration. Here we are configuring three different nodes: A main application instance, and two worker instances. These names are arbitrary and we will use them during launch to let each node know who it is.

That's it. Everything else will be taken care of by libcluster and Erlang.

To run the application now open three separate terminal windows in the project path and run

iex --name app@127.0.0.1 -S mix

iex --name worker1@127.0.0.1 -S mix

iex --name worker2@127.0.0.1 -S mix

Nothing much to see yet, but they should all run successfully. We get a list of the active nodes with Nodes.list/0.

iex(app@127.0.0.1)1> Node.list()
[:"worker1@127.0.0.1", :"worker2@127.0.0.1"]

Hit CTRL+C twice on each terminal to exit.

Distributed Computation

Now, it's time to add some functionality. For a simple example, we will calculate the prime factors of a list of numbers, distribute the workload across the workers, and collect the results.

First we need to create a GenServer that will accept the computation request.

defmodule ElixirCluster.Worker do
  use GenServer
  require Logger

  def start_link(_) do
    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
  end

  def compute(numbers) do
    nodes =
      Node.list()
      |> Enum.shuffle()

    node_count = length(nodes)
    number_count = length(numbers)
    chunk_size = div(number_count + node_count - 1, node_count)

    chunks = Enum.chunk_every(numbers, chunk_size)

    Enum.zip(nodes, chunks)
    |> Enum.map(fn {node, numbers} ->
      GenServer.call({__MODULE__, node}, {:compute, numbers})
    end)
    |> Enum.reduce([], fn {:ok, _node, factors}, acc -> acc ++ factors end)
  end

  @impl GenServer
  def init(state), do: {:ok, state}

  @impl GenServer
  def handle_call({:compute, numbers}, _from, state) do
    Logger.info("Receiving work request from for #{length(numbers)} number(s)")
    results = Enum.map(numbers, &{&1, factor(&1, 2, [])})
    {:reply, {:ok, node(), results}, state}
  end

  defp factor(1, _factor, acc), do: acc

  defp factor(number, factor, acc) when factor * factor > number do
    acc ++ [number]
  end

  defp factor(number, factor, acc) do
    if rem(number, factor) == 0 do
      factor(div(number, factor), factor, acc ++ [factor])
    else
      factor(number, factor + 1, acc)
    end
  end
end

First, we have start_link this is a GenServer helper that will be called by the supervisor to get the GenServer configuration.

Next, the compute function is our helper function to call the GenServers. We collect a list of remote nodes with Nodes.list. Using that list we split the list of numbers up into chunks for each node. The result is a tuple of the form {node, [numbers]}.

Finally, We map over the list of nodes/numbers and use GenServer.Call to make a call to the GenServer on each node. When the results are returned Enum.reduce compiles them into a list of factors.

After that, we have two GenServer implementations, init/1 which just initializes a starting state, for our purposes here the state will be an empty map.

The handle_call/3 function is our receiving function for the GenServer. it receives the payload {:compute, numbers}, maps across each number and calls factor/1. Then it replies with the new mapped list of factors.

The last function is just the factor/3 function, which computes the factors of a given number, using a little bit of recursion.

The last thing we need to do is add this GenServer to our supervision tree. We will add it to the children list in application.ex.

children = [
  {Cluster.Supervisor, [topologies(), [name: ElixirCluster.ClusterSupervisor]]},
  ElixirCluster.Worker
]

Taking it for a drive

So now we should have everything setup and ready. We can run the three terminals like before and we should get an iex console on each.

Now, on the first console, execute our compute helper function:

ElixirCluster.Worker.compute(Enum.to_list(5..21))

Like magic the workload is split up and distributed to each node, each node calculates the factors for its list and returns them. You should see the full list in iex:

[
  {5, [5]},
  {6, [2, 3]},
  {7, [7]},
  {8, [2, 2, 2]},
  {9, [3, 3]},
  {10, [2, 5]},
  {11, [11]},
  {12, [2, 2, 3]},
  {13, [13]},
  {14, [2, 7]},
  {15, [3, 5]},
  {16, [2, 2, 2, 2]},
  {17, [17]},
  {18, [2, 3, 3]},
  {19, [19]},
  {20, [2, 2, 5]},
  {21, [3, 7]}
]

Final Thoughts

Elixir and Erlang are like the cool kids of the system world, always ready with elegant solutions to complex problems. There is a lot more power available in Elixir clustering. This example shows how easy it is to work with distributed systems.

Source Code

The source code for this example is available in GitHub:

GitHub - 5sigma-blog/distributed-elixir

Contribute to 5sigma-blog/distributed-elixir development by creating an account on GitHub.

GitHub5sigma-blog