8

Inspired by this question, I started to play with ordered vs unordered streams, parallel vs sequential streams and terminal operations that respect encounter order vs terminal operations that don't respect it.

In one answer to the linked question, a code similar to this one is shown:

List<Integer> ordered = Arrays.asList(
    1, 2, 3, 4, 4, 3, 2, 1, 1, 2, 3, 4, 4, 3, 2, 1, 1, 2, 3, 4);
List<Integer> result = new CopyOnWriteArrayList<>();

ordered.parallelStream().forEach(result::add);

System.out.println(ordered);
System.out.println(result);

And the lists are indeed different. The unordered list even changes from one run to another, showing that the result is actually non-deterministic.

So I created this other example:

CopyOnWriteArrayList<Integer> result2 = ordered.parallelStream()
        .unordered()
        .collect(Collectors.toCollection(CopyOnWriteArrayList::new));

System.out.println(ordered);
System.out.println(result2);

And I expected to see similar results, as the stream is both parallel and unordered (maybe unordered() is redundant, since it's already parallel). However, the resulting list is ordered, i.e. it's equal to the source list.

So my question is why the collected list is ordered? Does collect always respect encounter-order, even for parallel, unordered streams? Is it the specific Collectors.toCollection(...) collector the one that forces encounter-order?

fps
  • 33,623
  • 8
  • 55
  • 110
  • Very interesting questions, but the last one is off-topic. You'd better remove it from your post. – JB Nizet Jul 06 '17 at 16:38
  • Yes, it would be better. – JB Nizet Jul 06 '17 at 16:40
  • 3
    btw, it is slowly becoming the standard on SO for streams; here is an absolute must read from Holger : https://stackoverflow.com/questions/41894173/java-8-findfirst-and-encounter-order – Eugene Jul 06 '17 at 16:52
  • 5
    `unordered()` is *not* redundant, as parallel does not imply unordered. That has been explained [here](https://stackoverflow.com/a/29218074/2711488). It’s wrong to say that an unordered terminal operation “does not respect” the encounter order. Its unordered characteristic implies that the order has no meaning for the result, like when summing ints or collecting to a `HashSet`. When you are saying `unordered()` explicitly, you are also saying that the order is irrelevant, even if you can recognize an order in the result list. And if the order is irrelevant to you, there’s no question about it… – Holger Jul 06 '17 at 17:15

3 Answers3

6

Collectors.toCollection returns a Collector which lacks the Collector.Characteristics.UNORDERED characteristic. Another collector which specified Collector.Characteristics.UNORDERED might behave differently.

That said: "unordered" means no guarantees, not guaranteed to vary. If the library finds it easiest to treat an unordered collection as ordered, it is allowed to do so, and that behavior is allowed to change release to release, on Tuesdays, or if there is a full moon.

(Note also that Collectors.toCollection does not require you to use a concurrent collection implementation if you're going to use parallel streams; toCollection(ArrayList::new) would work fine. That's because the collector doesn't have the Collector.Characteristics.CONCURRENT characteristic, so it uses a collection strategy that works for non-concurrent collections even with parallel streams.)

If you use an unordered stream but a collector that isn't UNORDERED, or vice versa, I doubt you get any guarantees from the framework. If there were a table, it would say "HERE BE DRAGONS UNDEFINED BEHAVIOR." I'd also expect some differences for different kinds of chained operations here, e.g. Eugene mentions findFirst varies here, even though findFirst is inherently an ordered operation -- unordered().findFirst() becomes equivalent to findAny().

For Stream.collect, I believe the current implementation has three strategies it chooses between:

  • Sequential: starts one accumulator, accumulates elements into it (in encounter order, because why would you bother doing work to shuffle the elements? just accept them in the order you get them), calls the finisher.
  • Parallel execution, concurrent collector, and the stream or the collector are unordered: one accumulator, shards the input, worker threads process the elements from each shard and add elements to the accumulator when they're ready, calls the finisher.
  • Parallel execution, anything else: shards the input into N shards, each shard gets sequentially accumulated into its own distinct accumulator, the accumulators get combined with the combiner function, calls the finisher.
Louis Wasserman
  • 191,574
  • 25
  • 345
  • 413
  • 2
    So, in this case, the fact that the order is preserved is an implementation detail that could change later, right? toCollection commits to respecting the order, but the stream is unordered, so the output might be in a different order from the source, right? – JB Nizet Jul 06 '17 at 16:43
  • 1
    @JBNizet I'd say yes, that would be spec-compliant. – Louis Wasserman Jul 06 '17 at 16:45
  • 4
    @JBNizet exactly. In the linked answer this exact diff is provided using `findFirst` that actually changed from 8 to 9. – Eugene Jul 06 '17 at 16:46
  • 1
    @JBNizet On a side note, implementations are allowed to do that *generally*. Another example would be `AtomicInteger#weakCompareAndSet` and `AtomicInteger#compareAndSet` that did the *same thing* before java-9. – Eugene Jul 06 '17 at 16:57
  • 2
    @Eugene I was more referring to the presence of the word "first" in the method name as making it intuitively an ordered operation. I already said that `unordered().findFirst()` is equivalent to `findAny()`, an unordered operation. We're on the same page, we just phrased it differently. – Louis Wasserman Jul 06 '17 at 17:06
  • @LouisWasserman ah! makes sense, I'll remove the comment then. – Eugene Jul 06 '17 at 17:12
4

Under the current implementation I've checked java-8 and java-9 the unordered flag is ignored for the collect phase for non-concurrent collectors (Collector.Characteristics.UNORDERED is not set). The implementations are allowed to do that, somehow similar question.

In the same question that you linked, I did provide the example of how findFirst actually changed from jdk-8 to jdk-9.

Eugene
  • 117,005
  • 15
  • 201
  • 306
1

the documentation Stream#collect has already mentioned:

When executed in parallel, multiple intermediate results may be instantiated, populated, and merged so as to maintain isolation of mutable data structures. Therefore, even when executed in parallel with non-thread-safe data structures (such as ArrayList), no additional synchronization is needed for a parallel reduction.

which means the Stream#collect does two primary thing: split & merge.

but I have a special example in jdk-8 that you can fetch the different results, :). when creates a unordered stream by Stream#generate, then you can fetch different result on Collectors#toList, for example:

Set<Set<Integer>> result = IntStream.range(0, 10).mapToObj(__ -> {
        return unordered().parallel().collect(toSet());
}).collect(toSet());

assert result.each.size() == 100000; // ok
//                   v--- surprised, it was pass 
assert result.size() > 1; 

Stream<Integer> unordered() {
    AtomicInteger counter = new AtomicInteger();
    return Stream.generate(counter::getAndIncrement).limit(10000);
}
holi-java
  • 29,655
  • 7
  • 72
  • 83