jnnks

jnnks

Understanding GenStage/Flow Error Propagation

I am building a data processing pipeline which may sporadically fail. I would like to know when Producers/Consumers fail, so I can keep track of running pipelines. When I crash my producer, other processes crash with the same error message, which is confusing.

Consider the following setup:

defmodule CrashingProducer do
  use GenStage
  def init([]), do: {:producer, 0}
  def handle_demand(_, _), do: raise RuntimeError, "producer crashed"
end

# single error when commented, as expected, 
#   two error messages when included?
# Process.flag(:trap_exit, true)

{:ok, prod} =
  GenStage.start_link(CrashingProducer, [])
  |> IO.inspect(label: "prod pid")

spawn_link(fn ->
  Flow.from_stages([prod], max_demand: 1, stages: 1)
  |> Flow.run(link: false)
end)
|> IO.inspect(label: "flow pid")

:timer.sleep(1000)

When running it as is, the producer crashes, as expected with a single error.
When adding the exit trap, the GenStages started by Flow also crash, repeating the same error message. It appears as if the error originated in another process. What is the design decision behind this? It does not make sense to me.
Would it not be easier if the consumers :shutdown upon an error in the consumer? Can I do this manually?

Can I ignore these errors somehow? I already know that my producer failed, so I can shut everything else down manually. It is inconvenient to trap and handle these exits as well.

This is especially annoying when there are multiple Flow Stages and the terminal is spammed with the same error message over and over.

First Post!

jnnks

jnnks

I am now using Task.Supervisor.async_stream and a pipeline definition using Keyword lists. Something like this:

stages = [
  add_one: fn i -> i + 1 end,
  sub_one: fn i -> i - 1 end
]
input_stream = 1..10 |> Enum.to_list()

{:ok, sup} = Task.Supervisor.start_link(name: :tasksup)

reduce_fn = &Enum.reduce(stages, &1, fn {_name, stage}, item -> stage.(item) end)
Task.Supervisor.async_stream(sup, input_stream, reduce_fn)
|> Stream.take_while(fn
  {:ok, _} -> true
  _ -> false
end)
|> Stream.map(fn {:ok, item} -> item end)
|> Enum.to_list()
|> IO.inspect()

Most Liked

dimitarvp

dimitarvp

My bad, I misread the function name. Apologies for the noise.

Where Next?

Popular in Questions Top

hariharasudhan94
I would like to know what is the best IDE for elixir development?
New
lastday4you
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
jay1
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New
gshaw
What is the idiomatic way of matching for not nil in Elixir? E.g., First way: defp halt_if_not_signed_in(conn, signed_in_account) when...
New
bsollish-terakeet
Credo is smart enough to check for (something like) this: assert length(the_list) == 0 with this response: Checking if an enum is empt...
New
stefanluptak
Hello everybody, usually, I use a 29" ultra-wide monitor for VSCode which can easily accomodate explorer (files panel) + file with code ...
New

Other popular topics Top

saif
Hello everyone, Long time lurker first time poster here. I’ve recently begun working on Elixir full-time again! :raised_hands: It’s been...
New
SoCreat
i’m a new one to elixir which editor can i use vs code? or atom? Thanks! :smiley:
New
dblack
I’ve got an issue with an app and I’ve no idea of how to troubleshoot it. I’m hoping someone here might have seen something similar. I p...
New
TunkShif
This post is an instruction guide to help you setup your Neovim for Elixir development from scratch. It includes general information on h...
274 42716 114
New
JorisKok
I have a server on AWS, and was running a load test using artillery. When looking at the Phoenix dashboard I see the Ports going to 100% ...
New
senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New

We're in Beta

About us Mission Statement