current community

your communities

more stack exchange communities

Stack Exchange
tour help
Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.
up vote 14 down vote favorite
1

I'm trying to get more familiar with the new Java 8 features, so I am rewriting one of my earlier projects. It includes a class that keeps track of the best entry seen so far:

import java.util.List;

public class Maxinator<T> {

    private final QualityFunction<T> qualityFunction;
    private T best;
    private double bestQuality;

    public Maxinator(QualityFunction<T> qualityFunction) {
        this.qualityFunction = qualityFunction;
        reset();
    }

    public void reset() {
        best = null;
        bestQuality = Double.NEGATIVE_INFINITY;
    }

    public T getBest() {
        return best;
    }

    public void updateBest(List<T> population) {
        population.parallelStream()
                .forEach(i -> {
                    double quality = qualityFunction.computeQuality(i);
                    if (quality > bestQuality) {
                        best = i;
                        bestQuality = quality;
                    }
                });
    }
}

Where QualityFunction is simply the following interface:

public interface QualityFunction<T> {
    public abstract double computeQuality(T individual);
}

My questions:

  • Are there any concurrency issues with the code in the forEach modifying best and bestQuality? Or are these automatically taken care of by the parallel stream processing?
  • Is there a more idiomatic way to write this, using the new APIs (collect, reduce, etc.)?

Note: the quality function could be very expensive, so I want to make sure that it is only called once per element.

share|improve this question

2 Answers 2

up vote 10 down vote

Your code is not thread-safe. Each of the threads will, in parallel, be accessing both the best, and the bestQuality variables.

Your Lambda is, in essence, modifying external data from the stream, and this is an anti-pattern for streams. It has side-effects.

You should change your code to use the collect mechanism. There are a few ways to do it, but, you should look at this example for guidance: Reduction

Some Notes about my following suggestion:

  • I converted the method to a static method, and created an inner accumulator class.
  • you should possibly refine this answer to suit your needs more in the class structure.

Putting these observations together, I would suggest something like:

public class Maxinator {

    @FunctionalInterface
    public interface QualityFunction<V> {
        double computeQuality(V value);
    }

    private Maxinator() {
//        no public instances
    }

    private static final class AccumulateResult<T> {
        T bestItem = null;
        double bestScore = Double.MIN_VALUE;

        public T getBestItem() {
            return bestItem;
        }

        public void accept(QualityFunction<T> function, T item) {
            double score = function.computeQuality(item);
            if (score > bestScore) {
                bestScore = score;
                bestItem = item;
            }
        }

        public AccumulateResult<T> combine(AccumulateResult<T> r) {
            if (r.bestScore > bestScore) {
                bestScore = r.bestScore;
                bestItem = r.bestItem;
            }
            return this;
        }

    }

    public static <T> T getBest(final QualityFunction<T> qualityFunction, final List<T> population) {
        return population.parallelStream().collect(Collector.of(
                    AccumulateResult<T>::new,
                    (a,t) -> a.accept(qualityFunction, t),
                    (a, b) -> a.combine(b))
               ).getBestItem();
    }
}
share|improve this answer
    
Updated with a code example using an accumulator. –  rolfl♦ 20 hours ago
    
I knew there had to be a more natural way to fit this into the streams framework, and this is it. Thank you! I do have two questions: 1. Why did you go for Double.MIN_VALUE, instead of Double.NEGATIVE_INFINITY as starting value? 2. If I really want my class to keep track of the best so far (over multiple calls of updateBest), I assume I can use this same idea, but initialize the accumulator with the current bestItem and bestScore? –  Mangara 20 hours ago
    
Double.NEGATIVE_INFINITY would be better, but really, I should have a boolean first flag, and go from there. As for the keeping-the-old value - you should realize that when streaming the values, there will be many 'Result' instances, and they will all be merged as the parallel threads complete. –  rolfl♦ 18 hours ago
    
Will this result in a parallel computation? The docs suggest that this only happens if the Collector defines the Collector.Characteristics.CONCURRENT and Collector.Characteristics.UNORDERED characteristics. –  Mangara 18 hours ago
    
The Collector does not need to be concurrent for the stream to be parallel. The collector has an efficient 'combine' method, which makes the non-concurrent collector efficient still. Instead of collecting all the values in one concurrent collector, the parallel stream collects in to multiple collectors, and then safely merges the results. –  rolfl♦ 14 hours ago
up vote 6 down vote

http://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#max-java.util.Comparator-

Returns the maximum element of this stream according to the provided Comparator. This is a special case of a reduction.

Furthermore, there's a handy static method on Comparator to lift a regular function to a comparator:

http://docs.oracle.com/javase/8/docs/api/java/util/Comparator.html#comparing-java.util.function.Function-

Accepts a function that extracts a Comparable sort key from a type T, and returns a Comparator that compares by that sort key.

e.g.,

public static <T> T getBest(final Function<T, Double> qualityFunction, final List<T> population) {
    return population.parallelStream()
      .max(Comparator.comparing(qualityFunction))
      .get(); // unwrap Optional<T>
}

The java.util.function package has some nice stuff in it (though not quite as rich as, say, Scala or clojure), but you have to hunt through the javadocs for a lot of it.

share|improve this answer
    
This is a really nice solution! My only concern is the number of times qualityFunction is evaluated here. In my test, it was evaluated roughly twice for each object. As some quality functions in my project are really expensive, this would effectively double the running time. –  Mangara 18 hours ago
1  
Yep, unfortunately the implementation of max and Comparator.comparing are naive, and don't do any caching, so you'd be better off making your own Collector which caches the result. The docs on MutableReduction describe this tradeoff. –  blendmaster 18 hours ago
    
Note that there is also Collectors.maxBy(Comparator<T>), but unfortunately has the same problem as Stream#max in that its implementation defers to Collectors#reducing (i.e., Stream#reduce) and therefore can't generically cache the Comparator.comparing(qualityFunction). –  blendmaster 18 hours ago

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.