Need an explanation for stream().reduce vs parallelStream().reduce output

[H Paul]

1. Simple Use case: Get the list of Class name.

2. See below Code 1.

I want to learn/understand the mechanical of many things simultaneously.
I want to use Stream reduce method and not using lambda expression.
I used old way: Interface interface = new InterfaceImpl();

3. See below Log output with sequential .stream() and Log output with .parallelStream()

From log, I did not see the output from combiner BinaryOperator ft (with code using .stream())
By chance/accident not by design   , I switch from sequential stream() to parallelStream(),
I saw the output from combiner BinaryOperator ft.

But with sequential stream(), the end result is correct: Got a list of 2 class name.
With parallelStream, the end result is NOT correct: Got a list of 2 class name appeared twice (i.e a list of 4)

How to make sense of this? Can some one provide an explanation for stream().reduce vs parallelStream().reduce output?

Code 1:

Stream<Class<?>> classStream = Arrays.asList(FunctionalInterface.class, Stream.class).stream();
List<String> identity =new ArrayList<>();
BiFunction<List<String>, Class<?>, List<String>> accumulator=new BiFunction<List<String>, Class<?>, List<String>>(){

	@Override
	public List<String> apply(List<String> t, Class<?> u) {
		System.out.println("=== accumulator called ====");

		System.out.println("Input Class<?> u=" + u.getName());
		System.out.println("Input List<String>=" + t);
		t.add(u.getName());
		System.out.println("Output List<String>=" + t);
		return t;
	}

};		
BinaryOperator<List<String>> combiner = new BinaryOperator<List<String>>(){
	@Override
	public List<String> apply(List<String> t, List<String> u) {
		System.out.println("=== combiner called ====");

		System.out.println("Input List<String>=" + t);
		System.out.println("Input List<String>=" + u);

		t.addAll(u);
		System.out.println("Output List<String>=" + t);				
		return t;
	}
};


List<String> toList= classStream.reduce(identity, accumulator, combiner);

Log output with sequential .stream()
		
Stream<Class<?>> classStream = Arrays.asList(FunctionalInterface.class, Stream.class).stream();
		
=== accumulator called ====
Input Class<?> u=java.lang.FunctionalInterface
Input List<String>=[]
Output List<String>=[java.lang.FunctionalInterface]
=== accumulator called ====
Input Class<?> u=java.util.stream.Stream
Input List<String>=[java.lang.FunctionalInterface]
Output List<String>=[java.lang.FunctionalInterface, java.util.stream.Stream]

Log output with .parallelStream()

Stream<Class<?>> classStream = Arrays.asList(FunctionalInterface.class, Stream.class).parallelStream();

=== accumulator called ====
Input Class<?> u=java.util.stream.Stream
Input List<String>=[]
Output List<String>=[java.util.stream.Stream]
=== accumulator called ====
Input Class<?> u=java.lang.FunctionalInterface
Input List<String>=[java.util.stream.Stream]
Output List<String>=[java.util.stream.Stream, java.lang.FunctionalInterface]
=== combiner called ====
Input List<String>=[java.util.stream.Stream, java.lang.FunctionalInterface]
Input List<String>=[java.util.stream.Stream, java.lang.FunctionalInterface]
Output List<String>=[java.util.stream.Stream, java.lang.FunctionalInterface, java.util.stream.Stream, java.lang.FunctionalInterface]

[H Paul]

1. Basically, I rewrite using lambdas expression for stream().reduce() and parallelStream().reduce().

The output is the same as I posted earlier using non-lambdas expression.

a. stream().reduce() output correct result but combiner is not called
b. parallelStream().reduce() did NOT output correctly but combiner is called.

2. If possible, can some check if the code for identity, accumulator and combiner are correct?

Thanks.

Code for stream().reduce() and parallelStream().reduce()

System.out.println("==================================== stream().reduce() ====");									

Arrays.asList(FunctionalInterface.class, Stream.class)
					.stream()
					.reduce(new ArrayList<String>(), //identity
							(List<String> t, Class<?> u) -> { //accumulator
								System.out.println("=== accumulator called ====");									
								t.add(u.getName());
								return t;
								},
							(List<String> t, List<String> u)->{ //combiner
								System.out.println("=== combiner called ====");
								t.addAll(u);
								return t;
							})
					.forEach(System.out::println);

System.out.println();									
System.out.println("==================================== parallelStream().reduce() ====");									

Arrays.asList(FunctionalInterface.class, Stream.class)
					.parallelStream()
					.reduce(new ArrayList<String>(), //identity
							(List<String> t, Class<?> u) -> { //accumulator
								System.out.println("=== accumulator called ====");									
								t.add(u.getName());
								return t;
								},
							(List<String> t, List<String> u)->{ //combiner
								System.out.println("=== combiner called ====");
								t.addAll(u);
								return t;
							})
					.forEach(System.out::println);

Log for stream().reduce() and parallelStream().reduce()


==================================== stream().reduce() ====
=== accumulator called ====
=== accumulator called ====
java.lang.FunctionalInterface
java.util.stream.Stream

==================================== parallelStream().reduce() ====
=== accumulator called ====
=== accumulator called ====
=== combiner called ====
java.util.stream.Stream
java.lang.FunctionalInterface
java.util.stream.Stream
java.lang.FunctionalInterface

[Stephan van Hulst]

The problem is caused because your accumulator and combiner functions are not stateless. The reduce() method requires stateless and associative functions. Stateless means that the functions are based on nothing else but the input arguments, and they cause no side-effects. Your accumulator calls List.add() and your combiner calls List.addAll(), both causing side-effects in the input arguments.

This usually doesn't cause problems in serial reductions, because the combiner is never called, and the accumulator is called once for each element in the stream. However, here's what happens in a parallel reduction:

The elements are split in two halves, which are both reduced in parallel.

The first half is reduced by adding FunctionalInterface.class to the identity list. Your identity list is now no longer empty.

The second half is reduced by adding Stream.class to the identity list. Because the previous reduction was stateful, the identity list is now [FunctionalInterface.class, Stream.class].

The reduction operation expects the results of your accumulator to be independent objects, so it now has two (supposedly independent) reduced halves which it has to combine. Essentially it performs identity.addAll(identity), which results in [FunctionalInterface.class, Stream.class, FunctionalInterface.class, Stream.class]

You can make the accumulator stateless by copying the input list, adding the element to that, and returning the new list. You can make the combiner stateless by copying the first input list, adding all elements of the second list to it, and returning the new list. Because copying the lists all the time can become very expensive, the Java designers added a mutable reduction operator: Stream.collect(). Stream.collect() does the same as Stream.reduce(), except it keeps in mind that the accumulation is reused by the different threads performing the reduction.

A handy guideline is to use Stream.reduce() if your result is an immutable type, and Stream.collect() if your result is a mutable type.

[Stephan van Hulst]

Here's the same operation using Stream.collect(), assuming we're NOT using Collectors.toList():

List<Class<?>> classes = Arrays.asList(FunctionalInterface.class, Stream.class);

List<String> classNames = classes.parallelStream().map(Class<?>::getName).collect(
  ArrayList<String>::new,
  List<String>::add,
  List<String>::addAll
);

System.out.println(classNames);

[H Paul]

To trouble shooting programmatically to explain output Logs that I posted earlier.

1) I use Thread information.
2) I use hashcode to see the List memory address. But this is NOT good.
3) I identityHashCode to see the List memory address. And this is GOOD! Help from

http://stackoverflow.com/questions/18396927/how-to-print-the-address-of-an-object-if-you-have-redefined-tostring-method

1 and 3 showed only 1 List passed around (one Sharedable List)

4)
See below code: I use three classes.
See below Log: I have 2 set of logs:
a. In 1 set of log: Luckly, I saw 2 different threads: ForkJoinPool.commonPool-worker-1 and main
b. In 2nd set of log: I saw the correct identityHashCode of List memory address.

Code: I use three classes.

List<String> identity=new ArrayList<String>();
System.out.println("=== identity thread: "
		+ Thread.currentThread().getName()
		+ " . identity List hashCode: " 
		+ identity.hashCode());		

//new ArrayList<Class<?>>()	
Arrays.asList(FunctionalInterface.class, Stream.class, Thread.class)
//.stream()	
.parallelStream()
.reduce(identity, //identity
		(List<String> t, Class<?> u) -> { //accumulator					
			System.out.println();
			System.out.println("=== accumulator entered by thread: "
					+ Thread.currentThread().getName()
					+ " . Input List t hashCode: " 
					+ t.hashCode()					
					+ " . Input List t: "
					+ t); 					
			t.add(u.getName());

			System.out.println("=== accumulator exit by thread: "
					+ Thread.currentThread().getName()
					+ " . Returned List t hashCode: " 
					+ t.hashCode()
					+ " . Returned List t: "
					+ t); 

			return t;
			},
		(List<String> t, List<String> u)->{ //combiner
			System.out.println();
			System.out.println("=== combiner entered by thread: "
					+ Thread.currentThread().getName()
					+ " . Input List t hashCode: " 
					+ t.hashCode()			
					+ " . Input List u hashCode: " 
					+ u.hashCode());	


			t.addAll(u);

			System.out.println("=== combiner exit by thread: "
					+ Thread.currentThread().getName()
					+ " . Returned List t hashCode: " 
					+ t.hashCode()
					+ " . Returned List t content: "
					+ t); 							
			return t;
		})
.forEach(System.out::println);

1st set of Log: No use of identityHashCode see the List memory address

=== identity thread: main . identity List hashCode: 1

=== accumulator entered by thread: ForkJoinPool.commonPool-worker-1 . Input List t hashCode: 1 . Input List t: []
=== accumulator exit by thread: ForkJoinPool.commonPool-worker-1 . Returned List t hashCode: 1081231553 . Returned List t: [java.lang.FunctionalInterface]

=== accumulator entered by thread: ForkJoinPool.commonPool-worker-1 . Input List t hashCode: 1081231553 . Input List t: [java.lang.FunctionalInterface]
=== accumulator exit by thread: ForkJoinPool.commonPool-worker-1 . Returned List t hashCode: 371241917 . Returned List t: [java.lang.FunctionalInterface, java.lang.Thread]

=== accumulator entered by thread: main . Input List t hashCode: 371241917 . Input List t: [java.lang.FunctionalInterface, java.lang.Thread]
=== accumulator exit by thread: main . Returned List t hashCode: 471008273 . Returned List t: [java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream]

=== combiner entered by thread: main . Input List t hashCode: 471008273 . Input List u hashCode: 471008273
=== combiner exit by thread: main . Returned List t hashCode: 620283393 . Returned List t content: [java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream, java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream]

=== combiner entered by thread: main . Input List t hashCode: 620283393 . Input List u hashCode: 620283393
=== combiner exit by thread: main . Returned List t hashCode: -1044050943 . Returned List t content: [java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream, java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream, java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream, java.lang.FunctionalInterface, java.lang.Thread, java.util.stream.Stream]
java.lang.FunctionalInterface
java.lang.Thread
java.util.stream.Stream
java.lang.FunctionalInterface
java.lang.Thread
java.util.stream.Stream
java.lang.FunctionalInterface
java.lang.Thread
java.util.stream.Stream
java.lang.FunctionalInterface
java.lang.Thread
java.util.stream.Stream

2nd set of Log: Use of identityHashCode see the List memory address: 7853278

=== identity thread: main . identity List hashCode: 1 and identityHashCode: 7853278

=== accumulator entered by thread: main . Input List t hashCode: 1 and identityHashCode: 7853278 . Input List t: []
=== accumulator exit by thread: main . Returned List t hashCode: 1847410765 and identityHashCode: 7853278 . Returned List t: [java.util.stream.Stream]

=== accumulator entered by thread: main . Input List t hashCode: 1847410765 and identityHashCode: 7853278 . Input List t: [java.util.stream.Stream]
=== accumulator exit by thread: main . Returned List t hashCode: -1647006287 and identityHashCode: 7853278 . Returned List t: [java.util.stream.Stream, java.lang.Thread]

=== combiner entered by thread: main . Input List t hashCode: -1647006287 and identityHashCode: 7853278 . Input List u hashCode: -1647006287 and identityHashCode: 7853278
=== combiner exit by thread: main . Returned List t hashCode: 422883169 and identityHashCode: 7853278 . Returned List t content: [java.util.stream.Stream, java.lang.Thread, java.util.stream.Stream, java.lang.Thread]

=== accumulator entered by thread: main . Input List t hashCode: 422883169 and identityHashCode: 7853278 . Input List t: [java.util.stream.Stream, java.lang.Thread, java.util.stream.Stream, java.lang.Thread]
=== accumulator exit by thread: main . Returned List t hashCode: 1305707873 and identityHashCode: 7853278 . Returned List t: [java.util.stream.Stream, java.lang.Thread, java.util.stream.Stream, java.lang.Thread, java.lang.FunctionalInterface]

=== combiner entered by thread: main . Input List t hashCode: 1305707873 and identityHashCode: 7853278 . Input List u hashCode: 1305707873 and identityHashCode: 7853278
=== combiner exit by thread: main . Returned List t hashCode: 1144806401 and identityHashCode: 7853278 . Returned List t content: [java.util.stream.Stream, java.lang.Thread, java.util.stream.Stream, java.lang.Thread, java.lang.FunctionalInterface, java.util.stream.Stream, java.lang.Thread, java.util.stream.Stream, java.lang.Thread, java.lang.FunctionalInterface]
java.util.stream.Stream
java.lang.Thread
java.util.stream.Stream
java.lang.Thread
java.lang.FunctionalInterface
java.util.stream.Stream
java.lang.Thread
java.util.stream.Stream
java.lang.Thread
java.lang.FunctionalInterface

[H Paul]

Master Stephan,

1. I read what your 1st post of lengthy explanation and wanted to give you a cow for that.  (Unless you tell me how, master.      

What you explained confirmed my trouble shooting code posted to see what is happening.

2. I will have look at your next "coded" post.

[Stephan van Hulst]

Ron McLeod gave me a cow in your name, thank you both

In turn, have a cow for your own investigations!

[H Paul]

0) Thank-you, Ron McLeod.  

1) Following Stephen's lengthy explanation: to make stream().parallelStream().reduce(...) work
See below code.
2) Following Stephen's lengthy explanation: to make stream()..parallelStream().collect(Collectors.reducing(...)) work
This is new. See below code.

3) How Imperative follows Functional declarative footstep (How OP sees Stephen's Functional declarative )
Reply to Stephen's posted code.    See below.


1) Following Stephen's lengthy explanation: to make stream().parallelStream().reduce(...) work

		public void testParallelStreamReduce() { 

			System.out.println("=== testParallelStreamReduce ====");

			Arrays.asList(FunctionalInterface.class, Stream.class)
			.parallelStream()
			.reduce(new ArrayList<String>(), //identity
					(List<String> t, Class<?> u) -> { //accumulator
						System.out.println("=== accumulator called ====");	
						//++++++++++=============== old code with Threading issues
						//t.add(u.getName());
						//return t;
						//========================= replaced with new code
						List<String> accumulatorList=new ArrayList<>(); //Use new List
						accumulatorList.addAll(t);										
						accumulatorList.add(u.getName());
						return accumulatorList;
						},
					(List<String> t, List<String> u)->{ //combiner
						System.out.println("=== combiner called ====");
						//++++++++++=============== old code with Threading issues
						//t.addAll(u);
						//return t;
						//========================= replaced with new code
						List<String> combinedList=new ArrayList<>(); //Use new List
						combinedList.addAll(t);
						combinedList.addAll(u);
						return combinedList;
					})
			.forEach(System.out::println);

		}

2) Following Stephen's lengthy explanation: to make stream()..parallelStream().collect(Collectors.reducing(...)) work

	public void testParallelStreamCollectReduceCollector() { 

		System.out.println("=== testParallelStreamCollectReduceCollector called ====");	


		Arrays.asList(FunctionalInterface.class, Stream.class).parallelStream()
		.collect(Collectors.reducing(
				new ArrayList<String>()
				, (Class<?> t) -> {				
					List<String> list = new ArrayList<String>();
					list.add(t.getName());
					return list;
					},
					(List<String> t, List<String> u)->{
							List<String> combinedList=new ArrayList<>(); //Use new List
							combinedList.addAll(t);
							combinedList.addAll(u);
							return combinedList;	
						}
				))
		.forEach(System.out::println);

	}	
	

3) How Imperative follows Functional declarative (How OP sees Stephen's Functional declarative )

	public void testParallelStreamMapCollect() { 

		System.out.println("=== testParallelStreamMapCollect called ====");	
		

		System.out.println("=== Imperative follows Functional declarative footstep ====");	
		
		Supplier<List<String>> supplier = ArrayList<String>::new;
		
		BiConsumer<List<String>, String> accumulator=new BiConsumer<List<String>, String>(){
			@Override
			public void accept(List<String> t, String u) {
				t.add(u);
			}};
	
		BiConsumer<List<String>, List<String>> combiner = new BiConsumer<List<String>, List<String>>() {
			@Override
			public void accept(List<String> t, List<String> u) {
				t.addAll(u);
			}};
	
			Arrays.asList(FunctionalInterface.class, Stream.class).parallelStream()
		.map(Class::getName)
		.collect(supplier, accumulator, combiner)
		.forEach(System.out::println);

		
		System.out.println("=== Imperative sees Functional declarative ====");	
		
		Arrays.asList(FunctionalInterface.class, Stream.class).parallelStream()
		.map(Class::getName)
		.collect(ArrayList<String>::new, 
				(List<String> t, String u)->t.add(u) 
				, (List<String> t, List<String> u)->t.addAll(u))
		.forEach(System.out::println);		
		
		System.out.println("=== OP sees Stephen's Functional declarative ====");	
				
		Arrays.asList(FunctionalInterface.class, Stream.class).parallelStream()
		.map(Class::getName)
		.collect(ArrayList<String>::new, List::add, List::addAll)
		.forEach(System.out::println);		
	}	

[Stephan van Hulst]

Note that using Stream.collect() in combination with a Collectors.reduce() just a roundabout way of calling Stream.reduce(). You're performing an immutable reduction which is not very efficient if you're working with a mutable result.

[H Paul]

Yes, Master! will follow your lengthy explanation on 2 points: performance/efficiency and readability.

(My examples are just a way to explore how things work. Thanks.)

[H Paul]

1. Back to the future: A question out of curiosity/notice (Not on performance nor readability nor on best practice)

From my  1st posted log and your explanation for sequential stream().reduce(...): the combiner is not called.

		Arrays.asList(FunctionalInterface.class, Stream.class)
		.stream()
		.reduce(new ArrayList<String>(), //identity
				(List<String> t, Class<?> u) -> { //accumulator
					System.out.println("accumulator");					
					t.add(u.getName());
					return t;

					},
				// null) //combiner ---> NullPointerException
				(t, u)->null) //combiner to make it happy?	
		.forEach(System.out::println);

2. Why does reduce required one (combiner)? See above code.

I just passed in null, it complained NullPointerException.
java.lang.NullPointerException
at java.util.Objects.requireNonNull(Objects.java:203)
at java.util.stream.ReduceOps.makeRef(ReduceOps.java:71)
at java.util.stream.ReferencePipeline.reduce(ReferencePipeline.java:484)

So to make it happy? :-D. I passed a FT combiner which has no meaning: simple return null.

[Stephan van Hulst]

The reduce() method doesn't know whether it's being called on a parallel stream or not. To prevent bugs from occurring somewhere later in the application when it tries to apply a null combiner to two partial reductions, it immediately throws an exception when you try to pass it null. You should strive to write programs the same way. If it doesn't make sense for your methods and constructors to accept null, they should forbid it.