lud
Hello,
Is there an idiomatic way to write greedy streams in elixir?
Given this example:
1..1000
|> Stream.map(&computation_1/1)
|> Stream.map(&computation_2/1)
|> ...
If I understand streams correctly, this means that for each item we will have computation_1, then computation_2, and then we go to the next item.
What I would like is for computation_1 to be run as fast as possible without needing to be pulled from downstream. This is useful when computation_1 has variable times and computation_2 has steady times.
I could use Enum:
1..1000
|> Enum.map(&computation_1/1)
|> Stream.map(&computation_2/1)
|> ...
This is very greedy, but it blocks computation_2. It will not start until all computation_1 calls have been made.
So, is there a well-known pattern somewhere for that? In my specific case I’d like not to resort to aync_stream because of memory consumption.
Edit: hmmm while proofreading my post I realize that what I want is essentially concurrent. There is no solution on a single process. So I guess I can just run the top stream in a process, send all results to a second process, and receive in a stream from there.
Where Next?
Trending in Questions
Other Trending Topics
Latest on Elixir Forum
Sponsor Spotlight
At WyeWorks, we believe great software is built by combining technical excellence with genuine human connection
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #library
- #deployment
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #elixirconf
- #channels
- #exunit
- #discussion
- #code-sync
- #podcasts
- #javascript
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ai
- #ecto-query
- #elixirconf-us
- #blog-post
- #elixir-ls
- #phoenix_html
- #iex
- #graphql
- #genstage
- #websockets
- #supervisor
- #advent-of-code
- #distillery
- #processes
- #elixirconf-eu
- #api
- #forms
- #metaprogramming
- #hex
Our Sponsors
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.
Elixir, Erlang & BEAM conferences from Code Sync - join us in 2026!
Practical resources that improve the lives of professional developers.
Producing high quality Elixir screencasts since 2017.
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!"
Showing Posts 1 to 10- Show Best Posts
- Show All (oldest first)
- Show All (newest first)
D4no0
I think Task.async_stream/3 might be something you could use.NVM, saw you mentioning it as not the optimal solution.
dimitarvp
A nice compromizing solution – or a good prototype – would be to insert
Stream.chunk_every(1000)in the middle and see if it helps.D4no0
If your real-world processing gets more advanced, I think you could also take a look at GenStage, it will allow to make a much more readable and configurable processing pipeline.
lud
Not optimal but definitely an improvement in some cases. If I have a slow computation for every 10 computations, then a chunk size around 10 should help.
@D4no0 yes that’s the kind of setup that would involve two processes I was talking about in my Edit. Not sure if pulling that library would always be desirable but if yes then it would make a simple-to-write solution!
dimitarvp
I agree it’s not optimal, I simply don’t know anything about your problem. It seems like something that would accelerate computation_2 a bit.
If we’re talking truly optimal I’d just have a Kafka / NATS queue and send the results of the first computation to it and computation 2 would be pulling from that queue with the maximum speed and parallelism possible i.e. you would code a solution that’s hard-bottlenecked on the speed with which items from computation_1 can be emitted.
D4no0
I think it depends on the use-case, but I would highly recommend to read the documentation as it should contain the list of things it solves compared to self-rolled processing pipelines.
lud
Oh I know, I meant it was actually a simple and good enough solution. Simplicity is a virtue
My current use case is generating an ex_unit test from external resources (computation_1) and syncing that file to the disk or an external API.
This is not something where you want external services such as Kafka.
Honestly there is not much to optimize in here, I created that topic for the general case. We could reduce the scope a little bit. I think the real question can be reduced: Given two steps in a stream chain, how can we run both steps in parallel. While
computation_2runs for itemi, how to makecomputation_1run for itemi+1.dimitarvp
Then a very normal task pool, a library or homegrown, will do the job just fine?
lud
Yeah I’m running homegrown
One process running the top stream and sending every reesults to the parent process.
The parent process starts it’s stream by looping over
receive.dimitarvp
You can do even better than that. If you use f.ex.
opqorpoolexyou can distribute the results of computation_1 to a job pool and then it will run jobs on each complete computation – in parallel.The way you seem to be describing your usage seems like the distribution of tasks for downstream processing is serial (by a single process running
receivein a loop). And it can be made parallel.EDIT: you can probably even use
Phoenix.PubSub.