CODEWITHSUNDEEP
javajava-8java-streamreduce
gstackoverflow
gstackoverflow

Reputation: 37034

Reduce for parallel stream without combiner executes by several threads correctly. When should I use combiner at this case?

I have read following about:

https://stackoverflow.com/a/22814174/2674303

and I made resolution that combiner uses only in parallel stream for correct merge accumulator results. one accumulator instance on each thread.

Thus I made resolution that reduce without combiner will not work correctly.

To check this I have wrote following example:

   Person reduce = Person.getPersons().stream()
                .parallel() 
                .reduce(new Person(), (intermediateResult, p2) -> {
                    System.out.println(Thread.currentThread().getName());
                    return new Person("default", intermediateResult.getAge() + p2.getAge());
                });
        System.out.println(reduce);

model:

public class Person {

    String name;

    Integer age;
    ///...

    public static Collection<Person> getPersons() {
        List<Person> persons = new ArrayList<>();
        persons.add(new Person("Vasya", 12));
        persons.add(new Person("Petya", 32));
        persons.add(new Person("Serj", 10));
        persons.add(new Person("Onotole", 18));
        return persons;
   }
}

As you can see I don't provide combiner
sample output:

ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-1
Person{name='default', age=72}

I have executed application several times and always I see correct result.

Please, explain how does reduce working for parallel stream if combiner is not provided.

Upvotes: 3

Views: 751

Answers (3)

Misha
Misha

Reputation: 28133

3-argument reduce exists for a somewhat uncommon situation that looks like this:

Imagine that instead of reducing a stream of Persons into a Person, you had a different intermediate value like PopulationStats. 

class PopulationStats {
  // make new stats that includes this person
  PopulationStats addPerson(Person p) {
    return new PopulationStats(........);
  }

  // make new stats that combines this and other stats
  PopulationStats addStats(PopulationStats other) {
    return new PopulationStats(........);
  } 
}

In a case like that, the 3-argument reduce serves to avoid the intermediate step of making a PopulationStats for each Person before reducing.

PopulationStats stats = people.stream()
  .reduce(new PopulationStats(), PopulationStats::addPerson, PopulationStats::addStats);

Upvotes: 0

Holger
Holger

Reputation: 298123

You have specified a combiner. In this case, the combiner function is identical to your accumulator function.

This is always possible, if the result type is identical to your stream element type.

Compare with reduction by summing up the values, a+b+c+d might evaluated in parallel by calculating (a+b)+(c+d). Here, the accumulator is the addition, which is the same operation as the combiner function which processes the intermediate results of (a+b) and (c+d).

This example also shows that, unless there is a type conversion involved, it would be strange if you need a different combiner function as the associativity constraint of the accumulator function implies that it is normally sufficient as a combiner function. Keep in mind, that it should be irrelevant, whether the stream calculates, a+b+c+d, (a+b+c)+d, (a+b)+(c+d) or a+(b+c+d).

Upvotes: 4

Tagir Valeev
Tagir Valeev

Reputation: 100159

In this case your accumulator works as combiner as well. This is a shorthand when reduction type is the same as stream element type. Thus

myStream.reduce(identity, accumulator);

Is fully equivalent to

myStream.reduce(identity, accumulator, accumulator);

You can even check the source code of these methods in OpenJDK:

@Override
public final <R> R reduce(R identity, BiFunction<R, ? super P_OUT, R> accumulator, 
                          BinaryOperator<R> combiner) {
    return evaluate(ReduceOps.makeRef(identity, accumulator, combiner));
}

@Override
public final P_OUT reduce(final P_OUT identity, final BinaryOperator<P_OUT> accumulator) {
    return evaluate(ReduceOps.makeRef(identity, accumulator, accumulator));
}

The three-argument version is more flexible as the reduction operation may produce an object of another type. In this case you cannot use two-argument reduction, because you don't provide a rule how to combine two elements of the resulting type. However when resulting type is the same, accumulator and combiner work on the same object type, thus if it's associative, it should be just the same operation.

Upvotes: 6

Related Questions