When Greed Is Good: Understanding Java Stream Performance

.png

📥 Download source code

About the Speaker

.jpg

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?

.png

Can be parallelized (sometimes)

Declarative (“what, not how”)

Point largest = Stream.of(points)
    .filter(Points::isShort)
    .max(Comparator.comparingInt(Points::lengthSquared))
    .orElseThrow();

instead of

Point largest = points.get(0);
for (int i = 1; i < points.size(); i++) {
    Point p = points.get(i);
    if (Points.isShort(p)) {
        if (Points.lengthSquared(p) > Points.lengthSquared(largest))
            largest = p;
    }
}

Briefly gorgeous in 2014
Having a moment again ten years later
Gatherer API in Java 25

Why Not Always Streams?

  static int numberOfBrickWithChildrenWithAtLeastTwoParents(String input) {
    var grid = new HashMap<Pos, Level>();
    return input.lines()
        .map(l -> {
          var array = Arrays.stream(l.split(",|~")).mapToInt(Integer::parseInt).toArray();
          return new Brick(array[0], array[3], array[1], array[4], array[2], array[5]);
        })
        .collect(groupingBy(Brick::z1, TreeMap::new, toList()))
        .values().stream()
        .flatMap(List::stream)
        .map(brick -> {
          var onTop = brick.positions()
              .map(pos -> grid.getOrDefault(pos, new Level(null, 0)))
              .collect(groupingBy(Level::z, TreeMap::new, flatMapping(l -> Stream.ofNullable(l.brick), toSet())))
              .lastEntry();
          var level =  new Level(brick, 1 + brick.z2 - brick.z1 + onTop.getKey());
          brick.positions().forEach(pos -> grid.put(pos, level));
          return new Pair<>(brick, onTop.getValue());
        })
        .collect(teeing(
            toMap(Pair::u, pair -> pair.v().size()),
            flatMapping(pair -> pair.v().stream().map(b -> new Pair<>(b, pair.u())),
                groupingBy(Pair::u, mapping(Pair::v, toList()))),
            (parentMap, childrenMap) ->
                (int) parentMap.keySet().stream()
                    .filter(parent -> childrenMap.getOrDefault(parent, List.of()).stream()
                        .allMatch(child -> parentMap.get(child) > 1))
                    .count()
        ));
  }

Source Rémi Forax | Advent of Code 2023 Day 22

What About Performance?

.webp

JMH benchmark with pre-initialized state.randomPointArray, comparing

List<Point> points = List.of(state.randomPointArray);
Point largest = points.get(0);
for (int i = 1; i < points.size(); i++) {
    Point p = points.get(i);
    if (Points.isShort(p)) {
        if (Points.lengthSquared(p) > Points.lengthSquared(largest))
            largest = p;
    }
}

with

Point largest = Stream.of(state.randomPointArray)
    .filter(Points::isShort)
    .max(Comparator.comparingInt(Points::lengthSquared))
    .orElseThrow();

Streams are almost twice as fast!

FilterMaxBenchmark.withCollection  avgt  100  355.147 ± 3.803  ms/op
FilterMaxBenchmark.withStream      avgt  100  205.245 ± 2.769  ms/op

Did you know that List.of makes a copy?

With array and points.get[i]:

FilterMaxBenchmark.withArray       avgt  100  178.251 ± 2.087  ms/op
FilterMaxBenchmark.withStream      avgt  100  205.245 ± 2.769  ms/op

Nice intro to JMH by François Martin
Benchmarks by Nikolai Parlog indicate modest stream overhead

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 FilterMapDemo, 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:

Image created with https://imagetodrawio.com

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:

A Change of Strategy

Let’s add .limit(5) 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

Watching the JIT

.png

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/ into path/to/lib directory
Set environment variable
```
export LD_LIBRARY_PATH=path/to/lib
```

Run with flags:

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

View the logs with JITWatch

export GDK_SCALE=2
java --enable-preview -jar jitwatch-ui-1.4.9-shaded-linux-x64.jar

🔑 Key point: JIT inlines pipeline operations into a single loop
Of course, not as good as hand-coded loop

Memory Consumption

.png

Stream pipeline is made up of:
- A node for every intermediate operation
- Inner class instances for ops
- Spliterators
Structure set up once
No per-iteration allocations
Example: Looping 100_000 times over stream calculation produced about 20MB of garbage
Observe with Java Flight Recorder/Java Mission Control
🔑 Key point: Pay attention to streams inside loop, nested streams

Parallel Streams

Parallel stream splits into n subranges (n = number of cores)

Point largest = Stream.of(randomPointArray)
    .parallel()
    .filter(Points::isShort)
    .map(Points::lengthSquared)
    .max(Integer::compare)
    .orElseThrow();

Subrange tasks execute on global fork-join pool
Results are assembled in order

Happy day scenario: n-fold speedup

ConcurrentBenchmark1.withParallelStream  avgt  100   19.082 ± 0.212  ms/op
ConcurrentBenchmark1.withSerialStream    avgt  100  203.989 ± 3.187  ms/op

Not All Days Are Happy

What is wrong with this pipeline?

long result = Stream.of(state.randomPointArray)
    .parallel()
    .mapToLong(p -> (long) p.x * p.x + (long) p.y * p.y)
    .filter(p -> p < Integer.MAX_VALUE)
    .limit(ITERATIONS / 10)
    .min()
    .orElseThrow();

Serial is faster!

ConcurrentBenchmark2.withParallelStream  avgt  100  27.127 ± 1.876  ms/op
ConcurrentBenchmark2.withSerialStream    avgt  100  10.579 ± 0.339  ms/op

Ask the Stream Planner
Problem: Can’t parallelize since we don’t know how many elements are needed
Parallel map assumes that the mapper is expensive

Remedy:

long result = Stream.of(state.randomPointArray)
    .parallel()
    .unordered()
    ...

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:

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]

mapConcurrent—see below
Additional gatherers: gatherers4j, more-gatherers, packrat

Implementing Gatherers

Turn stream into stream of pairs (index, element):

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
MyBenchmark.parallelThenParallelizableGreedy avgt   25   396.633 ±  9.675  us/op
MyBenchmark.parallelThenParallelizablePicky  avgt   25   425.897 ± 17.134  us/op
MyBenchmark.parallelThenSequentialGreedy     avgt   25   771.995 ± 14.370  us/op
MyBenchmark.parallelThenSequentialPicky      avgt   25  1739.768 ± 74.636  us/op
MyBenchmark.serialThenParallelizableGreedy   avgt   25  1028.898 ± 15.994  us/op
MyBenchmark.serialThenParallelizablePicky    avgt   25  1026.294 ± 10.314  us/op
MyBenchmark.serialThenSequentialGreedy       avgt   25  1039.664 ± 20.719  us/op
MyBenchmark.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

Blocking Tasks

Parallel streams do not work well for tasks that mostly block

String task(int i) {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException _) {
    }
    return i + " " + Thread.currentThread().toString();
}

Sequential stream: 60 seconds:

IntStream.range(0, 60).mapToObj(this::task).toList()

Parallel stream: 5 seconds (on my machine)

IntStream.range(0, 60).parallel().mapToObj(this::task).toList()

But it blocks the global fork-join pool
Really bad if other concurrent tasks do the same
Isn’t that what virtual threads are for?

mapConcurrent: 1 second

IntStream.range(0, 60).boxed().gather(Gatherers.mapConcurrent(60, this::task)).toList()

Also check out Grzegorz Piwowarek’s Parallel Collectors

The End

.jpeg

Streams can lead to more maintainable code…
…when used with taste and restraint
Some performance penalty over hand-coded loop
- More code, memory allocations
- The JIT can do only so much
Understand pipeline optimizations
Understand pipeline hazards
Parallelization can work great in happy day scenarios
Multiple ways to unhappiness
Keep using streams
Have a mental performance model
Thank you!
Questions?