I have a function to receive a batch of items from...

# coroutines

[Michael Strasser]

I have a function to receive a batch of items from a Channel , returning a list of items after timeoutMs milliseconds (the list might be empty). Here is the code:

public suspend fun <E> receiveBatch(
    channel: ReceiveChannel<E>,
    timeoutMs: Long,
): List<E> {
    val batch = mutableListOf<E>()
    withTimeoutOrNull(timeoutMs) {
        for (item in channel) {
            batch.add(item)
        }
    }
    return batch
}

In simple testing everything is fine, but under load sometimes an item is not added to a list and is lost. What I think happens is that withTimeoutOrNull times out between receiving an item from the channel and adding it to the list.

Does anyone have a solution for this issue, or a better design for what I am trying to do?

[ephemient]

something like this perhaps?

val start = TimeSource.Monotonic.markNow()
whileSelect {
    onTimeout(timeout - start.elapsedNow()) { false }
    channel.onReceiveCatching { result ->
        result.onFailure { throw it }
            .onClosed { return@onReceiveCatching false }
            .onSuccess { batch += it }
        true
    }
}

[Michael Strasser]

Thank you, @ephemient! This approach works well. Here is the current version of the function, which also constrains batch size:

@OptIn(ExperimentalCoroutinesApi::class)
public suspend fun <E> receiveBatch(
    channel: ReceiveChannel<E>,
    maxTimeMs: Long,
    maxSize: Int,
): List<E> {
    val items = mutableListOf<E>()
    whileSelect {
        onTimeout(maxTimeMs) { false }
        channel.onReceiveCatching { result ->
            result.onFailure { if (it != null) throw it }
                .onClosed { return@onReceiveCatching false }
                .onSuccess { items += it }
            items.size < maxSize
        }
    }
    return items
}

[ephemient]

the way you've written it there, the timeout is reset on every receive from the channel

[ephemient]

if that's what you want, you might as well write

while (true) {
    items += withTimeoutOrNull(maxTimeMs) {
        channel.receiveCatching().getOrElse { if (it != null) throw it else null }
    } ?: break
}

[Michael Strasser]

Thanks for the timeout correction.

[ephemient]

actually nevermind about my second code there, it has a race issue just like your original one did. but the timeout concern remains

[Michael Strasser]

Yep. I tried it and found that as well. Here is the current code with corrected timeouts:

@OptIn(ExperimentalCoroutinesApi::class, ExperimentalTime::class)
public suspend fun <E> receiveBatch(
    channel: ReceiveChannel<E>,
    maxTimeMs: Long,
    maxSize: Int,
): List<E> {
    val items = mutableListOf<E>()
    val start = TimeSource.Monotonic.markNow()
    whileSelect {
        onTimeout(maxTimeMs - start.elapsedNow().inWholeMilliseconds) { false }
        channel.onReceiveCatching { result ->
            result.onFailure { if (it != null) throw it }
                .onClosed { return@onReceiveCatching false }
                .onSuccess { items += it }
            items.size < maxSize
        }
    }
    return items
}