lukestephenson

Diving into Monix / Akka performance compared to fs2 and ZIO-streams

Disclaimer: Firstly, I'm not an expert in any of these libraries. I'm writing this from a position of sharing what I have learnt so far, but also hoping to be criticised and learn a lot more in the process.

Both fs2 and zio-streams document the need to make use of "Chunking". Here is what the zio-streams docs say:

Every time we are working with streams, we are always working with chunks. There are no streams with individual elements, these streams have always chunks in their underlying implementation. So every time we evaluate a stream, when we pull an element out of a stream, we are actually pulling out a chunk of elements.

So why streams are designed in this way? This is because of the efficiency and performance issues. Every I/O operation in the programming world works with batches. We never work with a single element.

While this is true that IO operations work with batches, from a programmers perspective it is sometimes easier

lukestephenson

Background

I'm a long time user of reactive streams in Scala. Originally with akka-streams, and then Monix Observable.

The transition from akka streams to Monix Observable I found pretty straight forward. I was motivated by wanting to use a lazy effect system, rather than working with scala.concurrent.Future.

More recently I've been considering fs2 and Zio Streams as an alternative to Monix Observable. This has largely been motivated by the fact that Monix doesn't have much ongoing development and is stuck on cats-effect 2.

I've not found the transition from Monix Observable to Zio Streams as easy as my earlier transition (from akka -> monix). While the ergonomics of the API are similar, performance characterics differ hugely.

Chunking

lukestephenson

package luke

import scala.Tuple$package.EmptyTuple
import scala.compiletime._
import scala.compiletime.ops.int._
import scala.annotation.showAsInfix

type Bounded[MIN <: Int, MAX <: Int] <: Int = MAX match 
  case MIN => MIN
  case ? => MAX | Bounded[MIN,MAX-1]

lukestephenson

sbt:stream-benchmarks> jmh:run -i 3 -wi 3 -f1 -t1 .*LargeStreamBenchmark.*
[info] Running (fork) org.openjdk.jmh.Main -i 3 -wi 3 -f1 -t1 .*LargeStreamBenchmark.*
[info] # JMH version: 1.19
[info] # VM version: JDK 1.8.0_144, VM 25.144-b01
[info] # VM invoker: /Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home/jre/bin/java
[info] # VM options: <none>
[info] # Warmup: 3 iterations, 1 s each
[info] # Measurement: 3 iterations, 1 s each
[info] # Timeout: 10 min per iteration
[info] # Threads: 1 thread, will synchronize iterations

lukestephenson

➜  stream-benchmarks git:(master) sbt
[info] Loading settings from plugins.sbt ...
[info] Loading project definition from /Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/project
[info] Loading settings from build.sbt ...
[info] Set current project to stream-benchmarks (in build file:/Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/)
[info] sbt server started at 127.0.0.1:4112
sbt:stream-benchmarks> jmh:run -i 3 -wi 3 -f1 -t1 .*Benchmark
[info] Packaging /Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/target/scala-2.12/stream-benchmarks_2.12-0.1.jar ...
Processing 9 classes from /Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/target/scala-2.12/classes with "reflection" generator
Writing out Java source to /Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/target/scala-2.12/src_managed/jmh and resources to /Users/luke.stephenson/projects/lukestephenson/stream-benchmarks/target/scala-2.12/resource_managed/jmh