Throttling Observables

[abersnaze]

We have a problem where multiple code paths are making different calls to a backend service but we want to throttle the number of concurrent requests that each client is making. The problem is that different requests could be different types or without knowledge of the other happening something like this:

Service.getA()
    .subscribe(processResponseA())

I was thinking that it would be cool to have a ThrottlingScheduler and/or a MaxConcurrentScheduler that could limit the number of Observables that are active without having to interfere with the data of any of them and without having to make a new thread pool.

static Scheduler serviceLimiter = Schedulers.maxConcurrent(2, Schedulers.computation());

Service.getA()
    .subscribeOn(serviceLimiter)
    .subscribe(processResponseA())

[davidmoten]

So the completion of the subscribe action upstream of the call to subscribeOn(serviceLimiter) would release back to the pool?

[abersnaze]

I was thinking rx.Scheduler.Worker.unsubscribe() would be when the next Observable in the queue would be subscribed too.

[akarnokd]

With the existing API, you can submit Observables to a PublishSubject+flatMap with maxConcurrency of 2:

    Observable<T1> t1 = Observable.fromCallable(() -> 1).observeOn(Schedulers.computation());
    Observable<T1> t2 = ...

    Subject<Observable, Observable> ps = PublishSubject.create().toSerialized(); // raw Observable

    ps.flatMap(v -> v.ignoreElements(), 2).subscribe(() -> { }, e -> log(e));

    ps.onNext(t1);
    ps.onNext(t2);

[abersnaze]

I was thinking of using mergeMaxConcurrent inside the scheduler implementation. I would map createWorker() to Observable.create and onNext to a subject. Then map worker.unsubscribe() to onComplete(). While this strange Worker + Observable thing is subscribed to it schedules it's actions on the underlying actual Scheduler. I'm not sure if schedule() would be mapped on onNext or if any uncaught exceptions would be mapped to onError. It might be interesting to think about.

Having the user do all the subject stuff would mean significant constraint of all code paths to only make at most one call the service. The result of each service call Observable would need to be returned and sent through a publish subject for throttling. For example how would could this code be rewritten to only run 2 at a time across the whole of the VM.

void foo() {
   return zip(Service.getA(), Service.getB(), Service.getC(), (a,b,c) -> {...});
}

With the max concurrent as a scheduler the Service code and apply the subscribeOn to protect itself without the client being any the wiser of the throttling.

[abersnaze]

I made two versions of the idea. I'm kind of partial to the fine grained version because I think coarse grain throttling will result in deadlocks. For example in the zip in my last comment if getA() and getB() could end up blocking getC() from being subscribed to until getA() completes and deadlocking. The cost is the fine grain throttling creates an Observable for each scheduled action.

Fine grained throttling:
https://gist.github.com/abersnaze/c8c37847bc4bddf13dbc7cc75b99ae95
Coarse grained throttling:
https://gist.github.com/abersnaze/07d995f63ed6100ef60bd9ac980d59ed

[akarnokd]

Closing via #4154 and 1.1.7.