Parallel Streams and the Primitive Ones

One word runs the pipeline on every core.

long total = numbers.parallelStream().mapToLong(i -> (long) i * i).sum();

Nothing else changes. The same filter, the same map, the same terminal operation. Java splits the source, runs the pipeline on each piece on a different thread, and combines the answers.

You can also switch an existing stream:

numbers.stream().parallel()...    // same thing
someStream.sequential()           // and back again

Measured on an 8 core machine, ten million elements, once everything is warm:

Now the same word on a small job. A thousand elements, twenty thousand times:

Four times slower. Same machine, same operation, one word different.

Splitting is not free. The list has to be divided, tasks have to be handed to a pool, threads have to be woken, and the partial answers have to be combined. On a thousand elements that costs more than the work itself.

The primitive streams remove the boxing.

IntStream.rangeClosed(1, 10_000_000).sum();   //  9 ms
numbers.stream().mapToInt(Integer::intValue).sum();  // 34 ms

IntStream, LongStream and DoubleStream carry primitives the whole way. Nothing is allocated, and they gain methods that only make sense for numbers:

IntStream.of(3, 1, 4).sum();       // 8
IntStream.of(3, 1, 4).average();   // OptionalDouble[2.666...]
IntStream.of(3, 1, 4).max();       // OptionalInt[4]
IntStream.of(3, 1, 4).summaryStatistics();   // count, sum, min, average, max, in one pass

Moving between the two families:

list.stream().mapToInt(String::length)   // Stream<String> -> IntStream
intStream.boxed()                        // IntStream -> Stream<Integer>
intStream.mapToObj(i -> "n" + i)         // IntStream -> Stream<String>

average returning OptionalDouble is Section 11.5 again. An empty stream has no average, so the type says so.

Answer from memory before opening the explanation. Then compare your rule with the answer and revisit the example if they differ.

1. When does `parallelStream` actually make things faster?

   Show the answer

   When there is enough work to pay for splitting it up, and the work is independent.

   Measured on an 8 core machine, summing squares: over ten million elements, serial took 37 ms and parallel took 7 ms. Over a thousand elements run twenty thousand times, serial took 66 ms and parallel took 277 ms.

   Same code, same machine. Four times faster in one case and four times slower in the other, and the only difference is how much work there was per split.

   The other half is the work itself. Splitting only helps if the pieces do not need to talk to each other. Anything touching shared state, or that has to happen in order, gets slower or wrong or both.

2. A parallel stream adds 100,000 numbers to an `ArrayList` and the list ends up with 40,787. What happened, and why was there no error?

   Show the answer

   Several threads called add on the same ArrayList at the same time, and ArrayList was never built for that.

   Each add writes a value into a slot and then increases the size. Two threads reading the same size, writing to the same slot, and both setting size to the same number means one value is lost. Sixty thousand times over.

   There was no error because nothing detected anything. No lock was broken, no rule was checked. Each thread did a correct sequence of steps and the steps interleaved. This is the race condition Phase XIV is about, arriving early.

   The fix is not a thread safe list. It is to stop writing into shared state at all and let collect gather the results, which is what it is designed for.

3. Why is `IntStream` worth having when `Stream<Integer>` already works?

   Show the answer

   Because Stream<Integer> holds objects, and every number in it is an Integer on the heap.

   Generics cannot hold primitives, from Section 9.3, so ten million numbers means ten million small objects to make, chase and collect. Measured: 34 ms against 9 ms for the same sum.

   IntStream carries int values the whole way, so nothing is allocated. It also gets methods that only make sense for numbers: sum, average, max, and summaryStatistics.

Write each solution in VS Code, predict its result, and then run the checks. Reading an example does not replace implementing it.

4 exercises110 pointsabout 105 minutes

import java.util.*;
import java.util.stream.*;

public class Crossover {
    static long time(Runnable r) {
        long t = System.nanoTime();
        r.run();
        return (System.nanoTime() - t) / 1_000_000;
    }

    public static void main(String[] args) {
        System.out.println("cores: " + Runtime.getRuntime().availableProcessors());

        // TODO: sum of squares, serial and parallel, at sizes
        //       1_000, 100_000, 1_000_000, 10_000_000
        // TODO: warm both up before timing. Run each size three times.
        // TODO: find roughly where parallel starts winning on YOUR machine
        // TODO: do the small size 20_000 times in a loop and compare again
    }
}

import java.util.*;
import java.util.stream.*;

public class Lost {
    public static void main(String[] args) {
        // TODO: add 100_000 numbers to a plain ArrayList from a PARALLEL stream
        // TODO: print the size. Run it five times. Write down all five numbers.
        // TODO: do the same with a serial stream. How many now?
        // TODO: fix it with collect or toList, and confirm you get 100000
        // TODO: try Collections.synchronizedList. Does the count come right?
        //       Time it against the collect version.
    }
}

import java.util.*;
import java.util.stream.*;

public class Primitives {
    public static void main(String[] args) {
        // TODO: sum 1 to 10_000_000 two ways: Stream<Integer> and IntStream. Time both.
        // TODO: average, max and summaryStatistics on an IntStream. Note the return types.
        // TODO: convert between the two families: boxed, mapToInt, mapToObj
        // TODO: try collect(Collectors.toList()) on an IntStream. Read the error.
        // TODO: average on an EMPTY IntStream. What comes back, and why not zero?
    }
}

import java.util.*;
import java.util.stream.*;

public class Sources {
    static long time(Runnable r) {
        long t = System.nanoTime();
        r.run();
        return (System.nanoTime() - t) / 1_000_000;
    }

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

        // TODO: same parallel sum over both. Predict which wins and by how much.
        // TODO: forEach on a parallel stream. Is the order what you expected?
        // TODO: forEachOrdered. Time it against forEach.
        // TODO: Stream.iterate(1, x -> x + 1).parallel().limit(1_000_000).sum().
        //       Predict whether parallel helps here at all.
    }
}