kotlinlang #coroutines

allows the cancellation of blocking operations that are sensitive to thread interrupts (https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-interruptible.html). However in the java ecosystem there are several blocking operations that are not sensitive to interrupts, such as some sockets methods (IINM). In those cases, sometimes there is an alternative way to cancel the blocking operation, other than by interrupts. For instance, on sockets we can cancel a pending operation by closing the socket (from a different thread). In this context 1. Is there anything similar to

runInterruptible

, where the cancelling action is not to send an interrupt but instead to perform a custom action, such as calling a

close

on some object? 2. If there isn't, we could eventually try to build one. For that we probably need to register an

invokeOnCompletion

that triggers on the

cancelling

state and not only on the

cancelled

state. This would allow us to perform an action when the coroutine transitions to

cancelling

. There is an

invokeOnCompletion

that allows that, however it is marked as

@InternalCoroutinesApi

. Is there any alternative public and stable API to do that, i.e., trigger a callback when a coroutine transitions into the

cancelling

state? Thanks.

rudolf.hladik

11/03/2021, 1:22 PM

Hello there, I have a problem with resolution of correct version of coroutines during build. My project is KMM project. It uses a kmm library which uses internally coroutines. That library use version

1.5.1-native-mt

and forces it by

strictly

config. I use this library as

api

gradle dependency in

shared/commonMain

module. The problem is that I also use some dependencies in Android app that also use coroutines but not the

1.5.1-native-mt

branch.

Endre Deak

11/03/2021, 7:11 PM

Hi, what is the right way to kick off an async task but then immediately return with the result of my original operation?

suspend fun callback() { ... }

suspend fun execute(request: Request): Response {
    val response = ... 

    callback()
    
    return response
}

Norbi

11/04/2021, 8:15 AM

`runblocking`'s documentation says: "This function should not be used from a coroutine" My question is how can I detect that some code is running in a coroutine? Eg.

fun main() {

runBlocking {

suspendFun()

}

suspend fun suspendFun() {

normalFun()

}

fun normalFun() {

val runningInCoroutine = ???

check(!runningInCoroutine)

runBlocking {

...

}

Clament John

11/06/2021, 11:17 AM

[✅ Solved]How to start a multiple flows that can be "switched off" when its is not needed. Scenario: 1. A

List<Message>

in the UI 2. messages =

mutableStateListOf<Message>

in viewModel 3. messages depends on two actions

getAllPaginated

search

4. In both cases we observe SQLite, but only one flow should be active at a single instant (user is either viewing all or is searching) 5. Flow is used at SQLite because new messages can come from the network Questions: 1. How to switch between

getAll

and

search

flow 2. When a new search happens the old

search

flow should be cancelled and a new one should start

Gerard Klijs

11/07/2021, 10:14 AM

I created a small project to play with Java API's and coroutines, for a talk in a few weeks. Any feedback is highly appreciated. https://github.com/gklijs/kotlin-coroutines-demo

Stefan Oltmann

11/08/2021, 9:03 AM

Do Coroutines have something like

_Thread_._isMainThread_

which allows me to check that the current code is actually running in a Dispatchers.IO context for example?

✅ 1

K Merle

11/08/2021, 11:59 AM

How viable is it to have global coroutine scope inside

.stateIn

extension?

Clament John

11/08/2021, 12:14 PM

There is a great kotlin compiler doc which explains

coroutine compiled as continuations

and more. It was a

markdown

file in one of the main kotlin github repos (JetBrains/kotlin or Kotlin/kotlinx.coroutines). Does anyone know where it is? I wanted to do some reading on how coroutines work and how they are compiled

Lilly

11/09/2021, 11:07 AM

Maybe a trivial question, but I often see people using

flow

operator for functions returning a single value. What's the purpose of doing this? Are there any benefits by doing this?

Slackbot

11/09/2021, 2:08 PM

This message was deleted.

Michael Clancy

11/09/2021, 3:22 PM

message has been deleted

Ansh Tyagi

11/09/2021, 5:58 PM

Hey! so is I am getting started with multiprocessing and wanted to know if there is something similar to pythons Pool.map() in kotlin?

mcpiroman

11/09/2021, 6:33 PM

val foo = MutableStateFlow(10)
val baz = MutableStateFlow(10)
foo.collect { baz.value = it }

// Somewhere else
foo.value = 20
assertEquals(20, baz.value) // is this guaranteed? i.e. (how) can I expect value to be immediately updated?

Erik

11/10/2021, 9:18 AM

The coroutines guide about exception handling mentions (https://kotlinlang.org/docs/exception-handling.html#coroutineexceptionhandler):

On Android,
uncaughtExceptionPreHandler
is installed as a global coroutine exception handler.

Where can I find that handler, as I'm interested to learn more about it? I can't find it on GitHub: https://github.com/search?l=Kotlin&q=uncaughtExceptionPreHandler&type=Code. I also cannot find it in the Android source: https://duckduckgo.com/?q=uncaughtExceptionPreHandler+site%3Ahttps%3A%2F%2Fandroid.googlesource.com.

✅ 1

Clament John

11/10/2021, 10:21 AM

Q: My MutableStateFlow Didn’t Emit I'm using this code snippet to update

messages

from two flows (using flatMapLatest)

private val searchString = MutableStateFlow<String>("")
    private val messages = searchString.flatMapLatest {
        if (it == "") {
            getAll()
        } else {
            search(it)
        }
    }.stateIn(viewModelScope, SharingStarted.Eagerly, null)

    suspend fun requestSearch(search: String) = searchString.emit(search)
    suspend fun requestAll() = searchString.emit("")

    fun getAll(): Flow<List<Message>> = TODO()
    fun search(search: String): Flow<List<Message>> = TODO()

But when I try

emit

to update

searchString

it doesn't work. Note: The initial condition of

""

was successfully executed.

Olivier Patry

11/10/2021, 5:01 PM

Hi all 👋 I have an issue linked to an Android integration test using

FragmentScenario

but I think it's something that could be discussed without considering such specific setup (let me know if it's not the case). In a JUnit Android instrumentation test, I launch an Android fragment using

FragmentScenario

, and my fragment impl uses

view.doOnLayout{}

(would be the same with

<http://view.post|view.post>{}

). To make my test pass, I need the code being executed in such postponed code. Here comes the link to coroutines: My test is launched on the main thread, once

fragment.onCreate

is called, I'd need to let "app logic" (here Android, but I guess it could be applicable to another use case) work and in particular get access to the main thread. My idea was to suspend the execution of the test to give back the access to main thread to whoever needs it but I can't make it work.

@Test
    fun testMyState() {
        launchMyFragmentFragment(Lifecycle.State.RESUMED) {
            mainCoroutineRule.runBlocking {
                delay(100)
                assertEquals(3, myState.count())
            }
        }
    }

I would have expected that triggering a suspension using

delay

would give back access to main thread to my "app logic" (here

Fragment.view.doOnLayout

) but it doesn't work. Is it, by chance, a use case some know how to address?

Joseph Hawkes-Cates

11/10/2021, 7:39 PM

This may be a basic question as I’m brand new to coroutines, but my Google fu is failing me at the moment. I am starting a coroutine to make a web service call. I would like to make it so that only one instance of this coroutine is running at a given time. Is saving off the job that is returned and checking it’s status before kicking off a new job the correct way to do this or is there a better way?

Clament John

11/11/2021, 7:36 AM

Updating a single item's quantity in a

MutableStateFlow<List<Item>>

does not trigger flow emission. Add / remove of an item triggers an emission. I'm listening to this flow (cartItems) with a

transform

which is never triggered when I modify the quantity of a product within the list. But the value does change in the list (Can confirm because if I click

n times and then

n times recomposition happens. Question How should I update an item

MutableStateFlow<List<Item>>

such that

emission

is triggered and

transform

is triggered?

Vivek Modi

11/11/2021, 10:21 AM

Hey I want to use async call, I have some doubt in async call

suspend fun fetchTwoDocs() =
    coroutineScope {
        val deferredOne = async { fetchDoc(1) }
        val deferredTwo = async { fetchDoc(2) }
        deferredOne.await()
        deferredTwo.await()
    }

and what the difference of

viewModelScope.launch { }

coroutineScope { }

Marko Novakovic

11/11/2021, 1:32 PM

I have two `ViewModel`s that are dependent on data from common class that exposes data with

StateFlow

. I would like to observe data changes but my concern is collecting that

StateFlow

. it never completes so how will it behave if I

collect

it from

ViewModel

? but using

flow.value

does not give me anything over having a property. what am I missing here?

martmists

11/11/2021, 4:59 PM

Why does the latter error when it should be pretty much equivalent to the former?

Ulrik Rasmussen

11/12/2021, 2:20 PM

I have a coroutine dispatcher

that I have obtained by calling

.asCoroutineDispatcher()

on a single-threaded executor service. I want to launch long-running coroutines on this dispatcher from synchronous code. The obvious choice seems to be

runBlocking(d) { launch { ... } }

, but that blocks the thread until the job launched by

launch { }

completes. Can I submit a long-running job to

from synchronous code without blocking the current thread?

TwoClocks

11/12/2021, 6:00 PM

I think Kotlin's coroutines are "stackless", and loom's are reported to be "stackfull". Any thinking from JB on what to expect when loom ships?

nerses

11/12/2021, 10:52 PM

looking at the docs for https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-jdk8/kotlinx.coroutines.future/await.html can some one explain what does “one-shot future” mean in this context. It’s not clear to me when one should use

.await()

.asDeffered().await()

dimsuz

11/13/2021, 12:44 PM

I am porting my library from Rx to Flow and one bug led me to the differences in the

scan

operator:

Observable.just(1)
  .doOnSubscribe { println("Hello") }
  .scan(0) { v1, v2 -> v1 + v2 }
  .blockingSubscribe { println("got $it") }

println("===")

flowOf(1)
  .onStart { println("Hello") }
  .scan(0) { v1, v2 -> v1 + v2 }
  .collect { println("got $it") }

results and question in the thread →

marcinmoskala

11/13/2021, 4:48 PM

It seems that

TestCoroutineDispatcher

is waiting for cancelled tasks.

@Test
fun test() = runBlockingTest {
    val job = launch {
        launch { delay(1000) }
        launch { delay(1000) }
    }
    delay(100)
    job.cancel()

    // then
    advanceUntilIdle()
    assertEquals(100, currentTime) // Should be true, because after 100 everything is cancelled,
    // but it says expected:<100> but was:<1000>
}

I've created an issue here.

zsperske

11/15/2021, 2:05 AM

I’m working on an Android app and I launch a coroutine using GlobalScope at the start of my application, that coroutine then uses a flow to emit a value every 30 seconds for the lifetime of my application. I know that GlobalScope is generally not recommended, what should I be using instead?

lhwdev

11/15/2021, 6:09 AM

Coroutine builder

coroutineScope {}

will return when the provided suspend block and child coroutines end. Instead, is there anything that returns immediately when the block returns, cancelling child coroutines? I thought there would be prebuilt builder. This would be extremely useful on situations like:

coroutineScope { // this suspends until someFlow is finished; needs alternative one
  val collected = someFlow.shareIn(this, ...)
  work(collected)
}

Phil Richardson

11/15/2021, 12:36 PM

Coroutine concepts have been one of the more difficult things to wrap my head around, so bear with me. Are there any good examples of mixing blocking i/o operations in coroutines? I am working with TarArchiveOutputStream from Apache commons-compress and it is not thread safe in the slightest. At the point you want to write something to it, you must create an entry, add it to the stream and close the entry as a single unit, before you can progress to another one of these cycles When I look at the description of

<http://Dispatchers.IO|Dispatchers.IO>

, it does not fit into the use case in the slightest, as you still have a thread pool backing it This is also a point at which coroutines start to confuse me some more. Even if this IO dispatcher was single threaded, any suspending operations that happen can do just that, which could result in improper interleaving of two coroutine instances wanting to create an entry in this TarArchiveOutputStream On the face, it would imply I should use runBlocking to ensure suspending operations don't suspect but get run to completion But that still leaves the fact there can be more than one thread in the IO dispatcher don't this at the same time. That then also implies I could create my own dedicated single thread executor service as a dispatcher Then with any coroutine launched inside of it, use runBlocking here to complete the create entry, write stream close entry elements But then this still tells me I am running afoul of the runBlocking rule that I should not use runBlocking from within a coroutine So overall, are there any good examples of using something like the IO dispatcher for i/o on something that is not thread safe?

Title

Phil Richardson

11/15/2021, 12:36 PM

<http://Dispatchers.IO|Dispatchers.IO>

Joffrey

11/15/2021, 12:48 PM

I think you're mixing up 2 problems here: IO operations and thread-safety/atomicity. The fact that

TarArchiveOutputStream

is not thread safe isn't related to the fact that it's based on blocking IO.

any suspending operations that happen can do just that

Do what? Suspend? Yes, but if you're dealing with a blocking API, there is likely no suspend operation here. But in any case, I think what you're looking for is a coroutine Mutex. You can just do your operations under a mutex associated with your output stream (from any coroutine), and you'll be fine. You can also nest this under a

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

to solve the blocking-IO-in-coroutine problem (but these are really 2 different problems)

Something like this:

val tarArchiveStream = TODO("whatever constructor or method gives you the tar archive stream")
val tarStreamMutex = Mutex()

withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    tarStreamMutex.withLock {
        // create an entry, add it to the stream, close the entry
    }
}

If you need this repeatedly throughout your code, you might want to create a wrapper for the

TarArchiveOutputStream

that contains the mutex inside and provides atomic operations as methods

That then also implies I could create my own dedicated single thread executor service as a dispatcher. Then with any coroutine launched inside of it, use runBlocking here to complete the create entry, write stream close entry elements. But then this still tells me I am running afoul of the runBlocking rule that I should not use runBlocking from within a coroutine

Indeed, you should not use

runBlocking

inside a coroutine. It's true that on a single-threaded dispatcher, it would kind of have the effect of locking, but it's not supposed to be used as a synchronization primitive. Mutex is meant for synchronization, so you should rather use this.

ephemient

11/15/2021, 6:35 PM

often Channel or Flow is a better way of ensuring that certain operations are not interleaved

for example, if you launch one coroutine that reads from a channel and performs the tar operations in a loop, then you can have any number of concurrent writers to the channel and it doesn't matter

Joffrey

11/15/2021, 6:38 PM

Yes it's an option too, using an actor coroutine. But I guess it's more complicated to abstract that away than abstracting a mutex (because of the scoping of that coroutine)

uli

11/15/2021, 8:36 PM

Or simply use a thread.... If you do not want to suspend and do not want any concurrency whatsoever, what do you want to gain from using coroutines?

Joffrey

11/15/2021, 8:37 PM

I believe @Phil Richardson wants concurrency (and maybe even parallelism) for the rest of the coroutine's code

ephemient

11/15/2021, 8:39 PM

one thing to note. even though code in one coroutine runs sequentially, it may still switch threads at any suspend point, depending on the dispatcher. if you need to avoid that (doesn't sound like it matters for the

TarArchiveOutputStream

case, but it might for other APIs use their own recursive locks or thread local variables, for example), then you either need to switch to a single-threaded dispatcher or ensure that there are no suspend points within your "critical section" (can be guaranteed by writing it in a non-suspend fun, for example)

View count: 10