How to use a Kotlin coroutine to call a blocking function?

user_923846 0
Viewed 2344

I would like to use Kotlin coroutines to call blocking functions.

For example, I would like to have a suspend function that would "block" until the user responds to a modal dialog, or until a network call returns, when the dialog and network libraries don't provide suspend functions.

I implemented it using a Mutex.

For example, for the blocking network call scenario, I did something like this:

class Connection {
    private val mutex = Mutex()
    
    public suspend fun receive(): ByteArray {
        mutex.lock()

        val buf = ByteArray(42)
        
        thread {
            sock.getInputStream().read(buf, 0, 42) // blocking
            mutex.unlock()
        }

        mutex.lock()

        return buf
    }
}

It works.

Please ignore networking-related issues in the above code (like error handling, checking the number of bytes actually read, etc.).

Please also ignore performance aspects of using a thread (the dedicated thread above is just for the sake of example).

From the perspective of using a Mutex to "convert" a blocking flow to a coroutines flow:

  1. Is there a standard or a better way to do this?
  2. Aside for networking or performance - do you see any downsides to my approach?
kotlin-coroutines kotlin
2 Answers

The standard way is to use the IO scheduler, which is designed to have a large pool of threads to handle blocking operations (as the name implies, IO).

withContext(Dispatchers.IO) {
    sock.getInputStream().read(buf, 0, 42)
}

This will suspend the coroutine until the blocking operation is finished.

The downsides of your approach is that the order of execution is not entirely clear and is prone to deadlocking. For example if your blocking call throws an exception the mutex never ends up unlocking and the coroutine will be stuck. In addition it creates new threads for each blocking operations, while the IO dispatcher is designed to reuse a pool of threads to avoid this cost.

user_14520940

You can just do:

suspend fun receive(): ByteArray {
    return withContext(Dispatchers.IO) {
        val buf = ByteArray(42)
        sock.getInputStream().read(buf, 0, 42) // blocking
        buf
    }
}

call it from your Activity as:

lifecycleScope.launch{
    withContext(Dispatchers.Main) {
        //showLoading
        val result = receive()
        //hideloading
    }
}

the receive function will run in the IO dispatcher which is designed for offloading blocking IO tasks to a shared pool of threads see more about dispatchers here.

user_40942600
Related