Luke Stephenson 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

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

	➜ 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

	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

	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]