Experiments for adding Flow abstraction

[tassiluca]

Hi!

I'm opening this discussion (hopefully in the right place) to share some experiments I did to introduce cold Flows in the codebase for modeling a stream of asynchronously computed values.
I don't know if they can be useful, but I think it's worth sharing them.

Flows can be implemented using a Task and a Channel:

• At Flow creation, the client provides a block of code that emits values, which is wrapped inside a Task that is started only when the collect method is called;
• The emitted values are sent on a Channel that is created when the collect method is called.

/** A cold stream of asynchronously computed values. */
trait Flow[+T]:

  /** Start the flowing of data that can be collected reacting through the given [[collector]] function. */
  def collect(collector: Try[T] => Unit)(using Async, AsyncOperations): Unit

/** An interface modeling an object capable of [[emit]]ting [[Flow]]able values. */
trait FlowCollector[-T]:

  /** Emit a value to the flow. */
  def emit(value: T)(using Async): Unit

object Flow:

  /** Create a new asynchronous cold [[Flow]] from the given [[body]].
    * Since it is cold, it starts emitting values only when the [[Flow.collect]] method is called.
    * To emit a value use the [[FlowCollector]] given instance with `it.emit(value)`.
    */
  def apply[T](body: (it: FlowCollector[T]) ?=> Unit): Flow[T] =
    val flow = FlowImpl[T]()
    flow.task = Task:
      val channel = flow.channel
      flow.sync.release()
      given FlowCollector[T] with
        override def emit(value: T)(using Async): Unit = channel.send(Success(value))
      try body
      catch case e: Exception => channel.send(Failure(e))
    flow

  private class FlowImpl[T] extends Flow[T]:
    private[Flow] var task: Task[Unit] = uninitialized
    private[Flow] var channel: TerminableChannel[Try[T]] = uninitialized
    private[Flow] val sync = Semaphore(0)

    override def collect(collector: Try[T] => Unit)(using Async, AsyncOperations): Unit = Async.group:
      val myChannel = TerminableChannel.ofUnbounded[Try[T]]
      synchronized:
        channel = myChannel
        task.start().onComplete(() => myChannel.terminate())
        // Ensure to leave the synchronized block after the task has been initialized
        // with the correct channel instance.
        sync.acquire()
      myChannel.foreach(collector)

Some notes:

• TerminableChannel is a specialization of Channel allowing to terminate it, preventing further values from being sent, while still allowing to read the values that have been sent before the termination (currently, read returns Left(Closed) once the channel is closed)
  • the foreach method can be implemented to react to each sent value suspending the execution until all items have been consumed
• if body throws an exception, it is sent on the channel wrapped inside a Failure
• since the collect method could be called concurrently by multiple threads a synchronized block and a Semaphore are used to ensure the task can get the correct newly generated channel instance before someone else can replace it with another instance
• the emit function signature could be turned to be non-suspendable using sendImmediately on something like a TerminableChannel backing to an UnboundedChannel

To create a Flow:

def f()(using Async): Flow[Int] = Flow:
  sleep(1_000) // something meaningful
  it.emit(1)   // emit a value
  sleep(1_000)
  it.emit(2)

Client-side:

Async.blocking:
  val flow = f()
  println("Not collecting yet")
  sleep(5_000)
  println("Collecting...")
  flow.collect(println)
  println("End!")

The output will be (the square brackets are my annotations containing the time elapsed since the previous message):

Not collecting yet
Collecting...         [after 5 seconds]
Success(1)            [after 1 second]
Success(2)            [after 1 second]
End!

In case of failures, the flow stops:

def failingFlow(using Async): Flow[Int] = Flow:
  sleep(1_000)
  throw new Exception("Boom!")
  sleep(1_000)
  it.emit(10) // this will never be emitted

Async.blocking:
  failingFlow.collect(println)
  println("End!")

The output:

Failure(java.lang.Exception: Boom!)   [after 1 second]
End!

On top of Flow it is possible to implement functions like map and flatMap:

object FlowOps:

  extension [T](flow: Flow[T])

    /** @return a new [[Flow]] whose values has been transformed according to [[f]]. */
    def map[R](f: T => R): Flow[R] = new Flow[R]:
      override def collect(collector: Try[R] => Unit)(using Async, AsyncOperations): Unit =
        catchFailure(collector):
          flow.collect(item => collector(Success(f(item.get))))

    /** @return a new [[Flow]] whose values are emitted by flattening the flows generated
      *         by the given function [[f]] applied to each emitted value of this.
      */
    def flatMap[R](f: T => Flow[R]): Flow[R] = new Flow[R]:
      override def collect(collector: Try[R] => Unit)(using Async, AsyncOperations): Unit =
        catchFailure(collector):
          flow.collect(item => f(item.get).collect(x => collector(Success(x.get))))

    private inline def catchFailure[X](collector: Try[X] => Unit)(inline body: => Unit): Unit =
      try body
      catch case e: Exception => collector(Failure(e))

For instance, the following code:

Async.blocking:
  f().map(_ * 2).collect(println)
  println("End!")

results in:

Success(2)
Success(4)
End!

flatMap waits for the inner flow to complete before collecting the next item:

Async.blocking:
  val startTime = System.currentTimeMillis()
  f().flatMap: i =>
    Flow:
      it.emit(i)
      sleep(2_000)
      it.emit(i + 1)
  .collect(i => println(s"$i at ${System.currentTimeMillis() - startTime} from start"))
  println("End!")

Result:

Success(1) at 1052 from start
Success(2) at 3054 from start
Success(2) at 3066 from start
Success(3) at 5072 from start
End!

[natsukagami]

This seems very promising! My initial idea with streams would be something similar, but letting the caller provide its own "source" and "sink", something like

type Flow[In, Out] = (ReadableChannel[In], SendableChannel[Out]) => Async ?=> Unit

(where Async ?=> Unit is fairly analogous to Task[Unit).

The idea of an operation which closes the channel but doesn't immediately disrupt the reader sounds like something we might consider with the existing channels themselves. It looks every useful!

It's only my initial impression though, let me ping @m8nmueller who might have more in-depth thoughts about streaming!

[He-Pin]

FYI, the current Java 22's Gather API does not define how to interactive timer/outer world, the new api in gears should support this.
And I did some experiment in pekko too apache/pekko#628 .

If we can built all this right, we can even build a Java wrapper for Java user:).

Thank you @tassiluca for sharing, but seems the API a little like the Kotlin https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/flow/FlowCollector.kt

I would like to see this api works in a Reactive way(dynamic push pull), the current pulling patten in Kotlin Flow has bad performance.

[m8nmueller]

Hi there!
Thanks a lot for your input already!

@tassiluca The idea of the TerminableChannel is indeed an important topic for our channel architecture that's necessary considering before putting Flows on top of it. I would rather call your Flow implementation a Source (Publisher, Observable, ...). For Flow, I agree with @natsukagami, though it might be interesting to return multiple tasks (List[Async ?=> Unit]) to easily create nested Sources where each layer adds a task. Besides, your FlowCollector is structurally pretty much equivalent to the writing end of a channel. I'm not sure about the collect idea of the implementation because it limits the freedom of how/when to handle the items. I'd prefer the source to expose - at some point - the reading end of a channel. Your solution for thread-safe access to Flows is very interesting, though I have skipped this aspect in my first thoughts.

You can find my implementation idea here: https://github.com/m8nmueller/gears/blob/streams/shared/src/main/scala/async/Stream.scala. I will continue working on this as part of my bachelor project at EPFL.

@He-Pin The Gather API seems like an interesting approach to allow custom transformers. I'll have a look at both the JEP and your work on Pekko. What exactly do you mean by "interactive timer/outer world"? I will mostly focus on pushing streams using bounded buffers as means of back-pressure, as proposed by Martin Odersky.

[He-Pin]

When I try to use the current Gather API to implement groupWeightedWithin like thing in Akka/Pekko-stream, but soon it cause problem under high concurrency and long run. in Akka/Pekko, there is an AsyncCallback for the stage to interactive with other even/source.

And here, as we already have the channel, the timer can just put event on the channel.