kotlinlang #coroutines

Is it possible to set the default dispatcher to Dispatches.Main? I almost never want the default dispatcher and like to run my launch always on the main dispatcher and only switch the context when necessary

mertsimsek

12/23/2019, 11:46 AM

Hi, I am playing with the Flow to download files. I want to start a coroutine and return channel immediately. But channel returned from the ViewModel when startDownloadSendChannel() completes. I hope I could explain.

Untitled

Oleg Siboglov

12/24/2019, 4:16 PM

I would appreciate it if anyone could give me some feedback. I have a class that makes use of

withContext

functions since the majority of the use cases require that the code runs sequentially. However, I have one use case where I need to run two of these functions concurrently and then combine the results. My question is if there is would be something wrong with wrapping the

withContext

functions with an

async

function.

GlobalScope.launch {
        val deferredOne = async { doSomeWork() }
        val deferredTwo = async { doSomeWork() }
    }

suspend fun doSomeWork() = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    // Do some work here
}

jimn

12/24/2019, 9:58 PM

Has anyone built or seen an opensource example of coroutinecontext feature composition in action? the documentation of this feature is scarce, and not much comes from a github search. this seems like the right place for parsers to seat token sources and parse traits and work with less specialized generic components that can use the parts of context they require without elaborate function params.

Big Chungus

12/27/2019, 12:30 PM

How do I execute a blocking piece of code in common module with coroutines? In JVM there's

runBlocking {...}

, but I cannot find anything simmilar on common module.

Thiyagu

12/27/2019, 6:58 PM

How do i stop subscription of the flow. below is my scenario 1. poll data from some non blocking io (possibly http endpoint) 2. process the data 3. acknowledge that I processed the data Now let's say after some specific period, I want to stop polling the data. How to do it? so that my coroutines are not leaked. gist of my tried approach. https://gist.github.com/thiyagu06/974ede71f4ec0da24bee4574450ab6e7

nwh

12/28/2019, 3:19 AM

Can I (and should I) emit to a Flow from multiple threads? For example, mapping a list of objects and making an HTTP request using the IO Dispatcher, and emitting the result one by one.

ids.map { id ->
	async(<http://Dispatchers.IO|Dispatchers.IO>) {
		api.getMember(id) // I want to emit this
	}
}

Or is there a better-suited construct for this, like a channel?

Marc Knaup

12/28/2019, 2:31 PM

Could I theoretically use coroutines to split up long stacks into smaller stacks? Basically like “if function X was recursively called 500 times, suspend so that the next iteration starts with an empty stack”. But don’t suspend every call as that adds lots of overhead. In my case for a recursive visitor pattern could cause a huge stack if several are run in parallel on a huge AST 😄

Marc Knaup

12/28/2019, 2:48 PM

Oh wow, proof of concept 😮 This one runs forever, recursively.

suspend fun main() {
    recurse(1)
}


suspend fun recurse(n: Int) {
    if (n % 1000 == 0)
        delay(1)

    println(n)
    recurse(n+1)

    print("")
}

Question is now how to implement that manually, with

@RestrictsSuspension

and without the 1ms penalty 😄 And how to figure out a good

n % something

as it depends on the stack size usage of each function in the call stack.

Riku

12/29/2019, 8:19 AM

what is the use case/purpose of this?

Riku

12/29/2019, 8:23 AM

I am honestly confused about what dispatchers to use server-side. Everything I have read so far has been focused on Android. Any recommendations for a general guide to when to use what dispatcher?

marcoferrer

12/29/2019, 4:49 PM

@Riku `initialContext`looks like it’s from krotoplus. If you have any questions I’ll be glad to help. Is there a specific use case you’re trying to achieve?

Paul Woitaschek

12/30/2019, 9:53 AM

PSA; don't try-catch

IllegalStateException

because a

CancellationException

is a subclass and it puts your application into an unexpected state if you catch it. Just found out the hard way

Ruslan Demyanov

12/30/2019, 11:03 AM

Hello everyone! The question is, why doesn’t the channel make an “offer” for every “emit” from the list flow?

val channel = ConflatedBroadcastChannel<String>()
launch {
            listOf("1", "2", "3", "4", "5", "6").asFlow()
                .onEach { Log.d("ORIGIN FLOW(onEach()):", it) }
                .collect {
                    channel.offer("Channel: $it")
                }

            channel.asFlow()
                .onEach {
                    Log.d("TEST FLOW(onEach()):", it)
                }
                .flatMapMerge { merge(flow1(), flow2()) }
                .collect {
                    Log.d("TEST FLOW(collect()):", it)
                }

        }

Mark

12/31/2019, 2:58 AM

I vaguely remember reading that it doesn’t make sense to define a suspending function that accepts a

CoroutineScope

as an argument. In my case, I want to create a factory method to instantiate `MyClass`:

suspend fun createMyClass(scope: CoroutineScope): MyClass {
	return MyClass(scope, createMyArg())
}

suspend fun createMyArg(): MyArg {...}

class MyClass(
    private val scope: CoroutineScope,
    private val myArg: MyArg
)

and I would use it from my

AndroidViewModel

like:

viewModelScope.launch {
    createMyClass(this).doSomething()
    // do some more stuff
}

What is the proper way to write

createMyClass()

? For example, how about this?

suspend fun createMyClass(): MyClass {
	return MyClass(CoroutineScope(coroutineContext), createMyArg())
}

Lulu

12/31/2019, 8:34 AM

I'm having a very hard time understanding the difference between

CoroutineContext

and

CoroutineScope

. I did my research and I also read this article by Elizarov: https://medium.com/@elizarov/coroutine-context-and-scope-c8b255d59055 He says they're the same thing but they serve different purposes. He explains both of them in the article but he doesn't conclude. So what's the difference after all? I'm very confused. Just to clarify, I have been using coroutines since they were experimental so I'm not new to the concept, I just started questioning the difference between those two things that I interact with all the time. I know that a scope basically contains a context, but I still don't understand why scopes exist in the first place. Why not just contexts? Or what's the clear distinction between them? (e.g. "the purpose of

CoroutineContext

is ... while the purpose of

CoroutineScope

is ...") I feel I'm very close to the answer but I just can't put the pieces together.

👍 4

liminal

12/31/2019, 3:06 PM

Hi all, trying to wrap my head around this. The

lookupImagesWithAsync()

should be faster since two images are downloaded in parallel, correct?

imageRepository.downloadImage(imagePath)

is a

suspend

function that does its work in

withContext(<http://Dispatchers.IO|Dispatchers.IO>)

Untitled

liminal

12/31/2019, 6:56 PM

I am throwing a

RuntimeException

imageRepository.downloadImage("img-url-3")

only when getting

image3

. Since i am not calling

await()

on it, i was expecting the exception to go unnoticed and not propagate to default

CoroutineExceptionHandler

and not crash my app. Yet the app crashes still. can someone explain what is different about exception handling when

await()

is called vs when it is not? thanks!

Untitled

dave08

01/01/2020, 3:20 PM

or do I need to do

flow { val result = client.download(url); emit(result) }

instead of using

flowOf

Big Chungus

01/02/2020, 10:59 AM

Isn't

org.jetbrains.kotlinx:kotlinx-coroutines-core-native:1.3.3

available for wasm32 target? The dependency fails to resolve for me.

Thiyagu

01/02/2020, 2:01 PM

what is the equivalent of

onErrorContinue

in kotlin flow?

dave08

01/02/2020, 2:28 PM

callbackFlow

, when should I use

cancel(CancellationException("....", e))

and when should I just throw the exception? Is there any difference?

Thiyagu

01/02/2020, 2:49 PM

I'm having the following code. My jvm is not terminating after executing the last statement in the

runBlocking

lambda? Is any of my coroutine is leaking? https://gist.github.com/thiyagu06/9813da5a5b12ad91ce0ad8b26043d867#file-hangingcoroutines-kt-L79

Brian Carbone

01/02/2020, 3:26 PM

how do you create a coroutine scope for the main thread when you're not using Android? there's no Dispatchers.Main

myanmarking

01/02/2020, 4:42 PM

storage.findXXX(userId)
    .flatMapLatest {
        if(it == null){
            flowOf(
                storage.findYYY(userId)
            )
        } else{
            flowOf(it)
        }
    }

myanmarking

01/02/2020, 4:43 PM

is this the correct approach ?

nwh

01/02/2020, 8:47 PM

I have an event loop with its own dedicated thread, which has a suspending method for submitting events (suspends until the event is done dispatching). Let's say I'm hooking a pre-existing event loop into it, like a slack bot library. What should I consider when deciding which thread to submit events from (and therefore suspend)? Or should I submit events blocking from the main thread to keep everything in step with the event loop?

Aaron Stacy

01/03/2020, 3:49 AM

Why does this program not terminate?

suspend fun main() = withContext(Executors.newSingleThreadExecutor().asCoroutineDispatcher()) {}

Even if I call cancel() from within the withContext block, I see an exception logged, but the program just hangs.

Thiyagu

01/03/2020, 12:56 PM

how does scope works in this gist? https://gist.github.com/thiyagu06/549b03512818eac42c123c4bd0456003#file-producerconsumer-kt-L45 .. Both

producerContext

and

consumerContext

will have same

CoroutineScope

?does

flowScope.cancel()

make both coroutines context not to accept any more task in it?

Title

Thiyagu

01/03/2020, 12:56 PM

how does scope works in this gist? https://gist.github.com/thiyagu06/549b03512818eac42c123c4bd0456003#file-producerconsumer-kt-L45 .. Both

producerContext

and

consumerContext

will have same

CoroutineScope

?does

flowScope.cancel()

make both coroutines context not to accept any more task in it?

tseisel

01/03/2020, 1:20 PM

producerContext

and

consumerContext

are 2 independent contexts : one is backed by one thread, the other has a pool of 2 threads. When collecting from the flow, the code from

channelFlow

will be executed on the producer thread, and each element is notified on one of the consumer threads. Cancelling the

flowScope

results in : 1. Cancelling the consumer coroutine (the one started by

launchIn

) 2. Cancel collection of the flow, and therefore the block in

awaitClose

it called on the producer thread.

Thiyagu

01/03/2020, 1:26 PM

Thanks for the explanation. I understand that

flowScope.cancel()

is enough stop the flow and both coroutineContext will be stopped. both my context will be having the same scope?

tseisel

01/03/2020, 1:55 PM

Some vocabulary here : A

Job

is an unit of work that has a lifecycle : started, stopped, cancelled. A

CoroutineDispatcher

describes how coroutines are executed and resumed. It abstracts threads away. A

CoroutineContext

is the sum of multiple properties. `Job`s and `CoroutineDispatcher`s may be part of a

CoroutineContext

. A

CoroutineScope

is the primitive for structured concurrency. It simply wraps a

CoroutineContext

, and cancelling a scope indirectly cancels jobs that are part of its context. From those definitions, we deduce that : • being based on `CoroutineDispatcher`s, both

producerContext

and

consumerContext

will not stop, their thread pool will still be allocated. • Because the producer coroutine in

channelFlow

is launched in the scope of its

ProducerScope

, it is automatically cancelled when the flow collection is cancelled (you don't need to manually cancel the job in

awaitClose

)

👍 1

Then yes, cancellation of the coroutine that called

collect

(or

launchIn

in this case) will cancel any upstream operations, including the producer.

Note : you can simplify the

channelFlow

block by replacing it with the following :

flow {
    while(true) {
        emit(Random.nextInt(1000)
        delay(100)
    }
}.buffer()

Thiyagu

01/03/2020, 2:36 PM

Hey again thanks for the nice explanation of each concept. I update the gist based on your inputs. https://gist.github.com/thiyagu06/549b03512818eac42c123c4bd0456003. The reason i'm using using channelFlow is, I will replace the random number generation with http request.

Basically, I tried to achieve the below things. This will be independent JVM process which will run in background. When I stop the background thread, all coroutines must stopped properly. 1. poll indefinitely http endpoint and get the data in batches(say 10 per batch) 2. do some async process and store it in DB 3. acknowledge that received data is processed successfully.(call http endpoint)

View count: 5