I need bit of guidance, I'm trying to use coroutines to offload DB writes to another thread. I have ...

# coroutines

[Marko Mitic]

I need bit of guidance, I'm trying to use coroutines to offload DB writes to another thread. I have this DB table that is read on app start and is all writes after that thanks to caching. I wanted to avoid blocking the calls for clients so I'm offloading writes using a channel (writes should be executed in-order so I used a channel). Is this ok? (GlobalScope will be replaced later)

internal class BlockedAppsStoreDatabaseImpl(private val blockedAppDao: BlockedAppDao) : BlockedAppsStore {

    private val channel = Channel<DbOp>(Channel.UNLIMITED)

    init {
        GlobalScope.launch {
            select {
                channel.onReceive { op ->
                    when (op) {
                        is Get -> op.cb.complete(blockedAppDao.getAll().toSet())
                        is Put -> blockedAppDao.putOrReplace(BlockedAppEntity(op.uid))
                        is Remove -> blockedAppDao.remove(op.uid)
                        is PutSet -> blockedAppDao.putOrReplace(op.uids)
                        is Fill -> blockedAppDao.fill(op.uids)
                        is RemoveAll -> blockedAppDao.removeAll()
                    }
                }
            }
        }
    }

    override fun getAll(): Set<Uid> {
        val deferred = CompletableDeferred<Set<Uid>>()
        channel.sendBlocking(Get(deferred))
        return runBlocking { deferred.await() }
    }

    override fun put(uid: Uid) {
        channel.sendBlocking(Put(uid))
    }
    ...

[Dominaezzz]

This still blocks the calling thread anyway. You need to make those functions suspend.

[Marko Mitic]

Get is intentionally blocking, but put shouldn't block, right?

[Marko Mitic]

(since channel isn't limited in size)

[Dominaezzz]

That depends on what you want. Do you need the operations to conclude before the function returns?

[Dominaezzz]

Also why do you need the channel?

[Marko Mitic]

I don't need DB write to complete before returning, I'm trying to skip that wait

[Marko Mitic]

I'm using channel to keep write operations in order

[Dominaezzz]

Then you need to do a launch in the set function.

[Dominaezzz]

The use suspending send.

[Marko Mitic]

sendBlocking doesn't block unless channel is full

[Marko Mitic]

it should work for me since channel is unlimited, right?

[Marko Mitic]

I've confirmed this works (after putting select in a while loop), I'm checking whether I'm missing some big-picture issue

[mbonnin]

You could use offer() directly if you know your channel will never block

[Marko Mitic]

nice one, thanks

[bdawg.io]

You could simplify your code to avoid using a select and just use channel.consumeEach { op -> directly as well.

[bdawg.io]

One other thing you could check out is using an actor directly for your channel as well.

private val channel = GlobalScope.actor<DbOp>(capacity = Channel.UNLIMITED) {
    for (op in channel) { ... }
}

That will ensure that your channel is properly cleaned up when your scope gets cancelled. (Which would make moving away from GlobalScope easier in future refactors)

[Marko Mitic]

Thanks, that's all usefull

[Marko Mitic]

both actor and consumeEach are experimental, I can't really use them in my library

[Marko Mitic]

although they look real nice

[bdawg.io]

`actor`s are "obsolete" because they are being refactored to support complex cases with provided classes, but I doubt they will force us to implement a class just to use an actor with the newer complex stuff.

consumeEach is tentatively graduating in 1.4.0. even if it doesn't, it's implementation is fairly simple

suspend fun <T> ReceiveChannel<T>.consumeEach(block: (T) -> Unit) {
    try {
        for (item in this) block(item)
    } catch (e: Throwable) {
        cancel(e as? CancellationException ?: CancellationException("Channel cancelled, consumer failed", e))
    } finally {
        cancel()
    }
}