Usage of rate-limiter example is unclear

[TylerPachal]

👋 Hello again,

The rate_limiter.exs example is very useful for understanding manual demand, but it is not clear how that RateLimiter would fit into a larger pipeline, and if it is intended for production usage.

I understand that because GenStage is "pull based" meaning the RateLimiter is actually rate limiting whatever is upstream from itself, in the case of the example, a single Producer. But what if there were multiple Producer instances? I believe the intention is for the RateLimiter to subscribe to all of them like this:

Does that continue to work if there are 100 producers? 1000 producers?

Additionally, all of the examples I have seen are based on stages which make external (HTTP) calls, this is the first example I have seen where the stage makes no external calls. I guess it needs to be a GenStage because it is stateful.

What if I wanted to create a GroupBy stage, would that also be its own GenStage? It seems unnecessary for that to be its own process - perhaps that logic should be in my producer or my consumer? I know that Flow "fuses" certain "steps" together into single processes, so maybe the bottom example of this diagram is what I should aim to do?

Sorry if that is covered somewhere else, but I haven't seen any examples of grouping/filtering/reducing in GenStage.

Thanks!

[whatyouhide]

Hey @TylerPachal, thanks for asking some questions that will likely lead to better documentation 😌

Starting with the GroupBy stage: do you want to group events coming from different kinds of producers before feeding them to consumers? If so and if I get the problem correctly, then whether or not you need a separate stage depends on the specific case I think. If you want to, for example, group by some properties and then route to different consumers, the you want a separate stage (provided you can't do this with a partition dispatcher).

As far as the rate limiter goes, scaling to many producers is a matter of usual OTP process scaling: how many messages do you want to pass around? How big do you want the stored pending demand to get? Do you think it would be helpful to add clarification to the docs around this?

[TylerPachal]

Thanks for the quick response - and sorry for overloading my questions with additional questions. I would fix the docs myself but I want to make sure I am understanding everything correctly first!

do you want to group events coming from different kinds of producers before feeding them to consumers

I am trying to model something like this:

So yes I think a separate stage. But I guess I will have to learn how to emit my own batches of events because this would not quite do what I want:

# Batch of events comes in - however big it is does not matter, just do our 
# best to group the events.
def handle_events(events, _from, state) do
  {group1, group2} = Enum.split_with(events, hash_fun)
  {:noreply, [group1, group2], state} # This wouldn't work, well, events would now be groups of events which I don't want
end

Once I get a better implementation of this GroupBy stage I may add an example to /examples because I think that would clear up a lot of confusion in the future. But initially I was scared off because it felt like an instance of When not to use a GenServer.

Do you think it would be helpful to add clarification to the docs around this?

Yes I think that would be helpful for the RateLimiter. What I think is missing is either a statement about usage in "the real world". The implementation of the RateLimiter uses a state called producers (emphasis on the plural) but the example doesn't match up with that. Perhaps changing the example to use multiple producers and to add your comment about the "usual OTP process scaling" would be useful.

At first glance it felt weird to me that a single stage could potentially fan-in and fan-out (if RateLimiter happened to be a producer_consumer) because visually it looks like a bottle neck, but because that stage would be so much faster than the others I see that it could be okay.

[whatyouhide]

@TylerPachal for the use case you show, you can use GenStage.PartitionDispatcher somewhere. The Kafka producer can use this dispatcher so that it can hash events and send events with the same hash always to the same HTTP consumer stage. The consumer stage can then do the batching in-process. Maybe that would work?

On an unrelated note, this seems like a pretty good use case for Broadway too, in case you didn't come across that yet 😉

In any case, is there any action to take on this issue? Do we have a reason to keep it open?

[TylerPachal]

No I think this is good to close - thanks for your insights!