Parallel streams and when they slow things down

Understanding Parallel streams and when they slow things down

A parallel stream does not hand your pipeline to threads element by element. The terminal operation asks the source's Spliterator to keep calling trySplit() until there are enough chunks for the common ForkJoinPool, each chunk is reduced independently on some worker, and the partial results are combined back up the tree. That adds three costs a sequential stream never pays: splitting the source, forking and stealing tasks, and merging partials. Parallelism wins only when element count times CPU work per element is large enough to hide those fixed costs.

The source decides whether the split step is nearly free. An array, an ArrayList, or IntStream.range knows its exact size and splits by halving an index range, so a chunk boundary costs a couple of integer assignments and the halves are balanced. A LinkedList, a Stream.iterate sequence, or anything built on an Iterator can only move forward one element at a time, so trySplit() has to walk the source and copy elements into an array before a worker can touch them, which means the calling thread performs the very traversal you wanted to parallelize. Boxed elements make it worse: chasing Integer references across a huge list is memory-bound, so extra threads compete for cache lines and bandwidth instead of adding compute.

The other half of the cost model is what happens on the way back. sum(), count(), max() and a primitive reduce merge in constant time, while collect(toList()) or toMap() copies or rehashes in proportion to the result size, and every order-sensitive stage such as limit, skip, findFirst, sorted or forEachOrdered forces buffering so the parallel answer still respects encounter order. Those are exactly the pipelines where parallel does strictly more work to produce the same answer. And because every parallel stream in the JVM shares one common pool by default, a lambda that blocks on I/O or nests another parallel stream stalls unrelated code elsewhere in the process.

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Spliterator;

public class SplitBudget {

    static void inspect(String label, Spliterator<Integer> s) {
        long before = s.estimateSize();
        Spliterator<Integer> firstPart = s.trySplit();
        long taken = (firstPart == null) ? 0 : firstPart.estimateSize();
        System.out.println(label + ": estimate=" + before
                + " firstSplit=" + taken + " remaining=" + s.estimateSize());
    }

    public static void main(String[] args) {
        List<Integer> arrayList = new ArrayList<>();
        List<Integer> linkedList = new LinkedList<>();
        for (int i = 0; i < 1000; i++) {
            arrayList.add(i);
            linkedList.add(i);
        }

        inspect("ArrayList", arrayList.spliterator());
        inspect("LinkedList", linkedList.spliterator());

        long sum = arrayList.parallelStream().mapToLong(Integer::longValue).sum();
        System.out.println("parallel sum=" + sum);
    }
}

Worked examples

import java.util.Spliterator;
import java.util.stream.Stream;

public class UnsplittableSource {
    public static void main(String[] args) {
        Spliterator<Integer> s = Stream.iterate(1, i -> i + 1).spliterator();

        System.out.println("SIZED=" + s.hasCharacteristics(Spliterator.SIZED));
        System.out.println("estimate=" + s.estimateSize());

        Spliterator<Integer> batch = s.trySplit();
        System.out.println("batch handed out=" + (batch != null));
        System.out.println("estimate after split=" + s.estimateSize());
    }
}

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.IntStream;

public class OwnPool {
    public static void main(String[] args) throws Exception {
        Set<String> threads = ConcurrentHashMap.newKeySet();
        ForkJoinPool pool = new ForkJoinPool(2);
        try {
            ForkJoinTask<Long> task = pool.submit(() -> IntStream.rangeClosed(1, 1000)
                    .parallel()
                    .peek(i -> threads.add(Thread.currentThread().getName()))
                    .asLongStream()
                    .sum());
            long sum = task.get();
            System.out.println("sum=" + sum);
            System.out.println("touched common pool="
                    + threads.stream().anyMatch(n -> n.contains("commonPool")));
        } finally {
            pool.shutdown();
        }
    }
}

import java.util.stream.IntStream;
import java.util.stream.Stream;

public class ModeIsPipelineWide {
    public static void main(String[] args) {
        Stream<Integer> a = Stream.of(1, 2, 3).parallel().map(i -> i * 2);
        System.out.println("a parallel? " + a.isParallel());
        System.out.println("a sum=" + a.mapToInt(Integer::intValue).sum());

        IntStream b = IntStream.range(0, 10).parallel().map(i -> i * 2).sequential();
        System.out.println("b parallel? " + b.isParallel());
        System.out.println("b sum=" + b.sum());
    }
}

Important notes

Common mistakes

Try it yourself