When Greed Is Good: Understanding Java Stream Performance

📥 Download source code

About the Speaker

• Cay Horstmann
• Author of Core Java (14 editions since 1996), Java/JavaScript/Scala for the Impatient, etc.
• Java Champion
• https://horstmann.com

Why Use Streams?

• Declarative (“what, not how”)

  int largest = Stream.of(points)
      .filter(Points::isShort)
      .mapToInt(Points::shortLengthSquared)
      .max()
      .orElseThrow();
  

• instead of

  int largest = Integer.MIN_VALUE;
  for (int i = 0; i < points.length; i++) {
      Point p = points[i];
      if (Points.isShort(p)) {
          int lengthSquared = Points.shortLengthSquared(p);
          if (lengthSquared > largest) {
              largest = lengthSquared;
          }
      }
  }
  

• Can be parallelized (sometimes)
• Briefly gorgeous in 2014
• Having a moment again, e.g. Venkat Subramaniam at Dev.java
• Are everywhere in AI generated code
• Gatherer API in Java 25

What About Performance?

• Use a benchmarking tool, not time or System.nanoTime
• Nice intro to JMH by François Martin
• Make sure expensive setup operations (random numbers, copying) are not measured
• Streams are holding up well in this example (lower is better):

  $ mvn clean install
  $ java -jar target/benchmarks.jar FilterMap
  Benchmark             Mode  Cnt     Score    Error  Units
  FilterMap.withArray   avgt  100  1711.600 ± 10.471  ms/op
  FilterMap.withStream  avgt  100  1798.188 ± 11.118  ms/op
  

• Benchmarks by Nikolai Parlog indicate modest stream overhead
• Benchmarks by Magnus Åkerfeldt: Streams sometimes less competitive
  • Small data sets
  • Long pipelines

Peeking Under the Hood

• Trace into stream calls with debugger
• Intellij: Settings → Debugger → Stepping
  • ☑ Always do smart step into
  • ☐ Do not step into these classes (you want to step in)
• In FilterMap demo, set breakpoint to .filter(...)
• Step into (F7) to see that it returns new StatelessOp<>(...).
• Look at accept method of sink.
• Step out (Shift+F8), step into .map(...)
• Also returns new StatelessOp<>(...). Look at accept method.
• 🔑 Key point: Not doing any filtering or mapping yet, just setting up the pipeline
• This is the pipeline:

Pushing Elements Through

• On to the terminal op. It goes through a number of steps, which we’ll skip. Just set a breakpoint in AbstractPipeline.copyInto (line 566)
• Note that the first branch executes:

  spliterator.forEachRemaining(wrappedSink);
  

• That is good—it is an efficient way of pushing all elements through the pipeline
• The spliterator is an iterator through the array elements
• Step inside forEachRemaining
• Aside: Coding style helps the JIT
• Step inside action.accept: First pipeline stage checks the filter predicate
• The element might not pass, so set a breakpoint at downstream.accept(u) and resume (F9)
• Step inside to see the map being computed
• We could step more to see that a ReducingSink accepts the element, updating the maximum
• All elements are pushed through the pipeline:

Watching the JIT

• Stream pipeline must be less efficient than hand-crafted loop
• But the JIT can help—let’s watch it
• Download hsdis-amd64.so from https://www.chriswhocodes.com/hsdis/
• Set environment variable to directory containing library:

  export LD_LIBRARY_PATH=/opt/lib
  

• Run with flags:

  java -XX:+UnlockDiagnosticVMOptions -XX:+PrintAssembly -XX:+LogCompilation \
      -XX:LogFile=jitwatch.log -cp target/classes/ --enable-preview \
      com.horstmann.streams.demos.FilterMap 300000
  

• View the logs with JITWatch

  export GDK_SCALE=2
  java -jar /opt/jitwatch/jitwatch-ui-1.5.0-shaded-linux-x64.jar
  

• 🔑 Key point: JIT inlines pipeline operations
• As long as the code is 🔥
• In this case, body of loop in forEachRemaining completely inlined

A Change of Strategy

• Let’s add limit after the map:
• Trace again into AbstractPipeline.copyInto. As before, set a breakpoint into line 566
• Step through, note the second branch executes (copyIntoWithCancel)
• Step inside, then step inside forEachWithCancel

  boolean cancelled;
  do { } while (!(cancelled = sink.cancellationRequested()) && spliterator.tryAdvance(sink));
  return cancelled;
  

• Cancellation means the downstream cancels, nothing to do with threads
• Happens when the limit is reached
• 🔑 Key point: Evaluation strategy depends on pipeline structure
• Here, directed by StreamOpFlag.SHORT_CIRCUIT

Short Circuit Affects Performance

• Twice as slow as the equivalent loop:

  FilterMapLimit.withArray   avgt  100   664.089 ± 5.617  ms/op
  FilterMapLimit.withStream  avgt  100  1169.370 ± 7.015  ms/op
  

• Make flame graph:

  java -jar target/benchmarks.jar \
      -prof async:output=flamegraph\;libPath=/opt/async-profiler/lib/libasyncProfiler.so \
      FilterMapLimit
  

• Note how much time is spent in Sink$ChainedReference.cancellationRequested
• Compare with array
• JITWatch gives a clue: Sink.cancellationRequested can't be inlined
• 🔑 Key point: Change in evaluation strategy can have substantial performance impact

Parallel Streams

• Parallel stream splits into subranges

  int largest = Stream.of(points)
      .parallel()
      .filter(Points::isShort)
      .mapToInt(Points::shortLengthSquared)
      .max()
      .orElseThrow();
  

• Subrange tasks execute concurrently (one thread per core)
• Happy day scenario: n-fold speedup

  Benchmark                                           Mode  Cnt        Score       Error  Units
  ParallelFilterMap.withParallelStream                avgt  100   154229.126 ±  2323.488  us/op
  ParallelFilterMap.withSerialStream                  avgt  100  1828765.262 ±  9945.505  us/op
  

Not All Days Are Happy

• What is wrong with this pipeline?

  long result = Stream.of(points)
      .parallel()
      .mapToLong(Points::lengthSquared)
      .filter(p -> p < DIST)
      .limit(POINTS / 10)
      .max()
      .orElseThrow();
  

• Serial is faster!

  ParallelMapFilterLimit.withParallelStream           avgt  100  275946.600 ± 14045.715  us/op
  ParallelMapFilterLimit.withSerialStream             avgt  100  236369.864 ±  2717.238  us/op
  

• Problem: Parallelization is limited since we don’t know how many elements are needed
• Parallel map never speculatively executes—assumes that the mapper is expensive
• 🔑 Key point: Parallel scheduling can be slow if the pipeline is unsuitable
• Remedy:

  long result = Stream.of(state.points)
      .parallel()
      .unordered()
      ...
  
  ParallelMapFilterLimit.withUnorderedParallelStream  avgt  100   56743.103 ±  1841.688  us/op
  

Gatherers

• JEP 485 introduces gather for intermediate operations
• Five built-in gatherers
• Fixed and sliding window:

  IntStream.range(0, 10).boxed().gather(Gatherers.windowFixed(4)).toList()
      // [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9]]
  IntStream.range(0, 10).boxed().gather(Gatherers.windowSliding(4)).toList()
      // [[0, 1, 2, 3], [1, 2, 3, 4], [2, 3, 4, 5], [3, 4, 5, 6], [4, 5, 6, 7], [5, 6, 7, 8], [6, 7, 8, 9]]
  

• Useful to compare/combine/group adjacent elements
• Scan and fold (not common):

  Integer[] digits = { 1, 7, 2, 9 };
  Stream.of(digits).gather(Gatherers.scan(() -> 0, (x, y) -> x * 10 + y)) // [1, 17, 172, 1729]
  Stream.of(digits).gather(Gatherers.fold(() -> 0, (x, y) -> x * 10 + y)) // [1729]
  

• Run concurrently on virtual threads:

  var images = Stream.of(urls).gather(Gatherers.mapConcurrent(60, this::fetchImage)).toList()
  

• Additional gatherers: gatherers4j, more-gatherers, packrat

Implementing Gatherers

• Example: zip with index ●●● ... → (0, ●), (1, ●), (2, ●) ...

  static <T> Gatherer<T, ?, Map.Entry<Long, T> zipWithIndex() {
      class Index { long index = 0; }
      return Gatherer.ofSequential(
              Index::new,
              Gatherer.Integrator.ofGreedy((state, element, downstream) ->
                      return downstream.push(Map.entry(state.index++, element))));
  }
  

• “Integrate” upstream elements into state, send results downstream when ready
• Gatherer can be sequential or not (parallelizable)
• Declared with Gatherer.ofSequential/Gatherer.of
• Parallelizable gatherer combines states
• Not always feasible, e.g. windowSliding, zipWithIndex
• Integrator can be greedy or not (picky)
• Declared with Gatherer.Integrator.ofGreedy/Gatherer.Integrator.of
• A greedy gatherer never short-circuits

When Greed is Good

• When a gatherer isn’t greedy, pipeline can’t use forEachRemaining, must check for cancellation
• Any other benefits of greed?
• JEP/API docs are silent, core-libs-dev discussions unilluminating: “Greedy allows for less signal tracking which in aggregate over several operations can have a noticeable advantage. Keep in mind that this is primarily around sequential performance, which in my experience is vastly more common than parallel streams.”
• At JChateau 2024, Sven Woltmann presented a curious example about a parallel stream with a sequential gatherer:

  var result = listOfIntegers.parallelStream()
      .map(...)
      .gather(mapSequential(...))
      .toList();
  

• Benchmarks:

  Benchmark                                  Mode  Cnt     Score    Error  Units
  Gathering.parallelThenParallelizableGreedy avgt   25   396.633 ±  9.675  us/op
  Gathering.parallelThenParallelizablePicky  avgt   25   425.897 ± 17.134  us/op
  Gathering.parallelThenSequentialGreedy     avgt   25   771.995 ± 14.370  us/op
  Gathering.parallelThenSequentialPicky      avgt   25  1739.768 ± 74.636  us/op
  Gathering.serialThenParallelizableGreedy   avgt   25  1028.898 ± 15.994  us/op
  Gathering.serialThenParallelizablePicky    avgt   25  1026.294 ± 10.314  us/op
  Gathering.serialThenSequentialGreedy       avgt   25  1039.664 ± 20.719  us/op
  Gathering.serialThenSequentialPicky        avgt   25  1084.909 ± 30.541  us/op
  
  

• For serial performance, greedy is small beer
• For sequential gatherers in parallel streams, greedy is a big win
• Without it, parallel map won’t precompute values because they might never be needed

The End

• Streams can lead to more maintainable code…
• …when used with taste and restraint
• Expect a performance penalty over hand-coded loop
• The JIT can do only so much
• Understand pipeline hazards
• Parallelization can work great in happy day scenarios
• Multiple ways to unhappiness
• Keep using streams for clarity
• Profile to find performance issues
• Benchmark, JIT watch
• Develop an accurate mental performance model
• Thank you!
• Questions?
• Feedback (QR code at right)