Has anybody written something like a concurrent FI...

# coroutines

[Marc Knaup]

Has anybody written something like a concurrent FIFO processing queue with a dynamic concurrency limit?
I want to process operations in parallel but the desired concurrency changes over time.

Example:
I throw 50,000 elements at once into the queue.
The concurrency starts with 100, so up to 100 elements will be processed in parallel at any given time.
After a while the error count increases, so I want to dynamically reduce the concurrency to 20.
The number of elements processed in parallel will go down from 100 to 20 as the processing for more elements completes.
Once the error count is back to zero for a little while, I change the concurrency to 100 again.
Now there will be a sudden jump from 20 elements processed in parallel back to 100.

The sender that puts something in the queue should be blocked and the result returned or an exception be thrown.

[Zach Klippenstein (he/him) [MOD]]

I don't know if any prefab implementations of this exist, but certainly seems like it wouldn't be too hard to implement with the standard channel + worker coroutines pattern.

You'd just need your own poison pill value to send to the channel n times to kill n workers without cancelling active tasks.

[Marc Knaup]

A “poison pill” sounds good in theory 🤔
I just have to make sure that it’s delivered before other pending elements in the channel.
I’ll try it out with using select 🙂

[Dominaezzz]

Could also use a Semaphore and acquire permits to limit concurrency.

[Marc Knaup]

So I would have to “hot-swap” them if I want to change the permit count?

[Zach Klippenstein (he/him) [MOD]]

You could use a secondary channel to deliver poison pills, and put it before the primary channel in a select so that it always gets read from first.

[Marc Knaup]

Yeah that’s what I’m doing now. Will share when I have something useful 🙂

[Marc Knaup]

it works 🙂

https://gist.github.com/fluidsonic/ba32de21c156bbe8424c8d5fc20dcd8e

Only downside with the kill approach is that if I set the limit to 100,000 it’ll instantly start 100,000 coroutines.

[Marc Knaup]

It’s probably a bit messy with the coroutine contexts/scopes. I still haven’t figured that out 🙄

[Marc Knaup]

E.g. it feels weird that I pass a parent context and make it Closeable.

[spand]

Sounds like a Circuit breaker. I think resilience4j have a kotlin implementation/wrapper that a coworker is implementing for us as we speak

[Dominaezzz]

So I would have to “hot-swap” them if I want to change the permit count?

No you would just have an high initial permit count like Semaphore(100) . Then if you want to limit concurrency by 15, call acquire 85 times, and adjust concurrency by releasing and acquiring permits.
(Unless I misunderstand what you mean by "hot-swap")

[Dominaezzz]

Since Semaphore is fair, you might want to acquire permits concurrently.

[Marc Knaup]

Makes sense. However I consider 100 quite low and prefer to have a solution that can easily go several orders of magnitudes higher.