I’m looking to process a bunch of tasks concurrently while not blocking the current process. I’d also like to aggregate the results into an object. I’d like the result to understand when an error occurred in the task, but instead of crashing when this happens, instead just add a note to the result indicating an error occurred.
For the purpose of making the problem simple, I’ve created a dummy “Calculator” which simulates a long running calculation.
defmodule Calculator do
def divide(val) do
# Simulating long calculation
:timer.sleep(1000)
100/val
end
end
I can send in 0 to this calculator to simulate an error.
I’ve implemented this behavior using a GenServer and TaskSupervisor.async. See code below:
defmodule Processor do
use GenServer
def start_link() do
GenServer.start_link( __MODULE__, [])
end
def handle_call(:get_state, _from, state) do
{:reply, state, state}
end
def handle_cast({:compute_values, values}, state) do
Enum.each(
values,
fn(val) ->
# Streams.TaskProcessor is a Task Supervisor supervised by application
Task.Supervisor.async_nolink(Streams.TaskSupervisor, fn() ->
Calculator.divide(val)
end)
end
)
{:noreply, state}
end
# For handling errors
def handle_info({:DOWN, _, _, _, {_error_type, _error_stacktrace}}, state) do
{:noreply, state ++ [:error]}
end
# For handling normal down states, do nothing
def handle_info({:DOWN, _, _, _, :normal}, state) do
{:noreply, state}
end
# For handling successful messages sent
def handle_info({ref, calculated_value}, state) do
{:noreply, state ++ [calculated_value]}
end
end
If I then call:
{:ok, p} = Processor.start_link
GenServer.cast(p, {:compute_values, [10, 0, 20, 5]})
And then after three seconds call:
GenServer.call(p, :get_state)
I get the expected result:
[10.0, 5.0, 20.0, :error]
It seems like this works, but I’m wondering:
-
Is anything that I’m doing here a bad practice? I’m new to the elixir
-
Is there an easier way to achieve this? Seems like a lot of boilerplate.
- When I saw Task.Supervisor.async_stream_nolink released in Elixir 1.4 I thought it may be helpful, but am having a hard time understanding if I could use it to achieve a similar result.
Thanks for any help!
