# How can I use co-routines to single-thread asynchronous responses

**URL:** <https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045>\
**Category:** Support\
**Created:** [September 28, 2021, 9:21am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045 "2021-09-28T09:21:44Z")\
**Posts on this page:** 17\
**Page:** 1

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 9:21am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/1 "2021-09-28T09:21:44Z")

</div>

In my code I wish to ‘single-thread’ actions that are invoked asynchronously for external sources. I do this by maintaining a queue of lambdas with a dedicated thread. I use the thread to remove lambdas from the list and execute them. When the list is empty the thread waits on a lock (with a timeout to poll the queue in case a lambda gets missed). When lambdas are added I signal the lock to wake up the dedicated thread.

This ensures that all changes (to a complex model) happen on a single thread and so no data-race conditions occur.

This works, but seems a bit cumbersome. Is there a simpler, more elegant solution using co-routines?

Would a buffered channel replace the queue? Would a consumer for that queue execute the lambdas sequentially?  
Also, this code is in a library (jar). What is the best way to launch the co-routine scope?

[EDIT] I do not require that all lambdas are executed on the same thread - and in fact that requirement requires a dedicated thread - just that only one lambda is executed at a time. So 'single-threaded may be misleading. I don’t care as long as each thread sees the state that the previous thread created.[/EDIT]

Thanks for any help

```kotlin
public typealias SimpleBlock = () -> Unit
/**
 * Executes the given blocks in order on a single daemon thread.
 */
public class DispatchQueue(name: String) {
    public fun add(block: SimpleBlock) {
        synchronized(lock) {
            eventQueue.addLast(block)
            lock.notify()
        }
    }

    public fun stop() {
        keepProcessing = false
        synchronized(lock) {
            lock.notify()
        }
    }

    private var keepProcessing = true
    private val lock = Object()
    private val eventQueue = ArrayDeque<SimpleBlock>()
    public val thread: Thread =
        thread(name = name) {
            while (keepProcessing) {
                var block: SimpleBlock?
                do {
                    synchronized(lock) {
                        block = eventQueue.removeFirstOrNull()
                    }
                    // need to invoke the block outside the lock!
                    block?.invoke()
                } while (null != block)

                synchronized(lock) {
                    try { lock.wait(20) } catch (e: TimeoutException) { }
                }
            }
        }
}

```

---

<div class="post-metadata">

**Author:** ![darksnake](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/darksnake/32/2479_2.png) [@darksnake](https://discuss.kotlinlang.org/u/darksnake)\
**Post date:** [September 28, 2021, 9:43am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/2 "2021-09-28T09:43:34Z")

</div>

You can run coroutines on a single-threaded dispatcher. It seemed like it will solve your problem.

```kotlin
val dispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
launch(dispatcher){
  //whatever
}

```

This will ensure that all coroutines will work on the same thread avoiding concurrent access, but aynchronously.

---

<div class="post-metadata">

**Author:** ![al3c](https://avatars.discourse-cdn.com/v4/letter/a/e47774/32.png) [@al3c](https://discuss.kotlinlang.org/u/al3c)\
**Post date:** [September 28, 2021, 9:51am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/3 "2021-09-28T09:51:11Z")

</div>

> [@phill](#):
>
> When the list is empty the thread waits on a lock (with a timeout to poll the queue in case a lambda gets missed).

Even with the threading approach if you use a Java `ArrayBlockingQueue` you won’t need a lock at all, just take elements from the queue in order. `queue.take` will block for you.

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 9:55am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/5 "2021-09-28T09:55:05Z")

</div>

Thanks - I missed the blocking Queue in my search. That will help if I do not use co-routines.

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [September 28, 2021, 10:02am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/6 "2021-09-28T10:02:33Z")

</div>

Or you can use single-thread executor in Java:

```kotlin
val executor = Executors.newSingleThreadExecutor()
executor.submit {
    ...
}

```

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 10:10am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/7 "2021-09-28T10:10:59Z")

</div>

Does the single thread executor reserve a thread from the pool? That would be as inefficient as declaring my own dedicated thread since the dispatch queue servicing runs for a long time.  
Actually, I guess my initial requirement might be stated wrongly. I do not care what thread my tasks are run on, as long as only one is run at a time… I will update my question.

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [September 28, 2021, 10:49am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/8 "2021-09-28T10:49:27Z")

</div>

So far all proposed solutions require to create a dedicated thread to handle actions. I thought your main concern is to simplify your code, not to get rid of this thread 🙂

But if you want to remove it then I don’t see a straightforward solution without using coroutines. It is technically possible, just not that easy. We can’t wait for new tasks by blocking, because then we require some thread to do this blocking. We would need to wait asynchronously and be careful not to run multiple tasks in parallel.

With coroutines this is pretty easy, using exactly the solution you mentioned in the first post. Just create a buffered channel and start a coroutine, e.g. using `Dispatchers.Default` that reads items from this channel in a loop. Yes, it will work sequentially as long as you don’t start asynchronous tasks (e.g. with `launch()`). It will reuse threads with other parts of your application, but won’t occupy any while waiting for new tasks.

Regarding the scope, it really depends. Probably the most universal way is to design this as some kind of a service, with clearly defined lifecycle. Then, you create `CoroutineScope()` and store it in a property of your service and in your `cancel()`/`close()` function you invoke `cancel()` on this scope. This will stop processing almost immediately. If you need to shutdown gracefully and process already queued items, you would need to do it a little differently.

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 11:00am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/9 "2021-09-28T11:00:30Z")

</div>

OK, thanks for that @broot . My apologies to all if my question was poorly phrased/thought out.

I guess that my main motivation was to explore whether coroutines would allow more elegant code. For me that means both ‘more efficient (get rid of a dedicated thread)’ and ‘syntactically simpler’.

A buffered channel and a default dispatcher in a simple loop sounds ideal. I will play with that idea and see where I get.

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 11:49am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/10 "2021-09-28T11:49:26Z")

</div>

Is this a correct solution, please? If so it is a lot simpler than my original solution.

Requirements:

- Execute lambdas in the order they were added
- Execute lambdas one at a time
- Execute each lambda with the state that the previous lambda created (not visible here)

```kotlin
package coroutines

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

typealias SimpleBlock = () -> Unit
/**
 * Executes the given blocks in order on a single daemon thread.
 */
class DispatchQueue(name: String) {
    private val channel = Channel<SimpleBlock>(Channel.BUFFERED)

    fun add(block: SimpleBlock) = runBlocking { channel.send(block) }

    fun stop() = sc.cancel()

    private val sc = GlobalScope.launch {
        while (true) { channel.receive()() }
    }
}

fun main() {
    val dq = DispatchQueue("dq")
    dq.add { println("first") }
    dq.add { println("second") }
    dq.add { println("third") }

    Thread.sleep(2000)

    dq.stop()
}

```

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [September 28, 2021, 12:25pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/11 "2021-09-28T12:25:12Z")

</div>

Yes, this implementation seems fine and it should meet all your requirements. There is a race condition on `add()` function: if multiple threads invoke this function concurrently, you don’t have guarantees which task will be added first. It may violate your first requirement, but I guess this shouldn’t be a problem. You can always synchronize this function if you need.

Note that `Channel.BUFFERED` does not create a channel with unlimited capacity. It has a fixed size and after filling it up, your `add()` will block, waiting for free space. I’m not sure if this is what you expect. If not then use `Channel.UNLIMITED` instead and then I think it would be better to implement `add()` as:

```kotlin
channel.trySend(block).getOrThrow()

```

`runBlocking()` always has some performance hit and we don’t really need to suspend if the channel is unlimited. Maybe I miss something here.

Also, someone will probably tell you that `GlobalScope` is evil, but I believe you used it correctly. Your solution is functionally the same as creating the scope with `CoroutineScope()`. You can simplify your loop to `for (item in channel)` or `channel.consumeEach {}`, but this is just a small improvement.

Also, keep in mind that coroutines are always a little cumbersome when you need to use them from outside of coroutines context and especially for some little feature. This is due to this scope and structured concurrency things. For this reason it may not be always a good idea to migrate to coroutines just to simplify the code. However, if you go fully-coroutines, many things become simpler. For example, your example could be reimplemented as:

```kotlin
class DispatchQueue {
    val channel = ...
    fun addTask() { ... }
    suspend fun run() {
        channel.consumeEach { ... }
    }
}

class AnotherService {
    suspend fun run() = coroutineScope {
        launch { DispatchQueue().run() }
        
        ...
    }
}

```

As coroutines support cancelling automatically, we don’t need to bother about creating the scope and cancelling manually. If we cancel coroutine that started `AnotherService`, it will automatically cancel all services, including the loop in `DispatchQueue`.

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 12:38pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/12 "2021-09-28T12:38:20Z")

</div>

Thank you for taking the time to pen that full reply. Those look like excellent points and I will digest them slowly.

All of them are valid for me in this context. I like the syntactic improvements, and synchronising the add will help to remove a source of entropy. As for how to embed this in the bigger picture - we that will take the most thought. I already use dependency injection, and the model that makes use of the dispatchQueue can be injected where it is needed, so it may be simpler than feared (Inversion of control becomes more than a buzzword). The GUI makes use of javafx and will likely need animations so co-routines may have some use there, and the whole system will need to deal with asynchronous comms, so again co-routines (once I understand them a bit more) may have a part to play there.

So again, many thanks. This has been a ‘focused lesson’ in co-routines at least as much as it has been solution finding.

Phill

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 28, 2021, 12:58pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/13 "2021-09-28T12:58:21Z")

</div>

And for the record here is my updated version.

```kotlin
package coroutines

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach

// with help and guidance from here:
// https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/11?u=phill
typealias SimpleBlock = () -> Unit
/**
 * Executes the given blocks in order on a single daemon thread.
 */
class DispatchQueue(name: String) {
    private val channel = Channel<SimpleBlock>(Channel.UNLIMITED)

    @Synchronized fun add(block: SimpleBlock) = channel.trySend(block).getOrThrow()

    fun stop() = sc.cancel()

    private val sc = CoroutineScope(Dispatchers.Default).launch { channel.consumeEach { it() } }
}

fun main() {
    val dq = DispatchQueue("dq")
    dq.add { println("first") }
    dq.add { println("second") }
    dq.add { println("third") }

    Thread.sleep(2000)

    dq.stop()
}

```

---

<div class="post-metadata">

**Author:** ![mtimmerm](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/mtimmerm/32/7160_2.png) [@mtimmerm](https://discuss.kotlinlang.org/u/mtimmerm)\
**Post date:** [September 28, 2021, 8:44pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/14 "2021-09-28T20:44:24Z")

</div>

It sounds like you have this requirement: [java - How can I create single-thread coroutine context under Common pool in Kotlin? - Stack Overflow](https://stackoverflow.com/questions/50355071/how-can-i-create-single-thread-coroutine-context-under-common-pool-in-kotlin)

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [September 28, 2021, 10:12pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/15 "2021-09-28T22:12:45Z")

</div>

Ahh, I just realized something very important in this discussion. It is crucial to specify if our goal is to limit _concurrency_ or _parallelism_.

It was a little simpler in Java world, because `Executors.newSingleThreadExecutor()` actually limited both. But this is not the case with coroutines. Single-threaded dispatcher (mentioned by @darksnake and @mtimmerm) limits parallelism, but not concurrency. Queue with single consumer limits concurrency, but not parallelism. From your description I believe you actually need the latter one.

I explain the difference with this code:

```kotlin
fun main() {
    Executors.newSingleThreadExecutor().asCoroutineDispatcher().use { dispatcher ->
        var data = 0

        runBlocking(dispatcher) {
            repeat(1000) {
                launch {
                    data = transformData(data)
                }
            }
        }

        println(data)
    }
}

suspend fun transformData(input: Int): Int {
    delay(500) // processing
    return input + 1
}

```

We increment an integer 1000 times, using a single-thread dispatcher. What is the result of this code? Value: `1`. The problem here is that even using single-thread dispatcher, we still execute all 1000 tasks concurrently. This is coroutines magic. We will get exactly the same result using `SerialContextDispatcher` from the linked StackOverflow question.

Using a channel and reading from it with a single coroutine does a really different thing. It makes sure that we only start a new task after the previous one is fully completed. I didn’t verify this, but it should return `1000` instead of `1` and take `500s` instead of `500ms`.

---

<div class="post-metadata">

**Author:** ![darksnake](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/darksnake/32/2479_2.png) [@darksnake](https://discuss.kotlinlang.org/u/darksnake)\
**Post date:** [September 29, 2021, 6:12am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/16 "2021-09-29T06:12:17Z")

</div>

Actually, a single-thread dispatcher prohibits concurrent (simultaneous) access to data. It won’t guarantee the order of access is linear though.

---

<div class="post-metadata">

**Author:** ![phill](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/phill/32/8551_2.png) [@phill](https://discuss.kotlinlang.org/u/phill)\
**Post date:** [September 29, 2021, 7:31am UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/17 "2021-09-29T07:31:01Z")

</div>

I think that I am trying to limit both concurrency and parallelism. I want to execute a sequence of tasks in the order that they were presented (for some definition of the word ‘presented’), but execute them one at a time, in such a manner that changes to shared state made by one task are honoured by the subsequent tasks. Calling this ‘single threading’ was a mistake - I actually do not care which thread the code is executed on.  
From my understanding the code I wound up will do this - the channel enforcing an ordering and the coroutine scope ensuring that only one task is handled at a time. I am unsure whether the shared data is treated correctly but I am assuming so.  
I also believe that the code will share its thread with a common pool so thread identity is not maintained. That increases efficiency without side effects (I believe).  
What I do like is that (given an understanding of the bits of coroutines I have used) the code is terse and the intent shines through the syntactic noise. This clarity is a part of Kotlin that I find very appealing.

[Edit] In case you had not spotted it, my solution is an attempt to replicate the dispatch queue concept I first encountered in swift/iOS (aka ‘Grand Central Dispatch’). It provides a simplification for handling complex state models with multiple asynchronous input. It might be helpful to have this capability as part of the core coroutines code.[/Edit]

---

<div class="post-metadata">

**Author:** ![westnordost](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/westnordost/32/6008_2.png) [@westnordost](https://discuss.kotlinlang.org/u/westnordost)\
**Post date:** [October 25, 2021, 10:46pm UTC](https://discuss.kotlinlang.org/t/how-can-i-use-co-routines-to-single-thread-asynchronous-responses/23045/18 "2021-10-25T22:46:36Z")

</div>

A new API for such use case with coroutines is also planned (for 1.6?):

> <https://github.com/Kotlin/kotlinx.coroutines/issues/2919>
>
> See the original idea behind slices here: https://github.com/Kotlin/kotlinx.coro…utines/issues/261
> Since then, a lot of API was introduced and refined, so the proposed API no longer fits \`kotlinx.coroutines\`.
> 
> \### Use-cases
> 
> \`newFixedThreadPoolContext\` and \`newSingleThreadContext\`, as well as regular Java executors, are often used as the only way to control concurrency. Various factors contribute to that -- limited external resources (e.g. pool of database connections), functional restrictions (e.g. CPU consumption or rate-limiting), or need to have an isolated serializable execution flow (an extreme example would be IDEA code-base that contains \*62\* single-threaded dispatchers).
> 
> As a result, an average application creates unnecessary many threads, most of which are idle, consuming the memory, CPU, and device battery.
> 
> \### Design overview
> 
> To address such use-cases, we aim to provide API to take a \_view\_ of the dispatcher. The dispatcher's view does not create any additional threads and does not allocate any resources. Instead, it piggybacks the original dispatcher but limits the parallelism of submitted tasks and coroutines. Views themselves are independent and it is possible to have views that have more \_declared\_ parallelism in total than the original dispatchers. It saves users from meticulous counting of resources, especially when views are used along with global \`Dispatchers.Default\` and \`Dispatchers.IO\`.
> 
> With that API, it is possible to have fine-grained control over the parallelism and leverage already existing dispatchers and executors, saving resources and the number of threads.
> 
> The API changes are minimal, i.e. instead of
> \`\`\`
> private val myDatabaseDispatcher = newFixedThreadPool(10, "DB") // At most 10 connections to the database
> private val myCpuHeavyImageProcessingDispatcher = newFixedThreadPool(2, "Image rescaling") // Prevent more than 200% of CPU consumption on images
> \`\`\`
> it is possible to write this instead:
> \`\`\`
> private val myDatabaseDispatcher = Dispatchers.IO.limitedParallelism(10) // At most 10 connections to the database
> private val myCpuHeavyImageProcessingDispatcher = Dispatchers.Default.limitedParallelism(2) // Prevent more than 200% of CPU consumption on images
> \`\`\`
> 
> \### API changes
> 
> Conceptually, \_any\_ coroutine dispatcher can limit its own parallelism. If we are allowing the sum of views to be greater than the original dispatcher, nothing prevents users to take a view of a single-threaded dispatcher (e.g. \`Dispatchers.Main\`) as long as its contract allows it to short-circuit the implementation to \`return this\`.
> 
> As a result, a single public signature will be added to \`CoroutineDispatcher\` class
> 
> \`\`\`
> public abstract class CoroutineDispatcher {
> 
> public open fun limitedParallelism(parallelism: Int): CoroutineDispatcher
>    
> ...the rest of the methods...
> }
> 
> \`\`\`
> 
> with the default implementation provided.
> 
> \### Rejected alternative namings
> 
> \`\`\`
> Dispatchers.Default.slice(parallelism = 2)
> Dispatchers.Default.view(2)
> Dispatchers.Default.limited(parallelism = 2)
> Dispatchers.Default.subset(2)
> Dispatchers.Default.dedicated(2)
> \`\`\`
> 
> While being short and visually attractive, it's easy to misinterpret their semantics.
> \- For \`slice\`, \`subset\`, and \`view\` it can be reasonable to assume that the sum of all the slices/subsets/views cannot be greater than the original
> \- For \`dedicated\` it can be reasonable to assume that \_dedicated threads\_ are used for such dispatcher
> \- The biggest concern for \`limited\` is its name. We cannot enforce the usage of named parameters on the language level, and seeing \`myDispatcher.limited(2)\` or \`myDispatcher.limited(myLimit)\` in the wild may be quite misleading.

A coroutine dispatcher you can use to enforce that only one thread can enter any `withContext(myLimitedParallelismDispatcher) { ... }` block.