danj
Flow (0.15.0) does not terminate
I have what seems to be a simple Flow pipeline:
Flow.from_specs( [ { Zipstage,
[ dat_file, zip_file ]
} ],
cancel: :permanent
)
|> Flow.map( &from_line/1 )
|> Flow.filter( fn nil -> false; _ -> true end )
|> Flow.into_specs( [
{ Loadstage, [] },
{ Loadstage, [] },
{ Loadstage, [] },
], cancel: :permanent )
It fully processes the data, but when Zipstage sends { :stop, :shutdown, _ } and terminates, the Flow pipeline (Flow.Coordinator) doesn’t shut down/terminate. I’ve tried, probably erroneously, to add cancel: :permanent with no effect.
What am I missing?
Thanks!
Dan
First Post!
danj
The problem is in Loadstage (the into_spec) pertaining to some very specific process and gen_stage semantics related to subscription and termination. The answer is not at all apparent in the module documentation for either flow or gen_stage. I did find what I needed by examining the flow module Flow.MapReducer in the source code.
The primary points being:
- In init() call
Process.flag(:trap_exit, true)to inhibit unmanaged exits see erlang manual - implement
handle_subscribeto track producers - implement
handle_canceland when the producer list drops to 0, callGenStage.async_info( self(), :stop ) - implement a
handle_info( :stop...to generate a{ :stop, :normal, _ }tuple
But these points have subtleties, so it is important to read Flow.MapReducer to account for them. It is a very simple module compared to the rest of Flow and provides a solid template.
See flow/lib/flow/map_reducer.ex at v0.15.0 · dashbitco/flow · GitHub
You may ask, why not just use Flow.reduce: I needed a stage that would receive the demand unit of events as a whole and I didn’t see another way to get them.
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
Practical resources that improve the lives of professional developers.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #metaprogramming
- #security
- #hex
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking, logs, uptime, and Just Enough APM. Built for developers.
Producing high quality Elixir screencasts since 2017.
Enabling companies to succeed by building software people love.
The deployment platform built for Elixir: PaaS ease, VPS control.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"