short circuiting reductions in flows

[jeffzoch]

I have a use case where I have a series of data points coming in over the network and I want to reduce them to a single value.

So i do something like:

fun compute() {
getValues()
.map{}
.onEach{}
.reduce { res: Set<V>, curr: Set<V> -> res.intersection(curr) //I dont actually use intersection but a slightly different method that handles empty sets.
}

What I'd like to do is short-circuit this reduction if the empty set is ever passed in. Some of the data points from getValues() may come in a minute after the first data point. If the first data point was the empty set {} then we ended up waiting for no reason.

Is there a simple mechanism to do this sort of short-circuiting in a general fashion without resorting to throwing exceptions in the reduction?

[jeffzoch]

My solution thus far is to do something like:

fun compute() {
  return try {
      getValues()
    .map{}
    .onEach{}
    .reduce { res: Set<V>, curr: Set<V> -> 
      if(res.isEmpty() || curr.isEmpty()) cancel()
      res.intersection(curr) 
  }
} catch (e: CancellationException) {
 emptySet()
}

I think this can be handled in a nicer way in Flows given that reduce is done incrementally and it would make sense to want to bail out under certain conditions.

[qwwdfsad]

There is no such shortcut out-of-the-box.

Depending on your specific problem, you can cancel the flow in onEach (to separate responsibility from reduce), filter empty sets out using filter or write your own operator.

I think this can be handled in a nicer way in Flows given that reduce is done incrementally and it would make sense to want to bail out under certain conditions.

Probably so, but to introduce such API, it would be nice to gather more use-cases and domain-specific problems

[gorttar]

My solution thus far is to do something like:

fun compute() {
  return try {
      getValues()
    .map{}
    .onEach{}
    .reduce { res: Set<V>, curr: Set<V> -> 
      if(res.isEmpty() || curr.isEmpty()) cancel()
      res.intersection(curr) 
  }
} catch (e: CancellationException) {
 emptySet()
}

I think this can be handled in a nicer way in Flows given that reduce is done incrementally and it would make sense to want to bail out under certain conditions.

You can actually write such a reduce yourself.

Implementation:

import kotlinx.coroutines.flow.Flow

sealed interface ShortCircuitContext

private data object ShortCircuitContextInstance : ShortCircuitContext

private data object BreakSignal : Throwable() {
    private fun readResolve(): Any = BreakSignal
}

private data object ContinueSignal : Throwable() {
    private fun readResolve(): Any = BreakSignal
}

@Suppress("UnusedReceiverParameter")
val ShortCircuitContext.finish: Nothing get() = throw BreakSignal

@Suppress("UnusedReceiverParameter")
val ShortCircuitContext.skip: Nothing get() = throw ContinueSignal

suspend inline fun <reified S, T : S> Flow<T>.scReduce(
    crossinline operation: suspend ShortCircuitContext.(acc: S, value: T) -> S
): S = ShortCircuitContextInstance.run {
    val nil = object {}
    var acc: Any? = nil
    try {
        collect { t ->
            when (val prev = acc) {
                nil -> acc = t
                is S -> try {
                    acc = operation(prev, t)
                } catch (_: ContinueSignal) {
                }
            }
        }
    } catch (_: BreakSignal) {
    }
    if (acc == nil) throw NoSuchElementException("Empty flow can't be reduced")
    acc as S
}

Usage:

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.runBlocking

fun main(): Unit = runBlocking {
    println(
        (1..10).asFlow().scReduce { acc, x ->
            acc + x
        }
    )
    println(
        (1..10).asFlow().scReduce { acc, x ->
            if (x % 2 == 0) skip
            acc + x
        }
    )
    println(
        (1..10).asFlow().onEach(::println).scReduce { acc, x ->
            if (x > 5) finish
            acc + x
        }
    )
    (1..10).asFlow().map(::setOf).onEach(::println).scReduce { acc, x ->
        if (acc.isEmpty()) finish
        acc intersect x
    }
}