Building a Robust Elixir Dispatcher

I’m pretty new to Elixir. This language fascinates me as it is based on a paradigm I never experienced before.
The ideas of functional programming, processes, message passing and fault tolerance are bundled together into a language and eco system which is fun and productive to work with.

Recently during a feature I was working on, I had to code an Elixir module that receives and dispatches tasks. I wanted to share with you my journey to the final module I ended up with.

The Spec

The requirements from the module were the following:

1. The module receives and runs tasks.
2. Each task is composed of a task identifier and a function to run.
3. While a task is being executed its result should be :pending.
4. A task that takes more than 10 seconds in considered timed out. Its result should return :timeout.
5. A task that throws an unhandled exception should return {:error, the_error} as its result.
6. Each task result should be saved until requested. This includes tasks that timed out, threw an exception or succeeded.

First Iteration — A Naive Dispatcher

My first thought was to create a GenServer and use the cast method to dispatch the task.
A task would be a simple zero arguments function passed to the GenServer.

This is the most simple solution I could come up with at first. The GenServerhas a run method which receives a task id and a function to perform. To retrieve a task result one would use the result method with the task id. The state of the GenServer is a task_id => result map.
Let’s run $ iex dispatcher.exs and start playing with the Dispatcher module.

We can see that our dispatcher handles well simple calculations — I performed two tasks: task a which resolves to the value 2 and task b which resolves to 4. Once a value is retrieved the task is removed from the Dispatcher.
Let’s try a dipatching long running task:

I dispatched a task that waits for 8 seconds and returns “finished”. When I tried to retrieve the task result I got a timeout exception instead of getting back :pending from the Dispatcher.
That’s expected since the Dispatcher currently performs the task in a synchronous way. The task blocks the Dispatcher so it becomes unresponsive. We can use processes to make the Dispatcher perform the task asynchronously. The simplest way would be to use spawn/1 and run the given task in the spawned process. This way the Dispatcher stays responsive regardless of the time it takes a task to complete.

Second Iteraion — Async

The new handle_cast callback spawns a process which completes the task and sends the result back to the Dispatcher along with its task_id.
In order to process the result message from the spawned process we had to implement the handle_info (line 26) callback.

Let’s try the async version of the Dispatcher:

Now the Dispatcher replied immediately with :pending since the task is being performed asynchronously in a spawned process which does not block the Dispatcher process itself.
When the task is done, the value is stored and we can retrieve it.
The Dispatcher still doesn’t handle tasks timeouts though — When a task takes longer than 10 seconds we should set its result to :timeout.
My strategy was to start the task and send the Dispatcher a delayed message indicating that the task run period was over. If that message arrived before the task had sent its result the Dispatcher marked it as timed out.

Third Iteration — Handling Timeouts

The Dispatcher calls Process.send_after/4 (line 19) to receive a delayed message indicating the task has timed out. I added another handle_info callback implementation (line 31) to handle those timeout messages. It will only set the result of task_id as :timeout in case it is still marked as :pending. This way, only if the timeout message arrives before the spawned process replies with the task’s result, the task is marked with :timeout.
Let’s test the new Dispatcher:

Looks like we’re good. The task ran for 12 seconds which is more than our 10 seconds timeout threshold and was marked with a :timeout.
The only requirement I haven’t fulfilled yet is dealing with crashed tasks. Let’s see how the Dispatcher currently handles crashed tasks:

The process we spawned for the task has crashed because of an exception. It did not report back to the Dispatcher so after 10 seconds it was marked with :timeout.

Final Iteration — Handling Exceptions

We would like to know when a process we spawned has ended its life and what was the cause. Did it end normally or with an exception?
Luckily, Elixir provides an easy way to get notified about processes we care about: Monitors. We can ask Elixir to notify us when a process ends by calling Process.monitor/1 on the spawned process PID. When the spawned process ends, the process which issued the Process.monitor/1 receives a message of the following format: {:DOWN, ref, :process, pid, reason}
ref is the value returned from the Process.monitor/1 call.
reason is :normal if the process ended without an uncaught exception, otherwise it holds the exception itself.
The strategy here is to spawn a process, monitor it and handle the incoming process :DOWN messages to identify processes that threw an exception.
The Dispatcher has to maintain a mapping between the ref returned by the monitor command and the task_id, so when it receives a process :DOWNmessage it knows what task_id it refers to.
This is the final and complete version of the Dispatcher:

The 3 interesting points in the above code are:

• Lines 31 and 33 handles the monitor creation and updating the ref => task_id mapping.
• The handle_info callback on line 52 handles :DOWN messages of tasks that ended normally.
• The handle_info callback on line 56 handles those processes that threw an exception by setting the corresponding task result with {:error, the_exception}

Let’s test the Dispatcher’s exception handling:

The exception was stored under the result of task ”a” as we wanted.

Summary

I learnt a lot about GenServers, Processes, Monitors and how processes in Elixir communicate between themselves. Building the Dispatcher was both fun and insightful.
There is no doubt I’m going to explore Elixir some more in the future!