mfclarke

mfclarke

GenStage ProducerConsumerSupervisor?

We have ConsumerSupervisor which starts 1 child per event. Is it possible to use this as a :producer_consumer ?

[A] -> [B(1, 2, 3, 4, 5, …)] -> [C]

Effectively what I’m after is being able to get B to subscribe to A and C to subscribe to B. Then manage concurrency of B just with :max_demand. As opposed to starting an arbitrary number of regular :producer_consumer stages for B and manually handling subscriptions of C to each B.

The example of using a ConsumerSupervisor in the gen_stage repo returns a Supervisor style term in it’s init/1 but the docs show a regular GenStage style return value where you specify :producer, :producer_consumer or :consumer: https://hexdocs.pm/gen_stage/ConsumerSupervisor.html#init/1

So I’m thinking this might be possible but I’m not understanding how to set it up.

Marked As Solved

peerreynders

peerreynders

In a sense ConsumerSupervisor + Producer “is a Consumer Producer” where :min_demand/max_demand provides the back pressure regulation for the Producer in front of it and the (B) Producer is regulated by the Consumer behind it. So you could fashion something this way:

A(Producer) -> (Bc(ConsumerSupervisor) ~cast/call~> Bp(Producer)) -> C(Consumer)

  • Bc Consumer portion of B
  • Bp Producer portion of B

Also Liked

peerreynders

peerreynders

Based on my GenStage experiments and reading the ConsumerSupervisor documentation here are my thoughts:

  • ConsumerSupervisor is a Supervisor process which creates a child process for each event that it receives. In general Supervisors are designed to be as simple as possible so that they can focus on one thing: “supervising child processes” (while not taking on any additional responsibility which may cause them to crash) - this one just happens to dynamically create children for any events received. By this nature the ConsumerSupervisor can only really be a Consumer in a GenStage scenario as a supervisor typically doesn’t collect and manage results from its children.

  • Your particular [A] -> [B(1, 2, 3, 4, 5, ...)] -> [C] scenario could be realized with C as a Producer (if there is a (producer-) consumer D). Essentially the children of the ConsumerSupervisor would simply deliver their result to C via GenStage.cast/2 or GenStage.call/3 - i.e. flow to C would be entirely governed by the :min_demand, :max_demand settings on B, the ConsumerProducer.

  • The documentation of init is strange to say the least. There is a ConsumerSupervisor.init/1 callback that is analogous to Supervisor.init/1 - so I would expect ConsumerSupervisor.init/1 function to be analogous to Supervisor.init/2. The text of the ConsumerSupervisor.init/2 function seems very similar to the text of the Genstage.init/2 callback - so I suspect a copy/paste error. The ConsumerSupervisor.init/1 function code simply prepares a tuple containing the supervisor flags and child specifications.

peerreynders

peerreynders

I’m by no means a GenStage expert - I have just happened to play around with it for a few days 4 months ago - and looking at the documentation right now I still find it a bit hard to digest.

For example it’s only when I started to write the code that I noticed that any of the callbacks that can return a tuple that includes [events] can release events to the next stage - skimming through the examples in the documentation you could easily be left with the impression that handle_demand/2 is the place where events are released - when in fact it is only one place where events are released.

If you have events during the handle_demand/2 callback then by all means release what you have that doesn’t exceed the demand. But as a producer any unfulfilled demand has to be “stored” - because the consumer is only going to issue another “demand” if it wants more than it already asked for.

So whenever a producer “acquires events” it can release them immediately provided it has “stored demand”.

So there are situations where handle_demand/2 will simply return an empty event list (because there are simply none available at the time) - and the events are only released later when the producer somehow gets ahold of them - and then they are released as a result of the callback that “delivers” the events to the producer.

Now a consumer is a GenStage that is at very end of the pipeline and is ultimately considered to be the “constraining operation” - that is why it gets to set the demand that propagates up all the way to the producer at the beginning of the pipeline (who has to honour that constraint). So ConsumerSupervisor is inherently designed to be at the end of that pipeline.

My guess is that it doesn’t need to be a producer because conceptually as it is the “constraining operation” back pressure isn’t an issue with any processing stages that follow - they can simply be implemented as GenServers because they can deal with any volume that the ConsumerSupervisor is capable of throwing at them.

Which brings me to the pertinent point - why would you think that you need ConsumerSupervisor feeding another ConsumerSupervisor? It’s a necessary question to ensure that we aren’t running into the XY problem - i.e. there may be a solution to your actual problem that has nothing to do with ConsumerSupervisors.

Where Next?

Popular in Questions Top

jerry
Good day to you all. I have been struggling to get a query involving like and ilike to work. Can anyone assist me on this, please? pro...
New
dokuzbir
Hello, I am trying to convert my lists to string without losing brackets.For start i have 3 map. They look like these buyer = %{ id: ...
New
lk-geimfari
What is most correct way to open, read and parse JSON file with poison? For example if we have example.json file in root of some projec...
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
chensan
I have a User schema with a :from_id field set to type :string: defmodule TweetBot.Repo.Migrations.CreateUsers do use Ecto.Migration ...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? https://hexdocs.pm/ecto/Ecto.Repo.h...
New
kostonstyle
Hi all I want to have a unix time, from the current time plus 1 hour. DateTime.now + 1 hour How to get it in elixir? Thanks
New
Codball
Mix format works fine if run from the cmd. I’ve followed this to facilitate the implementation into VSC which involves downloading an ext...
New
wernerlaude
In AR this is so simple @articles = current_user.articles How to do in Ecto? def index(conn, _params) do current_user = conn.assig...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New

Other popular topics Top

JakeBecker
TL;DR: I’ve just released an implementation of Microsoft’s IDE-independent Language Server Protocol for Elixir. It adds language support ...
1140 51847 244
New
sorentwo
Hello! tl;dr Announcing Oban, an Ecto based job processing library with a focus on reliability and historical observability. After spen...
977 41022 311
New
grych
Hi folks, Few months ago I have announced the proof-of-concept of the library to manipulate the browsers DOM objects directly from Elixi...
639 49522 488
New
quazar
How to set Jason to encode all fields in ecto schema, I don’t care about security and implementing only is taking long list of attributes...
New
lk-geimfari
What is most correct way to open, read and parse JSON file with poison? For example if we have example.json file in root of some projec...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? https://hexdocs.pm/ecto/Ecto.Repo.h...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New
romenigld
I am trying to run a deploy with docker and I successfully runned with this command: docker build -t romenigld/blog-prod . but when I t...
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
magnetic
Hey :wave:t3: Elixir community, I’ve been learning Elixir, and working on some side projects. My editor of choice is VSCode, and althoug...
New

We're in Beta

About us Mission Statement