kotlinlang #coroutines

Trying to better understand when coroutines switch threads when moving from Default to IO and back. And when the thread should be reused, to try to better optimize what can be. I'm building some POC so some things should be implemented differently probably and that will explain the switches, but understanding what will help building the migration path. So I have a coroutineContext on Dispatchers.Default that call a

launch {
	T1
	val result = function()
	T2
}

function is a suspend function that enqueue commands in a worker pool that works on Dispatchers.IO

supend function () : Result {
	val result = workerpool.execute()
	return result
]

The actual coroutine that run in the worker pool is just

doWork () : Result {
        try {
			T3
            task.response.complete(response)
			T4
        } catch (e: Throwable) {
            task.response.completeExceptionally(e)
        }
]

And here's the threads the different parts run in on different runs.

1)
T1: Pre queue 81396 DefaultDispatcher-worker-5
T3: Pre task 81396  DefaultDispatcher-worker-5
T4: Post task 81396 DefaultDispatcher-worker-5
T2: Post queue 81366 DefaultDispatcher-worker-1

Why is T2 switched to another thread? I suppose the IO thread is asked for removal before the end of the coroutine to keep the number of default thread constants? Is this normal and efficient?

2)
T1: Pre queue 81375 DefaultDispatcher-worker-2
T3: Pre task 81417  DefaultDispatcher-worker-7
T4: Post task 81417 DefaultDispatcher-worker-7
T2: Post queue 81423 DefaultDispatcher-worker-12

Here the switch to IO really switched to another thread. Should not the thread be converted to IO as per the doc? And same answer at the end why is there's another switch Without looking at the thread switching all works and fast, just wondering if there's something I miss as I suppose on larger scale the switch can have impact and from the docs it seemed some should not happens.

Paul Woitaschek

11/06/2018, 1:56 PM

Is there sth like

ReceiveChannel#distinctUntilChanged

jivimberg

11/06/2018, 3:40 PM

Hi guys! I’m wondering if there’s a way of writing this code so I can schedule all operations and exit the

map

block (i.e. get to line 9) and then wait for the result. Right now the

await()

on line 6 is preventing me to do that. Using

CoroutineStart.LAZY

works but then I don’t get the nice parallel execution because each async only starts when it’s needed.

Pipelining_async_await.kt

Dias

11/06/2018, 4:26 PM

Has anyone successfully used kotlin corotines with kafka consumers? We currently spin up a thread to consume from a topic, we end up spinning 30ish threads so far, which are not too busy most of the time. I want to convert it to a coroutine usage, but when you consume from a topic you need to commit offset from the same thread and committing should be done in order, I think, which will be a problem if I just send them to be executed in a threadpool. What approach can I take?

🤔 2

Paul Woitaschek

11/06/2018, 7:59 PM

* **Note**, that elements that are sent to this channel while there are no
 * [openSubscription] subscribers are immediately lost.

napperley

11/06/2018, 9:07 PM

Seeing a strange TypeError appear in the Browser console that seems to be coming from the KotlinX Coroutines Core JS library:

kotlinx-coroutines-core.js:19616 TypeError: Cannot read property 'iterator' of undefined
    at addAll (MutableCollections.kt:127)
    at GroupStatistics.mainLanguage_e0v785$ (_Arrays.kt:8034)
    at updateStatsLayout (main.kt:93)
    at Coroutine$processFetchedRepos$lambda.doResume (main.kt:85)
    at Coroutine$processFetchedRepos$lambda.CoroutineImpl.resumeWith_tl1gpc$ (CoroutineImpl.kt:47)
    at CancellableContinuationImpl.DispatchedTask.run (kotlinx-coroutines-core.js:2682)
    at WindowDispatcher$queue$ObjectLiteral.MessageQueue.process (kotlinx-coroutines-core.js:19877)
    at kotlinx-coroutines-core.js:19846

groostav

11/07/2018, 12:13 AM

hey guys, what happened to the CommonPool in 1.3 (kotlinx 1.0)? I'm using it in a bunch of places for some computation and in a couple places serialization, what should I use instead?

chi

11/07/2018, 1:51 AM

I'm trying to implement coroutines on my app, replacing async some async tasks. I have the following block

countries.add(country)
launch(Dispatchers.Main) {
                        val exp = runExpensiveCode(country)
                        countries.remove(country)
                        if (exp != null) doAction()
                    }

suspend fun runExpensiveCode(country: Country) =
            withContext(Dispatchers.Default) {
                performVeryExpensiveAction(country) // legacy code used in several parts
            }

This still freezes my main screen for some milliseconds, but the Async task implementation doesn't, how can I make this as seamless as possible?

Dmytro Danylyk

11/07/2018, 3:57 AM

So rx java has a way to inject scheduler, is there a plan to have similar thing for coroutines dispatchers? I want to test my code which rely on 3rd party component. This component doesn’t provide way to change dispatcher.

kotlin
// library
class Componenet {

   fun doSomething() {
       launch { // long running operation }
   }
}

// client
fun test() {
    Componenet().doSomething()
   // I want to wait until doSomething is finished but I can't
}

👍 1

➕ 3

sdeleuze

11/07/2018, 9:22 AM

Has anybody tried to compile Kotlin/JVM + Coroutines with GraalVM

native-image

tool? I get a

Non-instantiated type referenced by a compiled method: kotlin.Result$Companion

exception and I don't know if the issue is on Coroutines or Graal side https://github.com/oracle/graal/issues/783

Colin

11/07/2018, 10:37 AM

I need some help understanding how coroutines clean up after being cancelled. I have an example :- I have a function that returns a

ReceiveChannel

, and inside that function I am starting new suspended functions. Pseudocode:

suspend fun produceStrings(): ReceiveChannel<String> {
        return produce {
            send(suspendedFunction1())
            send(suspendedFunction2())
            send(suspendedFunction3())
        }
    }

The code that calls

produceStrings

is consuming the channel via

consumeEach

. If at some point I

cancel

the channel, I am expecting that the suspended functions inside the

produce

block are stopped. i.e. I have a breakpoint in

suspendedFunction3()

and I can see that the breakpoint is hit after I have cancelled the channel. Is this expected behaviour?

yusuf3000

11/07/2018, 10:39 AM

We are attempting to use coroutines with the new kotlin-multiplatform plugin. We are able to use it successfully inside

iosMain

but we cannot use it in

commonMain

. We get a compilation error with

Unresolved reference: GlobalScope

and the line it fails on is

import kotlinx.coroutines.GlobalScope

. Does anyone know why this is occurring and how we can fix this?

plastiv

11/07/2018, 3:32 PM

I'm doing migrate to coroutines refactoring in android application. Can someone help me with reviewing next implementation, please? To see if I has any obvious and nonobvious issues. The current idea (I don't want to rewrite whole app and every activity in one MR) is that it will allow to run multiple network requests in parallel between activity onCreate/onDestroy. Results are delivered through callbacks, yes. I know that this is not idiomatic, but I have to start conversion somewhere 😉

open class RetrofitCallManager( // Open because its mocked at some unit tests. 
    val uiDispatcher: CoroutineDispatcher = Dispatchers.Main, // To be able to switch to Unconfined at jvm tests.
    val ioDispatcher: CoroutineDispatcher = <http://Dispatchers.IO|Dispatchers.IO>
) : CoroutineScope {

    private lateinit var parentJob: Job

    override val coroutineContext: CoroutineContext
        get() = parentJob + uiDispatcher

    fun onCreate() {
        parentJob = SupervisorJob() // Job is supervised so one failed child coroutine wouldn't cancel other running coroutines.
    }

    open fun <T> execute(request: NetworkRequest<Result<T>>, listener: ResultListener<T>) = launch { // Launch returns Job which could be joined inside test runBlocking.
        try {
            val result = withContext(ioDispatcher) {
                request.loadDataFromNetwork() // Blocking retrofit network call.
            }
            if (result.isSuccess) {
                listener.onResult(result.data()!!) // Update ui.
            } else {
                listener.onError(result.error()!!) // Update ui.
            }
        } catch (e: CancellationException) {
            throw e // No need to propagate to UI since execution was cancelled by UI.
        } catch (e: Exception) {
            listener.onError(LocalError.create(ERR_CODE, e)) // Update ui.
        }
    }

    fun onDestroy() {
        parentJob.cancel()
    }
}

Is this something you would expect to see in android app?

dave08

11/07/2018, 4:07 PM

Does

withContext

"swallow" exceptions like

launch

christophsturm

11/07/2018, 4:46 PM

why is TestCoroutineContext marked as obsolete?

quiro

11/08/2018, 10:23 AM

I'm seeing a weird bug with proguard happening when using Coroutines 1.0.1. In my Android project I have both coroutines-core and the coroutines-android but when the build is proguarded I get:

Caused by: java.lang.IllegalStateException: Module with the Main dispatcher is missing. Add dependency providing the Main dispatcher, e.g. 'kotlinx-coroutines-android'

quiro

11/08/2018, 10:24 AM

Disabling proguard makes everything work as expected. The same is happening in Coroutines 1.0.0

matej

11/08/2018, 1:44 PM

java.lang.IncompatibleClassChangeError: kotlinx.coroutines.io.internal.MutableDelegateContinuation
        at kotlinx.coroutines.io.ByteBufferChannel.<init>(ByteBufferChannel.kt:2253)
        at kotlinx.coroutines.io.ByteBufferChannel.<init>(ByteBufferChannel.kt:24)
        at kotlinx.coroutines.io.ByteChannelKt.ByteChannel(ByteChannel.kt:13)
        at kotlinx.coroutines.io.CoroutinesKt.writer(Coroutines.kt:98)
        at io.ktor.util.cio.InputStreamAdaptersKt.toByteReadChannel(InputStreamAdapters.kt:21)
        at io.ktor.util.cio.InputStreamAdaptersKt.toByteReadChannel$default(InputStreamAdapters.kt:20)
        at io.ktor.client.engine.android.AndroidClientEngineKt.content(AndroidClientEngine.kt:117)
        at io.ktor.client.engine.android.AndroidClientEngine.execute(AndroidClientEngine.kt:77)
        at io.ktor.client.engine.android.AndroidClientEngine.access$execute(AndroidClientEngine.kt:21)
        at io.ktor.client.engine.android.AndroidClientEngine$execute$2.invokeSuspend(AndroidClientEngine.kt:35)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:32)
        at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:236)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:742)

Hadi Tok

11/08/2018, 4:02 PM

I am running a unit test on a coroutine with

Dispatchers.Main

and I am getting

java.lang.IllegalStateException: Module with the Main dispatcher had failed to initialize

I can provide

Dispatchers.Main

externally or use Roboelectric but is there anything else that I can do?

Slackbot

11/08/2018, 6:09 PM

This message was deleted.

tateisu

11/08/2018, 6:31 PM

Is there a way to get Coroutine object that current running ? I'm confusd why AbstractCoroutine is also Job and CoroutineScope.

tateisu

11/08/2018, 11:17 PM

In the package kotlin.coroutines.intrinsics,

fun createCoroutineUnintercepted()

make

Continuation<T>

from

suspend code block

and

completion: Continuation<T>

, but how it find coroutineContext for Continuation ?

tateisu

11/09/2018, 6:26 AM

this is full stack trace.

-.scala

Burkhard

11/09/2018, 10:13 AM

message has been deleted

Tolriq

11/09/2018, 4:23 PM

I'm wondering if there's general recommendation / way to measure coroutine context switch cost to make code decision, or if in the global picture (For Android apps in my case on potentially low end devices) the cost are not important enough to base decisions on this as the coroutine engine will always optimize better than doing it manually. A a simple example imagine you have a blocking network call to make and then a multicast lock to release.

launch {
   withContext(IO) { blockingCall() }
   lock.release()
}

This can trigger a switch and one may want to optimize to:

launch(IO) {
  blockingCall()
  lock.release()
}

Is there any interest to do that? Or is it better to keep clear separation between blocking calls on IO and non blocking calls?

Slackbot

11/10/2018, 6:47 PM

This message was deleted.

pakoito

11/10/2018, 7:53 PM

is there any compile or runtime cost associated with making all higher order functions being suspended?

dave08

11/11/2018, 2:02 PM

Apart from using a Channel/Actor, is there any easy way to implement an

ConcurrentHashMap

with coroutines that suspends the current Coroutine instead of blocking the thread? Sometimes retrieval of missing values can take time...

Romain

11/12/2018, 7:20 AM

I have a question about Coroutines on Android. A button in an Activity open a BottomSheetFragment that implement a CoroutineScope, I cancel the parent job on Fragment destroy. A button in the BottomSheet start a Coroutines and do a long running task. Of course if I close this fragment the job is cancelled. But if I implement CoroutineScope in the Activity and cancel the parent job on Activity destroy, can I pass the Coroutines scope/context to the Fragment so that if I start a Coroutine in the BottomSheet and close it the Coroutine is not cancelled?

Title

Romain

11/12/2018, 7:20 AM

tateisu

11/12/2018, 11:50 AM

coroutineContext can have only 1 job. structured concurrency is defined as job's parent/childlen relation. then you can make job on the fragment to child of activity's job.

👍 1

Romain

11/13/2018, 4:44 PM

thank you

View count: 2