Example of using Task.Supervisor.async_nolink with Elixir Tasks

async_no_link.ex

Code.load_file("computation.ex")
Code.load_file("aggregator.ex")

defmodule AsyncNoLink do
  def run do
    :random.seed(:os.timestamp)
    Task.Supervisor.start_link(name: :task_supervisor)

    1..10
    |> Enum.map(fn(_) -> :random.uniform(1000) - 500 end)
    |> Enum.map(&Task.Supervisor.async_nolink(:task_supervisor, fn -> Computation.run(&1) end))
    |> collect_results
  end

  defp collect_results(tasks) do
    timeout_ref = make_ref
    timer = Process.send_after(self, {:timeout, timeout_ref}, 900) # 900 < 1000 some will faill
    try do
      collect_results(tasks, Aggregator.new, timeout_ref)
    after
      :erlang.cancel_timer(timer)
      receive do
        {:timeout, ^timeout_ref} -> :ok
      after 0 -> :ok
      end
    end
  end

  defp collect_results([], aggregator, _), do: {:ok, Aggregator.value(aggregator)}
  defp collect_results(tasks, aggregator, timeout_ref) do
    receive do
      {:timeout, ^timeout_ref} ->
        {:timeout, Aggregator.value(aggregator)}

      {:DOWN, _, _, pid, _} ->
        if Enum.member?(tasks, pid) do
          # Handling task termination. In this case, we simply delete the
          # task from the list of tasks, and wait for other tasks to finish.
          collect_results(List.delete(tasks, pid), aggregator, timeout_ref)
        else
          collect_results(tasks, aggregator, timeout_ref)
        end

      msg ->
        case Task.find(tasks, msg) do
          {result, task} ->
            collect_results(
              List.delete(tasks, task),
              Aggregator.add_result(aggregator, result),
              timeout_ref
            )
          nil -> collect_results(tasks, aggregator, timeout_ref)
        end
    end
  end
end

AsyncNoLink.run
|> inspect
|> IO.puts