Flow outlives a Task that starts it

[ACBullen]

This behaviour is probably intentional, but sometimes using Enumerable the Flow can keep running after the process that consumes the stream it generates is gone. In the context I initially encountered this, we're running HTTP requests inside a Flow and might want to abort the group mid run (for a failure or a need to stop hitting the server we're pointed at). Here's a reduced example that demonstrates a Flow outliving the task that starts it (you may need to run it two or three times to witness)

  def example() do
    p =
      spawn(fn ->
        1..100
        |> Flow.from_enumerable()
        |> Flow.map(fn n ->
          :timer.sleep(1000)
          IO.inspect(n)
          n
        end)
        |> Enum.map(fn n ->
          n
        end)
      end)
    :timer.sleep(5000)
    Process.exit(p, :kill)
  end

In working around this, I've made a module that cribs off of your implementation of Enumerable that allows us to reduce over the results, but also be able to cancel the Flow's processing:

defmodule FlowUtils do

  @spec linked_reduce(Flow.t(), term, (term, term -> term)) :: term()
  def linked_reduce(flow, acc, fun) do
    {:ok, {pid, stream}} =
      flow
      |> flow_to_stream()

    result = Enum.reduce(stream, acc, fun)
    :ok = ensure_shutdown(pid)

    result
  end

  @spec linked_reduce_while(Flow.t(), term, (term, term -> term)) :: term()
  def linked_reduce_while(flow, acc, fun) do
    {:ok, {pid, stream}} =
      flow
      |> flow_to_stream()

    result = Enum.reduce_while(stream, acc, fun)
    :ok = ensure_shutdown(pid)

    result
  end

  defp flow_to_stream(flow) do
    opts = [demand: :accumulate]

      case Flow.Coordinator.start_link(flow, :producer_consumer, {:outer, fn _ -> [] end}, opts) do
        {:ok, pid} ->
          {:ok, {pid, Flow.Coordinator.stream(pid)}}

        {:error, reason} ->
          exit({reason, {__MODULE__, :flow_to_stream, [flow]}})
      end
  end

  defp ensure_shutdown(flow_pid) do
    Process.exit(flow_pid, :kill)

    :ok
  end
end

First, is there a more typical way to get this kind of assurance, that I can terminate a running Flow externally when enumerating over the emitted results? Second, is there a chance to add something like this to make cancellable Flows more user-friendly?

[josevalim]

Correct. The flow is tied to producers. The best way to solve this is to start the flow and its consumer Under the same supervisor. Make it so the strategy is one_for_all and max_restarts is zero. This way when the child terminates, the supervisor will kill the flow.

This supervisor itself can be under a dynamic supervisor, so you can start multiple of them.