You don’t need sleep in this case

just use thread pool that limits ammount of threads

CommonPool is coroutine dispatcher that uses thread pool under the hood

I think you had OOM because run too many threads, each thread consumes significant ammount of memory

Could you give me an example for using thread pool?

It’s just my guess

If you use coroutines with thread dispatcher you already use thread pool

Of course, I cannot give good advice without real code. General advice: avoid locking, avoid too many threads

In real case, I use less than 5 threads (dispatch topics into different processes) in the production environment. But each thread may read a huge amount of messages, if the sink process can not consume these messages, that cause a OOM.

but how sleep can help you?

if it can, just use delay, at least it doesn’t lock thread

If these is only one thread in the process, too many messages in the queue may also cause OOM

just use delay then

Thanks for your advise 🙂 @gildor

do you use some loop with delay inside?

No, will this cause any unexpected behaviour?

No, I’m just asking. So you just run thread and wait? Why not add some

while(true)

that checks queue size with or without delay

@sailxjx You could maybe look into

produce { }

, you can give it a buffer size, and when full, it will suspend the message requesting until its consumer is ready to receive more, so all you need is a simple

consumeEach{}

on the producer to process the messages...

@dave08 wonderful idea, I’ll try it.

Send the messages via a coroutine channel. Instead of concurrent queues. The channel can be limited in size and will suspend your coroutine until more space becomes available. No blocking, no sleep, no delay: https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#channels

@uli

produce {}

is backed by a Channel, from the original code it looks like needs to make multiple calls to an API and have to handle them downstream with back pressure, but upstream doesn't need that. I think a Channel should be avoided just like an rx Subject, when

produce

is enough... Unless I'm misunderstanding something?

@dave08 from the initial 100 threads thing, i was thinking the a consumer needs multiple producers to not starve. I would not know how to feed the

produce

from multiple API calls

With channels you can go for fan-in/fan-out: https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#fan-out

If

produce

is enough, I agree with @dave08. Then go with

produce

@uli @sailxjx I don't know kafka, bit it could even be like, from the sound of the use case:

fun getTopics() = buildSequence {
      yield(kafka.getTopic()) // This suspends until next value is requested
}

But then there's no buffer... You might be right @uli...

you can try with vertx coroutines and the vertx kafka client

so I have this coroutine

launch(CommonPool) {
   withLoading(view.show,view.hide) // <<-
     val result = ApiClient.getSomething().awaitResult()
     doSomethingWithResult(result)
}

I'd like to compose this

withLoading()

method in a way that it executes the

view.show()

at the start of the parentCouroutine and the

view.hide()

only after the

api call

with the

awaitResult()

finishes, but I'm struggling to undestand/do this 😞 any help is very welcome

I could simply do this

launch(CommonPool) {
    view.show()
    val result = ApiClient.getSomething().awaitResult()
    doSomethingWithResult(result)
    view.hide()
}

and it works just fine, but I would like to abstract this to avoid writing this show/hide every time

@leosan

inline fun withLoading(before: () -> Unit, after: () -> Unit, block: suspend () -> Unit) { before(); block(); after(); }

Then run like this:

withLoading(view::show, view::hide) { ... }

I think that should do it?