Java 8 Stream - parallel execution - different result - why?

Question

Let's say i have a List<Integer> ints = new ArrayList<>(); and i want to add values to it and compare the results of parallel execution using forEach() and Collectors.toList().

First i add to this list some values from an sequential IntStream and forEach:

 IntStream.range(0,10).boxed().forEach(ints::add);

And i get the correct result:

ints ==> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Now i .clear() the list and do the same thing in parallel:

IntStream.range(0,10).parallel().boxed().forEach(ints::add);

Now due to multithreading i get the incorrect result:

ints ==> [6, 5, 8, 9, 7, 2, 4, 3, 1, 0]

Now i switch to collecting the same Stream of Integers:

IntStream.range(0,10).parallel().boxed().collect(Collectors.toList());

And i get the correct result:

ints ==> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Question: Why does the two parallel executions produce different result's and why is the Collector producing the correct result?

If forEach produces a random result the Collector should too. I didn't specify any sorting and i think internally he is adding to a list like i did manually using forEach. Since he's doing it in parallel he's add method should get the values in unspecified order. Testing done i JShell.

EDIT: No duplicate here. I understand the linked question. WHy does the Collector produce the correct result? If he would be producing another random result i would not be asking.

`Collector` is designed to handle parallelism. It's part of the streams API, and it's why they have the `combiner()`. — Radiodef, Jul 23 '18 at 20:44
Okay, maybe the linked post if not 100% answering your question but it's at least 80% there and the remaining just requires a bit more research. nevertheless, reverted... — Ousmane D., Jul 23 '18 at 20:53
Collectors force the correct ordering to happen unless they're specifically constructed as order-independent. — Louis Wasserman, Jul 23 '18 at 21:07
Set c = Collectors.toList().characteristics(); System.out.println(c); [IDENTITY_FINISH] — K.Nicholas, Jul 23 '18 at 21:14
I feel like the title of this question could be a bit different regarding the answer of @teppic that i wanted. Any ideas? — Robert Niestroj, Jul 24 '18 at 09:13
@RobertNiestroj well it should have been a more _specific_ question and description in the first place. You modifying your post consistently which changes the topic entirely is something you should avoid for future posts. Also, if #teppic answer is what you were after, then it's a good gesture to mark it as the accepted answer. — Ousmane D., Jul 24 '18 at 09:25

score 11 · Accepted Answer · answered Jul 24 '18 at 02:45

The collect operation would produce unordered output if the Collector you passed it had different characteristics. That is, if the CONCURRENT and UNORDERED flags were set (see Collector.characteristics()).

Under the hood Collectors.toList() is constructing a Collector roughly equivalent to this:

Collector.of(
    // Supplier of accumulators
    ArrayList::new,
    // Accumulation operation
    List::add,
    // Combine accumulators
    (left, right) -> {
        left.addAll(right);
        return left;
    }
)

A bit of logging reveals the lengths that the collect operation is going to to maintain thread safety and stream order:

Collector.of(
    () -> {
        System.out.printf("%s supplying\n", Thread.currentThread().getName());
        return new ArrayList<>();
    },
    (l, o) -> {
        System.out.printf("%s accumulating %s to %s\n", Thread.currentThread().getName(), o, l);
        l.add(o);
    },
    (l1, l2) -> {
        System.out.printf("%s combining %s & %s\n", Thread.currentThread().getName(), l1, l2);
        l1.addAll(l2);
        return l1;
    }
)

logs:

ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 2 to []
ForkJoinPool-1-worker-1 accumulating 6 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 4 to []
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 5 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 3 to []
ForkJoinPool-1-worker-0 combining [3] & [4]
ForkJoinPool-1-worker-0 combining [2] & [3, 4]
ForkJoinPool-1-worker-1 combining [5] & [6]
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-0 accumulating 1 to []
ForkJoinPool-1-worker-1 accumulating 8 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 9 to []
ForkJoinPool-1-worker-1 combining [8] & [9]
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 7 to []
ForkJoinPool-1-worker-1 combining [7] & [8, 9]
ForkJoinPool-1-worker-1 combining [5, 6] & [7, 8, 9]
ForkJoinPool-1-worker-0 accumulating 0 to []
ForkJoinPool-1-worker-0 combining [0] & [1]
ForkJoinPool-1-worker-0 combining [0, 1] & [2, 3, 4]
ForkJoinPool-1-worker-0 combining [0, 1, 2, 3, 4] & [5, 6, 7, 8, 9]

You can see that each read from the stream is written to a new accumulator, and that they are carefully combined to maintain order.

If we set the CONCURRENT and UNORDERED characteristic flags the collect method is free to take shortcuts; only one accumulator is allocated and ordered combination is unnecessary.

Using:

Collector.of(
    () -> {
        System.out.printf("%s supplying\n", Thread.currentThread().getName());
        return Collections.synchronizedList(new ArrayList<>());
    },
    (l, o) -> {
        System.out.printf("%s accumulating %s to %s\n", Thread.currentThread().getName(), o, l);
        l.add(o);
    },
    (l1, l2) -> {
        System.out.printf("%s combining %s & %s\n", Thread.currentThread().getName(), l1, l2);
        l1.addAll(l2);
        return l1;
    },
    Characteristics.CONCURRENT,
    Characteristics.UNORDERED
)

Logs:

ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 6 to []
ForkJoinPool-1-worker-0 accumulating 2 to [6]
ForkJoinPool-1-worker-1 accumulating 5 to [6, 2]
ForkJoinPool-1-worker-0 accumulating 4 to [6, 2, 5]
ForkJoinPool-1-worker-0 accumulating 3 to [6, 2, 5, 4]
ForkJoinPool-1-worker-0 accumulating 1 to [6, 2, 5, 4, 3]
ForkJoinPool-1-worker-0 accumulating 0 to [6, 2, 5, 4, 3, 1]
ForkJoinPool-1-worker-1 accumulating 8 to [6, 2, 5, 4, 3, 1, 0]
ForkJoinPool-1-worker-0 accumulating 7 to [6, 2, 5, 4, 3, 1, 0, 8]
ForkJoinPool-1-worker-1 accumulating 9 to [6, 2, 5, 4, 3, 1, 0, 8, 7]

That's the answer i was looking for. – Robert Niestroj Jul 24 '18 at 06:16 — Robert Niestroj, Jul 24 '18 at 06:16

score 4 · Answer 2 · answered Jul 23 '18 at 21:04

First, I'd recommend going through Why is shared mutability bad? .

Second, there is an example provided by the authors under the "Side-effects" section which pretty much is doing something similar to what you're doing:

As an example of how to transform a stream pipeline that inappropriately uses side-effects to one that does not, the following code searches a stream of strings for those matching a given regular expression, and puts the matches in a list.

ArrayList<String> results = new ArrayList<>();
 stream.filter(s -> pattern.matcher(s).matches())
       .forEach(s -> results.add(s));  // Unnecessary use of side-effects!

If executed in parallel, the non-thread-safety of ArrayList would cause incorrect results, and adding needed synchronization would cause contention, undermining the benefit of parallelism. Furthermore, using side-effects here is completely unnecessary; the forEach() can simply be replaced with a reduction operation that is safer, more efficient, and more amenable to parallelization:

List<String>results =
         stream.filter(s -> pattern.matcher(s).matches())
               .collect(Collectors.toList());  // No side-effects!

So, you may still ask "Why does the Collector produce the correct result?".

Simply because the authors already have something in place to handle parallelism.

I could comment why I sort of disagree here, but instead... https://stackoverflow.com/a/51494189/1059372 — Eugene, Jul 24 '18 at 08:42

Eugene · Answer 3 · 2018-07-24T08:47:39.963

First things first, forEach is documented as :

The behavior of this operation is explicitly nondeterministic

So in a future version of jdk, even your non-parallel code could produce "incorrect" results, that is out-of-order results. Under the current implementation, only the parallel version will produce these kind of results; but again this is no guarantee, forEach is free to do whatever it wants internally, unlike forEachOrdered for example.

Preserving order or not is not a property of sequential or parallel, it solely depends on the operation that break this order or not; that's it (like explicitly calling unordered for example).

Collectors.toList on the other hand is a terminal operation that preserve the order. Generally, unless a terminal operation is explicit in it's documentation about order, it will preserver it. So for example, see Stream::generate:

Returns an infinite sequential unordered stream.

That being said, there are two orders in general, order in which intermediate operations are being processed and terminal operations are. The first ones are not defined, you can modify your example and check:

IntStream.range(0,10)
         .parallel()
         .peek(System.out::println) // out of order printing
         .boxed()
         .collect(Collectors.toList());

while the terminal operations order is preserved.

And the last point is that this:

....parallel().forEach(ints::add)

you simply got lucky to even see all the elements in the first place. You are adding from different threads multiple elements to a non-thread safe collection (ArrayList); you could have easily missed elements or have nulls in your ints. I bet that running this some number of times, would prove this.

Even if you switch to let's say Collections.synchronizedList(yourList), the order in which these will appear is still undefined, for the reasons stated above about forEach

Java 8 Stream - parallel execution - different result - why?

3 Answers3

Linked