kotlinlang #coroutines

. Sometimes I need to access the current value of this channel to perform some fire-and-forget operation . What is then more correct, less error-prone:

fooFlow.take(1).collect {  performSomeSuspendableOperation(it) }

performSomeSuspendableOperation(fooChannel.value)

Thanks guys!

koufa

01/23/2020, 2:56 PM

Hey to all. Quick question. How can I use flow to emit a value every X seconds for example ? Is there any API available or do I have to build a Flow myself ?

Jan

01/24/2020, 12:06 PM

hi i try to learn more about coroutine flows and i try to emulate the equivalent of rx throttleLatest i get confused about switching context for producer and consumer coroutines channels. can someone point me into a direction where to look for it? most flow/channel tutorials don't take callbacks into considerations that are not a suspend function, e.g. i can't call emit() in a callback for some reason

viralshah

01/24/2020, 6:09 PM

is there a good course or tutorial for Coroutines & Flows, I am a bit confused how and when to create which scopes

gotoOla

01/24/2020, 10:05 PM

Is there anyway to make the buffer of the Flow strict? Right now it seems to have the same behavior as channel with +2

taer

01/24/2020, 10:54 PM

Is there a better way to "yield" in a suspend function other than a delay(1)? I have this loop

private suspend fun poll2() {
    while (true) {
      consumer.poll(Duration.ofMillis(500))
        .forEach {
            channel.send("ping")
          }
        }
    }
  }

Channel is a kotlin Channel. If consumer.poll returns empty, the foreach doesn't execute. So we don't ever execute any suspendable code, so this loops forever, not giving back the thread to the coroutine. If I add a

delay(1)

immediately preceeding the poll, then other coroutines get a chance to run. Is there a cheaper alternative to an arbitrary 1ms delay? BTW, this is an attempt to wrap a Kafka consumer output into a easier to consume Kotlin channel.

ursus

01/26/2020, 4:48 PM

Is there a way it can collapse handler.posts so close together somehow? cc @Adam Powell

Paul Woitaschek

01/27/2020, 7:00 AM

Which version would you prefer: a or b? (and why)?

Esa

01/27/2020, 8:32 AM

Hi, I have a

taskChannel

, a

resultChannel

, a producer and multiple consumers. The producer is lightning fast and the consumers need a bit of time. Once the producer is done, it closes the taskChannel.

launch {

tasks.forEach {

taskChannel.send(it)

taskChannel.close()

Then this is the consumer side of things.

launch {

repeat (10) {

launch {

for (task in taskChannel) {

val result = process(task)

resultChannel.send(result)

This resultChannel undergoes one final step, which is logging the results:

for (result in resultChannel) {

log(result)

resultChannel.close()

My question is really just if I’ve closed the channels at the correct times etc. Also, is this a correct usage of the channels, or have I created some antipatterns or some such here? Any replies appreciated. It’s my first time looking into channels, and I like the way they let me start all three processes (production, consumption and logging) at the same time

gotoOla

01/27/2020, 9:45 AM

am I correct in my assumption that channels is at-most-once delivery? E.g. if you have a channel with 5 "processors" that iterates over the channel, none of them will get any duplicates

mingkangpan

01/27/2020, 1:49 PM

I have a question is there any difference between

launch(Dispatchers.Unconfined)

and

launch(start = CoroutineStart.UNDISPATCHED)

? afaiu both will excute job immediatley in the same thread right?

tseisel

01/27/2020, 3:07 PM

What's the purpose of

Dispatchers.Unconfined

? What kind of use-case does it solve ?

marcinmoskala

01/27/2020, 3:59 PM

Looking for a way to track the number of suspended coroutines in a context. My best guess for now is to make

ContinuationInterceptor

returning my custom continuation. Though cannot find an event on coroutine suspension.

interceptContinuation

is called much less often than

resumeWith

dimsuz

01/27/2020, 10:24 PM

Hi! What will happen if I call multiple suspend functions from

launch

like so:

myScope.launch {
  suspendingFunction1()
  someFlow
    .onEach { suspendingFunction2() }
    .collect { 
      someChannel.send(Unit) // 'send' is also a suspending fun
    }
}

Will each of them block? (inside this coroutine)

lupajz

01/27/2020, 10:29 PM

I have a quick question, or more like looking for insight from somebody: I have a ConflatedBroadcastChannel to which I open multiple subscribers each separately as a new Job within parent scope like:

parentScope.launch { sub.consumeEach { ... } } 
parentScope.launch { sub.consumeEach { ... } } 
parentScope.launch(NonCancellable) { sub.consumeEach { ... } }

Now the last one job has extra NonCancellable context, am I correct to assume that the last subscriber won't closed if I would cancel the parentScope ?

paulex

01/28/2020, 9:26 AM

How do i keep the JVM alive to receive connections.... e.g

fun main(args:Array<String>) = runBlocking<Unit>{
   InternalLoggerFactory.setDefaultFactory(Log4JLoggerFactory.INSTANCE)
   val server = AppServer(this)
   server.run()
}

I want to main thread to stay alive until a shutdown...

Timur Atakishiev

01/28/2020, 10:04 AM

someList.forEach {
    GlobalScope.launch {
        if (validator.validate(it)){
            someStorage.add(it)
        }
    }
}

Hi guys, I am expecting that validate function is going to be excecuted in N numbers(N = size of someList) of coroutines. hewever, my application is working sequentialy, first it is validating first element, second element and so on. Should it work sequentially? Or am I doing something wrong?

dimsuz

01/28/2020, 11:51 AM

Hi! What would be a cold equivalent of RxJava's

Single

in coroutines? for example

Single.fromCallable { doNetworkRequest() }

would emit only upon subscription. Should I just use

Flow

which calls a suspending function inside? or is this too complex?

Paul Woitaschek

01/29/2020, 3:39 PM

I inject `CoroutineDispatcher`s in some classes I want to test and have no idea how to get that from a TestCoroutineScope

Abhishek Bansal

01/29/2020, 3:54 PM

Hello Everyone, I recently have started using coroutine based API calls with Retrofit. While testing on slow(EDGE) network scenarios I am getting a series of

SocketTimeOutExceptions

which is fine. I have all my API calls wrapped in

try.. catch()

block like this

try {
viewModelScope.launch {
   apiService.doCall()
   // do other stuff
}
} catch(e: Exception) {}

and in case of flow I am using

.catch()

operator for catching exceptions. My problem is my app randomly crashes with crashstack looking like this.

DefaultDispatcher-worker-1
    Process: <appId>, PID: 6524
    <http://java.net|java.net>.SocketTimeoutException: timeout
        at okhttp3.internal.http2.Http2Stream$StreamTimeout.newTimeoutException(Http2Stream.java:677)
        at okhttp3.internal.http2.Http2Stream$StreamTimeout.exitAndThrowIfTimedOut(Http2Stream.java:685)
        at okhttp3.internal.http2.Http2Stream.takeHeaders(Http2Stream.java:154)
        at okhttp3.internal.http2.Http2ExchangeCodec.readResponseHeaders(Http2ExchangeCodec.java:136)
        at okhttp3.internal.connection.Exchange.readResponseHeaders(Exchange.java:115)
        at okhttp3.internal.http.CallServerInterceptor.intercept(CallServerInterceptor.java:94)

This does not happen with all timeouts but happen at random. It seem like there is something is off with coroutine setup/usage/exception handling. Any pointers will help. Thanks!

myanmarking

01/29/2020, 4:23 PM

the try-catch should be inside the launch block! try it

☝️ 3

myanmarking

01/29/2020, 4:29 PM

@Test
fun `test try-catch outside launch is useless`() = runBlockingTest {
    var result: Int = 0

    try {
        launch {
            throw IllegalStateException()
            result = 1
        }
    } catch (e: Exception) {
        result = 2
    }

    Assert.assertTrue(result == 0)
}

@Test
fun `test try-catch inside launch catch the exception`() = runBlockingTest {
    var result: Int = 0

    launch {
        try {
            throw IllegalStateException()
            result = 1
        } catch (e: Exception) {
            result = 2
        }
    }

    Assert.assertTrue(result == 2)
}

dimsuz

01/29/2020, 5:24 PM

Perhaps an FAQ, but I see that Kotlin/Native has no Dispatchers.IO, why is that? (I tried to google this, no answer) I am an Android dev, so far don't have great knowledge of iOS internals. Also not sure if this should be asked in #multiplatform instead

sbyrne

01/29/2020, 6:51 PM

If I have a flow being read by 2 collectors, the emitter goes as fast as the faster collector, and everything gets buffered until the slower collector reads it. How do I limit that buffer so I get backpressure from the slowest collector?

CLOVIS

01/29/2020, 9:13 PM

Looking at Project Loom (introducing Fibers that look a lot like coroutines to Java), will it change anything in how coroutines work? Will they be reimplemented on top of fibers? Apparently fibers will be able to suspend on blocking IO calls (which coroutines can't, AFAIK)... Will this be taken advantage of in Kotlin?

diesieben07

01/30/2020, 9:09 PM

If I create a

Job

with a parent (parent is long-running), is it a memory/resource leak to not complete or cancel the Job? Will the parent hold on to it and prevent it from being garbage collected?

rook

01/30/2020, 9:34 PM

I’m trying to prove out converting from a

Flow

to an Rx construct. Unfortunately, this never seems to complete.

Untitled

Dave Jensen

01/31/2020, 12:21 AM

I'm trying to run x number of async operations and then combine the results when the async operations are all done. Kinda like Promises.all

Matt

01/31/2020, 5:03 AM

Hello, I'm creating a CommonMark parser (spec.commonmark.com) and the second stage is being done in parallel, the second stage is where inlines are parsed from blocks, I've noticed that if I have a large number of blocks (say 1000) and I launch them into a coroutine about 20% of my runtime is actually switching coroutine contexts. This is obviously not ideal and I was wondering if anyone knew how to make it so that each 'job' is ran until it is completed then the next 'job' is picked up

Title

Matt

01/31/2020, 5:03 AM

octylFractal

01/31/2020, 5:05 AM

that already happens as much as possible, if you're switching dispatchers it'd be quite inefficient and cumbersome to block your previous dispatcher from running anything at all

if you're not switching dispatchers, no suspension happens at all iirc

Matt

01/31/2020, 5:07 AM

this is the flamegraph I am getting, (above the blue is where I'm assuming context switching is happening) from what you said, this would be the completion of one job and the thread changing context to the next job?

private suspend fun parseLeaves(leaves: List<Leaf>){
        coroutineScope {
            // launches all leaves into seperate coroutines to be dealt with as they can
            leaves.forEach { launch { analyzeBlock(it) } }
        }
    }

Is how Im launching my coroutines

octylFractal

01/31/2020, 5:08 AM

if you're getting a thread name change every context switch, you likely left the kotlin coroutines debug flag on

Matt

01/31/2020, 5:08 AM

ugh lol

how do i turn that off? mostly debugging but trying to see performance

Doing a report for school on my perofrmance vs other implementations

octylFractal

01/31/2020, 5:10 AM

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-d-e-b-u-g_-p-r-o-p-e-r-t-y_-n-a-m-e.html

Matt

01/31/2020, 5:10 AM

thanks so much!

View count: 10