Scala JMH Benchmark Samples

Readme.md

Scala Benchmark Samples using JMH

These samples use scala-cli for execution and can also write JSON files to be used on https://jmh.morethan.io/.

To generate JSON, run the with the flag -- -rf json. Eg. scli --jmh ZStreamBench.scala -- -rf json.

Multiple files can be dropped into https://jmh.morethan.io/ to compare benchmarks between ZIO versions for example, just replace the ZIO version in the import, run the benchmark and rename the output JSON file. Repeat with all needed versions.

Examples

PipelineOps Benchmark

Link

Comparing ZIO Stream across versions

Link

PipelineBench.scala

// Run with `scli --jmh PipelineBench.scala`, to generate json output, add `-- -rf json` args
//> using scala 3.3.0
//> using options "-Wunused:all"

package bench

import java.util.concurrent.TimeUnit
import scala.util.chaining.*
import org.openjdk.jmh.annotations.*

@State(Scope.Benchmark)
@BenchmarkMode(Array(Mode.AverageTime))
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@Warmup(iterations = 10, time = 100, timeUnit = TimeUnit.MILLISECONDS)
@Measurement(iterations = 15, time = 100, timeUnit = TimeUnit.MILLISECONDS)
@Fork(1)
@Threads(1)
class PipelineBench:

  @Benchmark
  def classic(): Int =
    half(sum(triple(3))(2))

  @Benchmark
  def infix(): Int =
    3 |> triple |> sum(2) |> half

  @Benchmark
  def inline(): Int =
    3 |>> triple |>> sum(2) |>> half

  @Benchmark
  def utilpiped(): Int =
    3 pipe triple pipe sum(2) pipe half

  @Benchmark
  def utilchaining(): Int =
    3.pipe(triple).pipe(sum(2)).pipe(half)

  @Benchmark
  def aliased(): Int =
    3 ||> triple ||> sum(2) ||> half

/** Test functions */
val triple = (x: Int) => 3 * x
val sum    = (x: Int) => (y: Int) => x + y
val half   = (x: Int) => x / 2

/** with extension infix */
extension [A, B](a: A)
  infix def |>(f:   A => B): B = f(a)

/** or with inline */
extension [A](a:              A)
  inline def |>>[B](inline f: A => B): B = f(a)

/**
 * finally aliasing pipe (just using ||> to avoid conflicts with above examples)
 */
extension [A, B](a: A) inline def ||>(inline f: (A) => B): B = a.pipe(f)

/** Instead of */
// val classic = half(sum(triple(3))(2))
// println(s"Classic result: $classic")

/** Pipe the functions */
// val piped = 3 |> triple |> sum(2) |> half
// println(s"Piped result: $piped")

/** or using scala.util.chaining */
// import scala.util.chaining._
// val utilpiped = 3 pipe triple pipe sum(2) pipe half    // or
// val chaining = 3.pipe(triple).pipe(sum(2)).pipe(half)
// println(s"Scala util.piped result: $utilpiped")
// println(s"Chaining result: $chaining")

// val aliased = 3 ||> triple ||> sum(2) ||> half
// println(s"Scala aliased util.piped result: $aliased")

simplebench.scala

// Run with `scli --jmh SimpleBench.scala`, to generate json output, add `-- -rf json` args
//> using scala 3.3.0
//> using options "-Wunused:all"

package bench

import java.util.concurrent.TimeUnit
import org.openjdk.jmh.annotations.*

@State(Scope.Benchmark)
@BenchmarkMode(Array(Mode.AverageTime))
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@Warmup(iterations = 5, time = 100, timeUnit = TimeUnit.MILLISECONDS)
@Measurement(iterations = 10, time = 100, timeUnit = TimeUnit.MILLISECONDS)
@Fork(1)
class SimpleBenchmark:

  @Benchmark
  def foo(): Long =
    (1L to 10000000L).sum

ZStreamBench.scala

// Run with `scli --jmh ZStreamBench.scala`, to generate json output, add `-- -rf json` args
// Benchmark provided from https://github.com/zio/zio/issues/8158
//> using scala 3.3.0
//> using lib "dev.zio::zio:2.0.14"
//> using lib "dev.zio::zio-streams:2.0.14"
//> using lib "dev.zio::zio-profiling-jmh:0.2.0"

//> using options "-Wunused:all", "-Wvalue-discard", "-Wnonunit-statement"

package bench

import zio.*
import zio.stream.*
import zio.profiling.jmh.BenchmarkUtils
// import zio.{Scope as _, *} // Import all from zio and blackhole zio.Scope to avoid name clash with jmh.Scope

import org.openjdk.jmh.annotations.{Scope, *}
import java.util.concurrent.TimeUnit

@State(Scope.Benchmark)
@BenchmarkMode(Array(Mode.AverageTime))
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(1)
@Measurement(iterations = 10, timeUnit = TimeUnit.SECONDS, time = 1)
@Warmup(iterations = 15, timeUnit = TimeUnit.SECONDS, time = 1)
@Threads(1)
class AggregateAsyncBench:
  @Benchmark
  def SingleStreamAggregateAsyncBench(): Unit = executeZio:
    ZStream
      .fromIterable(1 to 100, 10)
      .aggregateAsync(ZSink.foldLeft[Int, Chunk[Int]](Chunk.empty)((chunk, el) => chunk.appended(el)))
      .runForeach(_ => ZIO.sleep(5.millis))

  @Benchmark
  def ManyStreamsAggregateAsyncBench(): Unit = executeZio:
    ZStream
      .fromIterable(sources)
      .flatMapPar(16)(identity)
      .aggregateAsync(ZSink.foldLeft[Int, Chunk[Int]](Chunk.empty)((chunk, el) => chunk.appended(el)))
      .runForeach(_ => ZIO.sleep(5.millis))

  private val sources = (1 to 10).map(_ => ZStream.fromIterable(1 to 100, 10))

  // private def runZio[A](zio: Task[A]): A =
  //   Unsafe.unsafe(implicit unsafe => Runtime.default.unsafe.run(zio).getOrThrow())

  private def executeZio[A](zio: Task[A]): A =
    BenchmarkUtils.unsafeRun(zio)