@bj0 i care for the order. I want to abort with an exception, as soon as any of the jobs throws. If I'd do

deferredExports.foreach { it.await() }

and the first job takes the longest, I'd only get the exception of another job, after the first terminates.

So I recently had a horrifying revelation and I'm wondering if we could get a Doug Lea-like person to either confirm my suspicions or tell me I'm crazy. I wanted a

SequentialPoolExecutor

, so I wrote one, and tested it. The idea is that every

submit

or

execute

call is sequential-ized such that no submitted job will run before another submitted job completes. This is not necessarily single threaded as one very simple optimization we can employ is to back this SequentialExecutor with a pool, where the first-available thread is selected to run the job. My goal for such a component would be to allow me to write code like this:

class SomeStatefulComponent {
  private val data = MutableDataStructureThatIsntThreadSafe()

  suspend fun doMutation(args: Args): Double = run(SequentialExecutor) { 
    data += transform(args)
    return data.moreComputation()
  }
}

and not worry about using nasty

@Volatile

or

Unsafe

or

AtomicReference

or more generally CAS/locking strategies. Instead such an executor would elegantly serialize everything for me. But the problem, thinking back to concurrency in practice, is that which

@Volatile

was originally designed to solve: if some mutable state on

data

is put into a thread-local cache, then even squentially run jobds might get their correctness ruined by such a cache. Effectively this boils down to an apocalyptic assumption: is it really the case that fields not marked for explicit thread sharing cannot ever be shared between threads? Does somebody have a clever way to make this non-functional problem into a functional one via use of a fuzz-testing or other concurrency-testing strategy?

Unfortunately, especially on x86, hammering on stuff hoping for rare failures tends to only produce results that imply that x86 has a very strong memory model. 😉 If you can't convince yourself with sufficient perusal of JCIP that there's a happens-before chain, it's probably wrong.

Should I use

Channel(capacity = UNLIMITED)

+

offer(...)

or a rendez-vous

Channel()

+

launch { send(...) }

?

The default dispatcher for runBlocking coroutine, in particular, is confined to the invoker thread

Is this still true? Isn't

runBlocking

now using

DefaultDispatcher

, which is equal to

CommonPool

on JVM/Android? Quote from here, second paragraph, third sentence: https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#unconfined-vs-confined-dispatcher

I need to ramp up quickly on idiomatic Kotlin patterns using coroutines with both blocking and non-blocking I/O. Does anyone have any substantial experience in this area? In particular, I will like to know how suspend fuctions with blocking network I/O performs against non-suspend functions using NIO.

Coroutines make using nio simpler. For blocking io they provide little benefit as they will need one thread per io channel anyway

👋 I need a bit help, I'm trying go from an async complex op in rx to a simpler implementation using coroutines, so first step is to get my observable returning a value, to do so I'm using

suspendCoroutine

so:

suspend fun myFunc = suspendCancellableCoroutine{continuation ->
  myObservable.subscribe(onsuccess={continuation.resume(it)}, onError={continuation.resume(error(it)})
   continuation.invokeOnCompletion(onCancelling = true) {disposable.dispose()}}

Thing is that when I cancel the coroutine that calls the function, it doesn't dispose the observable, is there anyway to make the function to cancel immediately? as the doc says even if the coroutine is cancelled it doesn't mean that the code will be cancelled but instead we should check regularly inside that code whether is't cancelled or not, but, is there any alternative?

@gildor I've just come up with your lib and wonder, how you'd cancel a request when the coroutine get's cancelled? (sorry for the ping but the most similar scenario that comes to mi mind is that you launch a network call within a coroutine and cancel it, but it wouldn't cancel the request, would it?)

Urgent help needed! Is this adaptation of the coroutine tutorial correct, or did I miss something?

import org.springframework.util.concurrent.ListenableFuture
import org.springframework.util.concurrent.ListenableFutureCallback

public suspend fun <T> ListenableFuture<T>.await(): T = suspendCancellableCoroutine { cont: CancellableContinuation<T> ->
    val callback = ContinuationCallback(cont)
    this.addCallback(callback)
    cont.invokeOnCompletion {
        callback.cont = null // clear the reference to continuation from the future's callback
    }
}

private class ContinuationCallback<T>(@Volatile @JvmField var cont: CancellableContinuation<T>?) : ListenableFutureCallback<T> {
    @Suppress("UNCHECKED_CAST")
    override fun onSuccess(result: T?) { cont?.resume(result as T) }
    override fun onFailure(t: Throwable) { cont?.resumeWithException(t) }
}

For the sake of completeness, I should mention that cancellation is not only possible at suspension points, you can also check

isActive

yourself to stop execution -- all of Kotlin's suspending functions do that so they also enable cancellation

Is there an equivalent to

select

that doesn't abort if one of the coroutines throws an exception?

Is there a notion of priority with coroutines? I'd like to do some job in background but I'd like it to be executed ASAP, before other queued & less important jobs

is there any way to achieve something similiar with the

throttle

operator using coroutines?

I have a basic question about how

isActive

works:

fun main(args: Array<String>) = runBlocking {
  val job = Job()

  val pool = newFixedThreadPoolContext(2, "Cool")
  repeat(2) {
    launch(pool, parent = job) {
      startLoop()
    }
  }

  delay(1000)

  println("Canceling")
  job.cancel()

  delay(1000)

  println("End")

  Unit
}

suspend fun startLoop() {
  while (isActive) {
    println("I'm active: ${isActive} in ${Thread.currentThread().name}, my job is cancelled: ${coroutineContext[Job]?.isCancelled}")
    Thread.sleep(500)
  }
}

calling

job.cancel()

will not terminate the loops. The complete output is:

I'm active: true in Cool-1, my job is cancelled: false
I'm active: true in Cool-2, my job is cancelled: false
I'm active: true in Cool-1, my job is cancelled: false
I'm active: true in Cool-2, my job is cancelled: false
I'm active: true in Cool-2, my job is cancelled: false
I'm active: true in Cool-1, my job is cancelled: false
Canceling
I'm active: true in Cool-1, my job is cancelled: true
I'm active: true in Cool-2, my job is cancelled: true
I'm active: true in Cool-1, my job is cancelled: true
I'm active: true in Cool-2, my job is cancelled: true
End

In my real life situation, instead of

Thread.sleep

, the coroutine is waiting to read from a blocking queue (poll with timeout). If I use

delay(500)

instead of sleeping, the coroutines are cancelled properly. Does a Coroutine have to suspend in order for isActive to change?

How to cancel a coroutine inside the coroutine? Something like in normal coding, I can do

return

, then the statement after this

return

will not be executed. But what about inside a coroutine? You can’t

return

in a coroutine. Seems the answer is

coroutineContext().cancel()

, But with

Anko

, it said

coroutineContext of type CoroutineContext can't be invoked as a function

you can throw a

CancellationException

I don't think that function can be tail call optimized even without the suspend modifier. The recursive call is within the let-block.

are there any plans to include coroutines in the stdlib or will they always require an additional library?

even if we for-go the "use a channel not a blockable queue", which is rather abstract, you could use standard kotlin

@ReplaceWith

for something like

launch {
  //replace 
  staticallyKnownAsBlockingQueueInstance.take()
  //with
  run(BLOCKABLE_THREAD_POOL) { staticallyKnownAsBlockingQueueInstance.take() }
}

inducing a compiler error at

BLOCKABLE_THREAD_POOL

... unlness kotlinx.coroutines can be updated to include such a dispatcher embedded in itself.

Let me know if there is better way ⤴️

Can we set thread priority when using or initializing a dispatcher? Or maybe Coroutines system is designed to handle this? I see (on Android) that all threads from

CommonPool

or custom from

newSingleThreadContext

inherit the normail priority level. Which makes them equal to the main thread. Is there another mechanism to make the UI thread not affected? I'd like to make sure to prioritize it.

you should use

runBlocking

fun ReceiveChannel. toDeferred() = async { single() }

, @spierce7 maybe, but probably not such good practice in coroutines to do such a thing though...

Is there any way to convert guava ListenableFuture into Kotlin Deferred? I see that there is an extension method await on ListenableFuture in kotlin-coroutines-guava but it is not what I need

Hi, can't the

cancel()

methods in

Channel

and

Job

be optimized to not allocate a new

CancellationException

when the channel is already closed or the job is already cancelled or completed?

Hi! If I have a

newSingleThreadedCoroutrineContext

, and a bunch of coroutine launched with it, a) will the coroutines execute in the order of their creation? b) will the execution prefer the first coroutine more than the last after some suspends and resumes?

like: I create a channel, and once it's closed I need to invoke some callback to clean up