GenStage Semi-Manual backpressure

pba

I am writing a rate-limiter for a code executer using Genstage. My code looks like this [hope the post is not too long ]

        +-------------------+   +----------+
        | ExecutionRequester|   |  Ticker  |
        | :producer         |   |:producer |
        +-------------------+   +----------+
        |      +-------------------+    |
  :exec_request| TickerExecutor    |    |:tick
        +------> :producer_consumer<----+
               +-------------------+
                |
                |   +--------------+
                |   |ResultConsumer|
                +--->  :consumer   |
                    +--------------+

TickerExecutor should consume from both ExecutionRequester and Ticker, and do a zip operation:
Only 1 :exec_request is allowed for 1 :tick event produced by Ticker.

I have implemented :manual subscription handling for the TickerExecutor → ExecutionRequester subscription, but felt like 90% of the code (:min_demand, :max_demand) is reimplementing GenStages :automatic subscription. I’m, weary of doing the same for the Ticker subscription, before asking.
As of now I’m consuming and wasting :tick events when TickerExecutor is starved of :exec_requests
Thus my 2 questions:

Is there a way to announce to GenStage the actual consumption [Edit: or rejection] of events, and not just the delivery via GenStage.handle_events/3?
Is there a cleaner way to do a zip between two subscription event streams on a :consumer, than just going :manual on both subsctiptions ?

Edit: @admins I have found no Tags for GenStage or the Experimental Module. Would it be ok to add them?

4 comments

#genstage #experimental-module

1 1620 4

2016-09-17 15:22:05 UTC

Showing Posts 4 to 1

Show Best Posts
Show All (oldest first)
Show All (newest first)

pba

done: #83

Post #4

josevalim

Creator of Elixir

Well it buys you an easy way to subscribe to a multitude of tickers in a structured (a.k.a. documented by @josevalim ) way .

Sorry, to be clearer, I meant only the ExecutionRequester being GenStage. You should keep the rest!

So If you are undecided about Flow.zip

I will be glad to add that. Can you please open up an issue?

Post #3

pba

Hi,
Thanks for answering!

Right now accepting a handle_events/3 call [EDIT: in a :consumer] is the same has having actually handled the events. In reality the Consmer can wait for events from another subscription. What I would like is to use :automatic subscriptions, but be able to freeze the demand. so that backpressure is built up on ExecutionRequester while i still accept events from Ticker.

Well it buys you an easy way to subscribe to a multitude of tickers in a structured (a.k.a. documented by @josevalim ) way .

I only now realize that Ticker could provide a flexible rate-limiter for Flow pipes. Just zip in a :tick event stream, and you know exactly how many and how often items are processed, by controlling the emission behaviour of the Ticker producer.

Right now i have implemented:

rate per second
maximum item count per interval (e.g 1500 per day for free goolgle geolocation calls)
bucketed handling of [EDIT: :tick event allocation per interval].

So If you are undecided about Flow.zip I hope to find some time to separate that functionality from the Ticker and make a separate lib. Who knows who needs ratelimiting for Flow Pipes.

P.S. See you next week in Sofia

Post #2

josevalim

Creator of Elixir

Is there a way to announce to GenStage the actual consumption

I am not sure it is clear what you want to announce here? Which stage should tell the other stage what?

Is there a cleaner way to do a zip between two subscription event streams on a :consumer, than just going :manual on both subsctiptions ?

So after thinking about the use case you described above, I am wondering if it wouldn’t be better for you to have a ExecutionRequester as a separate process that you request every time there is a tick. With max_Demand of 1 and a zip, GenStage is not buying you anything for the first three processes: it would be quite equivalent to using regular processes.

The only scenario zip would make in GenStage is if we could still have a high demand value (and you would have back-pressure if one of the sources is slower but typically not). If we ever implement something like this, it would be as Flow.zip.

Post #1

— All posts loaded —