I have code that basicaly looks like this:

someFlow.collect {
  Log.d(TAG, ">")
  // Some stuff
  Log.d(TAG, "<")
}

So, I should in Android's logcat alternating arrows going left and right. And in most cases, I do. But on some rare occasions, I see 2 going right and 2 going left. Any idea why that might happen?!

I have a question, I’m trying to replace promises with coroutines but I can’t figure out a way to create a

Deferred<Int>

and then call it and complete it later by another function (Will add an example in a 🧵)

I have an actor, that delivers messages to 4 sub actors. How can I ensure synchronisation? Why do I have sub actors? To avoid deadlocking when needing to call a different internal method, This is the code is anyones curious https://github.com/ghosalmartin/AnySubscripts/blob/master/src/main/kotlin/Store.kt Thanks

Hello folks, I have a question regarding

suspendCancellableCoroutine

and

Mutex

suspend fun awaitCallback(): T = suspendCancellableCoroutine { continuation ->

    val callback = object : Callback { // Implementation of some callback interface
        override fun onCompleted(value: T) {
            // Resume coroutine with a value provided by the callback
            continuation.resume(value)
        }
        override fun onApiError(cause: Throwable) {
            // Resume coroutine with an exception provided by the callback
            continuation.resumeWithException(cause)
        }
    }

    // Register callback with an API
    api.register(callback)
    // Remove callback on cancellation
    continuation.invokeOnCancellation { api.unregister(callback) }
    // At this point the coroutine is suspended by suspendCancellableCoroutine until callback fires
}

suspend fun execute() {
    val mutex = Mutex(false)
    
    mutex.lock()
 
 	try {
   		awaitCallback()    
    } catch(e: Throwable) {
        throw e
    } finally {
        mutex.unlock()
    }
 
}

In the above setup, the moment suspension happens at

suspendCancellableCoroutine

mutex in

execute

method gets unlocked part of the

finally

block

Hello ! Is there a difference (except style) between the "first" and "second" way to associate the 3 flows? If yes, what is the preferred way, performance-wise ? To me, the second way is preferrable only because it's easier to read. But maybe the first one is more performant ?

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val flowA : Flow<Int> = flowOf(1, 2, 3)

    flowA.flatMapLatest { a: Int ->
        doSomeMemoryIntensiveStuffForFlowB(a).map { b: String ->
            doSomeCpuIntensiveStuff(b)
        }
    }.collect { c: Long ->
        println(c)
    }
    
    
    flowA.flatMapLatest { a: Int ->
        doSomeMemoryIntensiveStuffForFlowB(a)
    }.map { b: String ->
        doSomeCpuIntensiveStuff(b)
    }.collect { c: Long ->
        println(c)
    }
}

fun doSomeMemoryIntensiveStuffForFlowB(a: Int): Flow<String> {
    // Some Memory intensive stuff related to "a" (object creation, etc)
    var memoryGreedyList = listOf<Int>()

    repeat(10_000) { memoryGreedyList = memoryGreedyList + listOf(it) }

    return flowOf(a.toString(), a.toString(), a.toString())
}

fun doSomeCpuIntensiveStuff(b: String): Long {
    // Some CPU intensive stuff related to "b" (mapping, etc)
    repeat(10_000) { b.toDouble() * it }
    
    return b.toLong()
}

https://pl.kotl.in/sOl65cZgZ

Can somebody explain why the coroutine context of a

suspend fun main

is considered inactive?

suspend fun main() {
    println(currentCoroutineContext().isActive) // prints false
}

https://pl.kotl.in/sVFnLz9mM

Kotlin Native - Suspending Functions - Linux x64 For Linux (x64), we are building a library (C++) via Kotlin Native. In the resulting binary, all suspending functions get removed. Does anyone know how to make these functions available in the resulting binary?

Are there are drawback to the

currentCoroutineContext()

function? I want to cancel my listener when my coroutine is no longer active. I don’t have access to the scope, but I can use

currentCoroutineContext()

since i’m inside a suspend function. Could this have any negative impact? Like memory leak or a false-positive?

private suspend fun doEffects(colorChangeCallback: OnColorChangedCallback, valueAnimator: ValueAnimator) {
    val context: CoroutineContext = currentCoroutineContext()
    valueAnimator.addUpdateListener {
        if (context.isActive) {
            // Do your work
        } else {
            // Cancel other work
            valueAnimator.cancel()
        }
    }
}

Hi folks. Is there any reason `launch`ing a new coroutine in a cancelled scope doesn’t error at all, and instead fails silently?

I'm trying to asynchronously read files and wrap it inside a coroutine That part I got working with AsynchronousFileChannel, but if you try and use AsynchronousFileChannel for files that are inside a jar file (src/main/resources), it throws an UnsupportedException What are the options for asynchronously reading files from inside a JAR other than literally copying all the files out of the JAR on startup and reading it asynchronously from the file system?

Are coroutines really thread safe? As far as I understand this code can potentially print

0

. Because

i

can be modified from different threads and kotlin compiler doesn't mark it as

volatile

under the hood

In a coroutine, I’m using the

coroutineScope {}

, calling several async methods, and collecting the deferreds into a

MutableList<Deferred<Any>>

. Before the coroutine is over, I call

deferreds.awaitAll()

on the list. This isn’t actually necessary if I don’t care about the results, right? A parent suspend function won’t complete until all it’s children coroutines have completed, right?

Hello there, apart from

delay

what other functions actually suspends? I am trying to get an example of a function that actually suspends (like delay), Not just marked as a suspend function and goes into IO operations like many I have seen

Am looking at some code like following....a function in a view model that creates a

StateFlow

using

stateIn

(which is called/collected from Fragment)...seemingly it work but feels like there's an issue, or at least some redundancy here but can't put my finger on why right now? (reason that's it's a suspend function instead of property is that it needs to call

suspendFunctionReturningFlowB

here but probably other ways around that....)

suspend fun someFunction() = flowA.combine(suspendFunctionReturningFlowB()) { valA, valB ->
    <some logic>
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), null)

Hi all, is there any thing bad with a `collectIn`extension on Flow?

inline fun <reified T> Flow<T>.collectIn(scope: CoroutineScope, collector: FlowCollector<T>): Job {
    return scope.launch {
        collect(collector)
    }
}

It is so obvious, that I was thinking there must be something fundamentally broken if it is not part of the coroutines library.

Is there a feasible way to terminate a process if it takes too long to execute using Coroutines? Currently I have developed a project that simulates a Serverless platform running two Serverless Functions. The platform terminates the function (process) after intercepting its output from a text stream (a maximum output limit is applied). Serverless Function code (truncated):

// ...
if (args.isNotEmpty()) args.forEachIndexed { pos, item -> println("Arg ${pos + 1}: $item") }
    println("Ping")
    println("Pong")
    sleep(4u)
    println("Another item...")
    while (true) {
        usleep(500u)
    }

The function for running the Serverless Function (on the Serverless Platform side):

private fun CoroutineScope.runFunction(filePath: String, vararg args: String) = launch(context = Dispatchers.Default) {
    println("Running Serverless Function...")
    val file = openProcess(filePath = filePath, args = args)
    file.readLines(5u).forEach { println(it) }
    val processName = extractProcessName(filePath)
    val pid = fetchLatestPid(processName).toInt()
    if (pid > 0) kill(pid, SIGINT)
    file.closeProcess()
}

Is there an easy way to start a coroutine from a non-suspend function? E.g.

fun justSomeFunction() {
  launchACoroutine(<http://Dispatchers.IO|Dispatchers.IO>) {
    while (true) {
      // just do some work
    }
  }
}

coroutineScope {
    println("Entered scope")
    launch {
        println("why not??: ${Thread.currentThread().name}")
    }
}

How can I debug this printing nothing at all? I do have a non standard CorountineContext: https://srcb.in/y7TGICTNxq However, it seems to work with everything else so I'm not sure what I'm doing wrong here.

A question: I have several coroutines (~100) sending a message to a channel. This channel reads the messages in a

for( message in myChannel){ ... }

loop. I can see, via logging, that the events are being send and I see that they are read in the for loop. However, all except for one of the sending coroutines do not continue and keep being stuck in the suspended state. Any idea what could cause this ? Adding a buffer to the channel seems to help, but I would like to know what causes my issue ?

If I have a StateFlow, what is the best way to map it to another StateFlow? I get a Flow back from

map

, but if I want it to be a StateFlow again I need to use

stateIn

but then I require a scope. Or is the mapped result still a stateflow? The usecase is I have a class that I want to expose a StateFlow which is based on another one. But if I want to use

stateIn

I need to do this from a coroutine. So it becomes something like this:

interface History {
    val activeLocation: StateFlow<Location>
    // ...
}

class Router<R>(
    private val history: History,
    private val routing: Routing<R>,
    private val scope: CoroutineScope
) {
    lateinit var activeRoute : StateFlow<R>
        private set

    init {
        scope.launch {
            activeRoute = history.activeLocation
                .map { location -> routing.getRoute(location) }
                .stateIn(scope)
        }
    }
}

I am not sure if this is the most optimal way of doing it... I have the feeling I am overcomplicating stuff.

I am currently developing a desktop application, and I have taken an MVVM approach. A ViewModel has no lifecycle in the desktop world, which makes me wonder, if I should inject a coroutine scope, or dispatcher to it. What is the best way to go here?? Should I create a long lasting coroutine scope? or just inject a dispatcher and use withContex while invoke network requests when needed to?? N.B: I want to inject one of these so that I may be able to test it as well

Hello everybody :) I am trying to upgrade coroutines up to 1.6.2, but getting an error like this in any project I tried to update. Is there any info about this?

Why is something like this:

val ctx = currentCoroutineContext()
val test = object : CoroutineScope {
    override val coroutineContext = ctx
}
task = test.launch(context) { block() }

Not the same as:

coroutineScope { 
 task = launch(context)
}

I know the first one is bad practice so are they both pretty bad?

I am struggling a bit with some fundamentals I think. I am wondering what is a proper solution for the following: I am creating some sort of Redux architecture, and am now at the part where I am handling effects. These effects should be handled in separate coroutines. Now, the question is as follows. I can implement my

EffectDispatcher

two ways:

// as a suspending function
interface EffectDispatcher<E> {
    suspend fun dispatchEffect(effect: E)
}

// return a job
interface EffectDispatcher<E> {
    val scope: CoroutineScope
    fun dispatchEffect(effect: E): Job
}

As far as I understand: In the suspending case (1st) it is the responsibility of the callee to provide a coroutinescope, and in the case of the 2nd the EffectDispatcher itself has a scope in which the effects launch and the callee gets a reference to the dispatched effect job, maybe to cancel it or something, or can cancel every running job with

scope

. Is this correct? As my preference goes I'd go for the 2nd option. EDIT: Option number 3 would be to have

EffectDispatcher<E>

implement

CoroutineScope

. I like that one best actually.

is there a problem using

ConcurrentHashMap

with coroutines? I see 2 potential issues, assuming that I want to compute a value

computeIfAbsent

. I don’t think using

getOrPut

is an option since it is not concurrent and the value may already exist: 1. If my compute function is suspending (can be

runBlocking

with parallel computation to speed the process) then another coroutine (on the same thread) can run the same

synchronized

code. I think that is a problem 2. If I am running on a dispatcher that runs multiple threads and my compute function is suspending again then I can find myself waking up in a different thread then the one that I started in (can that happen?). Will it break the synchronized block inside the

ConcurrentHashMap

? I am not sure what happens here. Is there anything that is coroutine safe equivalent to

ConcurrentHashMap

or do you know if there is something planned to solve these issues? (if what I described are issues)

Is there an equivalent of

collectLatest

that can be followed by

launchIn

instead of wrapping it in a

launch

block? For now, I copied the implementation of

collectLatest

without the

collect

call but I was wondering if something already existed.

I'm to figure out the best way to maintain a semblance of structured concurrency while using an async cache. Currently what I have is a Caffeine AsyncCache, using which I do roughly this:

suspend fun get(key: K): V = supervisorScope {
    asyncCache.get(key) { k: K, _ -> future { loader.load(k) } }.await()
}

The problem I'm having is with cancellation. If multiple consumers are simultaneously waiting for a given key, and one of them stops waiting (i.e. is cancelled), the current implementation will have all the consumers receive a cancellation exception. I can change that by loading the key in a separate scope (e.g. a dedicated CoroutineScope belonging to the cache, rather than a scope belonging to one of the consumers) and replacing

await()

with

asDeferred().await()

as suggested by the docs. But then I have a problem where nobody can cancel the job. Even if all the consumers stop waiting, the cache entry will continue loading. To handle that problem I would need to introduce some lifecycle management into the cache, e.g. make it

Closeable

and have it clean up its own coroutines on close. Has anybody encountered this before and found a nice solution?

can some explain me the following?

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import sun.misc.Signal

suspend fun main() {
    withContext(Dispatchers.Default) {
        Signal.handle(Signal("INT")) {signal ->
            println("RECEIVED ${signal.name}")
            this.coroutineContext.cancelChildren()
        }

        launch {
            withContext(Dispatchers.Default) {
                launch {
                    while (true) {
                        try {
                            println("foo")
                            // delay(2000) // if I put delay here, it fails with an infinite loop on SIGINT
                        } catch (e: Exception) {
                            println("error: ${e.message}")
                        }
                        delay(2000) // if I put delay here outside try-catch, no problem on SIGINT
                    }
                }
            }
        }
    }
    println("finished")
}

Is it possible to make a CoroutineContext.Element that "falls off" in scopes that branch execution? For example:

runBlocking(MyContext) {
  launch(<http://Dispatchers.IO|Dispatchers.IO>) {
   //doesn't contain my context anymore
  }
  launch {
   //still contains my context
  }
}

I need to be an identifier that indicates that a call is part of the same sequence of calls. Worst case even if I could just get it to log some kind of warning that would be a good start.