I wrote a small helper today which I eventually named async_map, a constrained version of Task.Supervisor.async_stream, if you will.
# Helper function to run supervised tasks concurrently mapping results to the
# desired output.
#
# - `async_fun` runs in a task process, takes a single enumerable item as
# argument and must return {:ok, result} or {:error, reason}.
# - `post_fun` runs in the calling process, takes a tuple of the result of
# async_fun and the corresponding enumerable item as argument, and returns the
# final result.
defp async_map(enumerable, async_fun, post_fun, opts) do
Task.Supervisor.async_stream(
MyApp.TaskSupervisor,
enumerable,
async_fun,
Keyword.put(opts, :on_timeout, :kill_task)
)
|> Enum.map(fn
{:ok, result} -> result
{:exit, :timeout} -> {:error, :timeout}
end)
|> Enum.zip(enumerable)
|> Enum.map(post_fun)
end
It converts timeouts into errors so that I have a single place to handle errors in post_func. It also gives post_func the original input, be it for logging or generating a fallback value.
I know about the :zip_input_on_exit option to Task.Supervisor.async_stream, but the advantage of doing it like this is that I have access to the input in every case of sucess/error/timeout.
Example usage:
1..10
|> async_map(
fn n ->
cond do
n == 5 -> {:error, :bad_input}
n == 7 -> {:ok, Process.sleep(10_000)}
true -> {:ok, n*n}
end
end,
fn
{{:ok, sq}, n} -> "#{n}*#{n} = #{sq}"
{{:error, reason}, n} -> "Error on #{n}: #{inspect(reason)}"
end,
max_concurrency: 5,
timeout: 500
)
# returns:
# ["1*1 = 1", "2*2 = 4", "3*3 = 9", "4*4 = 16", "Error on 5: :bad_input",
# "6*6 = 36", "Error on 7: :timeout", "8*8 = 64", "9*9 = 81", "10*10 = 100"]
