changeset 7558:ab80ab9a7af2

Transform ReduceOp into a factory class ReduceOps.
author psandoz
date Mon, 04 Mar 2013 18:51:07 +0100
parents 18ea6bfc143c
children 3a9f4deac132
files src/share/classes/java/util/stream/DistinctOp.java src/share/classes/java/util/stream/DoublePipeline.java src/share/classes/java/util/stream/IntPipeline.java src/share/classes/java/util/stream/LongPipeline.java src/share/classes/java/util/stream/ReduceOp.java src/share/classes/java/util/stream/ReduceOps.java src/share/classes/java/util/stream/ReferencePipeline.java
diffstat 7 files changed, 721 insertions(+), 680 deletions(-) [+]
line wrap: on
line diff
--- a/src/share/classes/java/util/stream/DistinctOp.java	Mon Mar 04 12:10:03 2013 -0500
+++ b/src/share/classes/java/util/stream/DistinctOp.java	Mon Mar 04 18:51:07 2013 +0100
@@ -118,8 +118,8 @@
             // If the stream is SORTED then it should also be ORDERED so the following will also
             // preserve the sort order
             TerminalOp<T, LinkedHashSet<T>> reduceOp
-                    = ReduceOp.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,
-                                                            LinkedHashSet::addAll);
+                    = ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,
+                                                             LinkedHashSet::addAll);
             return Nodes.node(reduceOp.evaluateParallel(helper));
         }
         else {
--- a/src/share/classes/java/util/stream/DoublePipeline.java	Mon Mar 04 12:10:03 2013 -0500
+++ b/src/share/classes/java/util/stream/DoublePipeline.java	Mon Mar 04 18:51:07 2013 +0100
@@ -251,17 +251,17 @@
 
     @Override
     public double reduce(double identity, DoubleBinaryOperator op) {
-        return pipeline(ReduceOp.makeDouble(identity, op));
+        return pipeline(ReduceOps.makeDouble(identity, op));
     }
 
     @Override
     public OptionalDouble reduce(DoubleBinaryOperator op) {
-        return pipeline(ReduceOp.makeDouble(op));
+        return pipeline(ReduceOps.makeDouble(op));
     }
 
     @Override
     public <R> R collect(Collector.OfDouble<R> collector) {
-        return pipeline(ReduceOp.makeDouble(collector));
+        return pipeline(ReduceOps.makeDouble(collector));
     }
 
     @Override
--- a/src/share/classes/java/util/stream/IntPipeline.java	Mon Mar 04 12:10:03 2013 -0500
+++ b/src/share/classes/java/util/stream/IntPipeline.java	Mon Mar 04 18:51:07 2013 +0100
@@ -272,17 +272,17 @@
 
     @Override
     public int reduce(int identity, IntBinaryOperator op) {
-        return pipeline(ReduceOp.makeInt(identity, op));
+        return pipeline(ReduceOps.makeInt(identity, op));
     }
 
     @Override
     public OptionalInt reduce(IntBinaryOperator op) {
-        return pipeline(ReduceOp.makeInt(op));
+        return pipeline(ReduceOps.makeInt(op));
     }
 
     @Override
     public <R> R collect(Collector.OfInt<R> collector) {
-        return pipeline(ReduceOp.makeInt(collector));
+        return pipeline(ReduceOps.makeInt(collector));
     }
 
     @Override
--- a/src/share/classes/java/util/stream/LongPipeline.java	Mon Mar 04 12:10:03 2013 -0500
+++ b/src/share/classes/java/util/stream/LongPipeline.java	Mon Mar 04 18:51:07 2013 +0100
@@ -262,17 +262,17 @@
 
     @Override
     public long reduce(long identity, LongBinaryOperator op) {
-        return pipeline(ReduceOp.makeLong(identity, op));
+        return pipeline(ReduceOps.makeLong(identity, op));
     }
 
     @Override
     public OptionalLong reduce(LongBinaryOperator op) {
-        return pipeline(ReduceOp.makeLong(op));
+        return pipeline(ReduceOps.makeLong(op));
     }
 
     @Override
     public <R> R collect(Collector.OfLong<R> collector) {
-        return pipeline(ReduceOp.makeLong(collector));
+        return pipeline(ReduceOps.makeLong(collector));
     }
 
     @Override
--- a/src/share/classes/java/util/stream/ReduceOp.java	Mon Mar 04 12:10:03 2013 -0500
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,664 +0,0 @@
-/*
- * Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.
- * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
- *
- * This code is free software; you can redistribute it and/or modify it
- * under the terms of the GNU General Public License version 2 only, as
- * published by the Free Software Foundation.  Oracle designates this
- * particular file as subject to the "Classpath" exception as provided
- * by Oracle in the LICENSE file that accompanied this code.
- *
- * This code is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
- * version 2 for more details (a copy is included in the LICENSE file that
- * accompanied this code).
- *
- * You should have received a copy of the GNU General Public License version
- * 2 along with this work; if not, write to the Free Software Foundation,
- * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
- *
- * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
- * or visit www.oracle.com if you need additional information or have any
- * questions.
- */
-package java.util.stream;
-
-import java.util.Optional;
-import java.util.OptionalDouble;
-import java.util.OptionalInt;
-import java.util.OptionalLong;
-import java.util.Spliterator;
-import java.util.concurrent.CountedCompleter;
-import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
-import java.util.function.BinaryOperator;
-import java.util.function.DoubleBinaryOperator;
-import java.util.function.IntBinaryOperator;
-import java.util.function.LongBinaryOperator;
-import java.util.function.ObjDoubleConsumer;
-import java.util.function.ObjIntConsumer;
-import java.util.function.ObjLongConsumer;
-import java.util.function.Supplier;
-
-/**
- * A {@code TerminalOp} that evaluates a stream pipeline and sends the output into an {@code AccumulatingSink},
- * which performs a reduce operation. The {@code AccumulatingSink} must represent an associative reducing operation.
- *
- * @param <T> The output type of the stream pipeline
- * @param <R> The result type of the reducing operation
- * @param <S> The type of the {@code AccumulatingSink}
- * @since 1.8
- */
-// @@@ Can ReduceOp.AccumulatingSink be removed from the class type signature?
-abstract class ReduceOp<T, R, S extends ReduceOp.AccumulatingSink<T, R, S>> implements TerminalOp<T, R> {
-    private final StreamShape inputShape;
-
-    /**
-     * A type of {@code TerminalSink} that implements an associative reducing operation on elements of type
-     * {@code T} and producing a result of type {@code R}.
-     *
-     * @param <T> The type of input element to the combining operation
-     * @param <R> The result type
-     * @param <K> The type of the {@code AccumulatingSink}.
-     */
-    static interface AccumulatingSink<T, R, K extends AccumulatingSink<T, R, K>> extends TerminalSink<T, R> {
-        public void combine(K other);
-    }
-
-    /**
-     * Create a {@code ReduceOp} of the specified stream shape which uses the specified {@code Supplier} to create
-     * accumulating sinks
-     * @param shape The shape of the stream pipeline
-     */
-    private ReduceOp(StreamShape shape) {
-        inputShape = shape;
-    }
-
-    public abstract S makeSink();
-
-    @Override
-    public StreamShape inputShape() {
-        return inputShape;
-    }
-
-    @Override
-    public <P_IN> R evaluateSequential(PipelineHelper<P_IN, T> helper) {
-        return helper.into(makeSink(), helper.sourceSpliterator()).get();
-    }
-
-    @Override
-    public <P_IN> R evaluateParallel(PipelineHelper<P_IN, T> helper) {
-        return new ReduceTask<>(helper, this).invoke().get();
-    }
-
-    /**
-     * State box for a single state element, used as a base class for {@code AccumulatingSink} instances
-     * @param <U> The type of the state element
-     */
-    private static abstract class Box<U> {
-        U state;
-
-        Box() {} // Avoid creation of special accessor
-
-        public U get() {
-            return state;
-        }
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on reference values
-     * @param seed The identity element for the reduction
-     * @param reducer The accumulating function that incorporates an additional input element into the result
-     * @param combiner The combining function that combines two intermediate results
-     * @param <T> The type of the input elements
-     * @param <U> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static<T, U> TerminalOp<T, U>
-    makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
-        class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
-            @Override
-            public void begin(long size) {
-                state = seed;
-            }
-
-            @Override
-            public void accept(T t) {
-                state = reducer.apply(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                state = combiner.apply(state, other.state);
-            }
-        }
-        return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on reference values
-     * producing an optional reference result
-     * @param operator The reducing function
-     * @param <T> The type of the input elements, and the type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static<T> TerminalOp<T, Optional<T>>
-    makeRef(BinaryOperator<T> operator) {
-        class ReducingSink implements AccumulatingSink<T, Optional<T>, ReducingSink> {
-            private boolean empty;
-            private T state;
-
-            public void begin(long size) {
-                empty = true;
-                state = null;
-            }
-
-            @Override
-            public void accept(T t) {
-                if (empty) {
-                    empty = false;
-                    state = t;
-                } else {
-                    state = operator.apply(state, t);
-                }
-            }
-
-            @Override
-            public Optional<T> get() {
-                return empty ? Optional.empty() : Optional.of(state);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                if (!other.empty)
-                    accept(other.state);
-            }
-        }
-        return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a mutable reduce on reference values
-     * @param collector A {@code Collector} defining the reduction
-     * @param <T> The type of the input elements
-     * @param <R> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static<T,R> TerminalOp<T, R>
-    makeRef(Collector<? super T,R> collector) {
-        Supplier<R> supplier = collector.resultSupplier();
-        BiConsumer<R, ? super T> accumulator = collector.accumulator();
-        BinaryOperator<R> combiner = collector.combiner();
-        class ReducingSink extends Box<R> implements AccumulatingSink<T, R, ReducingSink> {
-            @Override
-            public void begin(long size) {
-                state = supplier.get();
-            }
-
-            @Override
-            public void accept(T t) {
-                accumulator.accept(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                state = combiner.apply(state, other.state);
-            }
-        }
-        return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a mutable reduce on reference values
-     * @param seedFactory A factory to produce a new base accumulator
-     * @param accumulator A function to incorporate an element into an accumulator
-     * @param reducer A function to combine an accumulator into another
-     * @param <T> The type of the input elements
-     * @param <R> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static<T, R> TerminalOp<T, R>
-    makeRef(Supplier<R> seedFactory, BiConsumer<R, ? super T> accumulator, BiConsumer<R,R> reducer) {
-        class ReducingSink extends Box<R> implements AccumulatingSink<T, R, ReducingSink> {
-            @Override
-            public void begin(long size) {
-                state = seedFactory.get();
-            }
-
-            @Override
-            public void accept(T t) {
-                accumulator.accept(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                reducer.accept(state, other.state);
-            }
-        }
-        return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on integer values
-     * @param identity The identity for the combining function
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Integer, Integer>
-    makeInt(int identity, IntBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Integer, Integer, ReducingSink>, Sink.OfInt {
-            private int state;
-
-            @Override
-            public void begin(long size) {
-                state = identity;
-            }
-
-            @Override
-            public void accept(int t) {
-                state = operator.applyAsInt(state, t);
-            }
-
-            @Override
-            public Integer get() {
-                return state;
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                accept(other.state);
-            }
-        }
-        return new ReduceOp<Integer, Integer, ReducingSink>(StreamShape.INT_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on integer values,
-     * producing an optional integer result
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Integer, OptionalInt>
-    makeInt(IntBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Integer, OptionalInt, ReducingSink>, Sink.OfInt {
-            private boolean empty;
-            private int state;
-
-            public void begin(long size) {
-                empty = true;
-                state = 0;
-            }
-
-            @Override
-            public void accept(int t) {
-                if (empty) {
-                    empty = false;
-                    state = t;
-                }
-                else {
-                    state = operator.applyAsInt(state, t);
-                }
-            }
-
-            @Override
-            public OptionalInt get() {
-                return empty ? OptionalInt.empty() : OptionalInt.of(state);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                if (!other.empty)
-                    accept(other.state);
-            }
-        }
-        return new ReduceOp<Integer, OptionalInt, ReducingSink>(StreamShape.INT_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a mutable reduce on integer values
-     * @param collector A {@code Collector} defining the reduction
-     * @param <R> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static <R> TerminalOp<Integer, R>
-    makeInt(Collector.OfInt<R> collector) {
-        Supplier<R> supplier = collector.resultSupplier();
-        ObjIntConsumer<R> accumulator = collector.intAccumulator();
-        BinaryOperator<R> combiner = collector.combiner();
-        class ReducingSink extends Box<R> implements AccumulatingSink<Integer, R, ReducingSink>, Sink.OfInt {
-            @Override
-            public void begin(long size) {
-                state = supplier.get();
-            }
-
-            @Override
-            public void accept(int t) {
-                accumulator.accept(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                state = combiner.apply(state, other.state);
-            }
-        }
-        return new ReduceOp<Integer, R, ReducingSink>(StreamShape.INT_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on long values
-     * @param identity The identity for the combining function
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Long, Long>
-    makeLong(long identity, LongBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Long, Long, ReducingSink>, Sink.OfLong {
-            private long state;
-
-            @Override
-            public void begin(long size) {
-                state = identity;
-            }
-
-            @Override
-            public void accept(long t) {
-                state = operator.applyAsLong(state, t);
-            }
-
-            @Override
-            public Long get() {
-                return state;
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                accept(other.state);
-            }
-        }
-        return new ReduceOp<Long, Long, ReducingSink>(StreamShape.LONG_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on long values, producing an optional long result
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Long, OptionalLong>
-    makeLong(LongBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Long, OptionalLong, ReducingSink>, Sink.OfLong {
-            private boolean empty;
-            private long state;
-
-            public void begin(long size) {
-                empty = true;
-                state = 0;
-            }
-
-            @Override
-            public void accept(long t) {
-                if (empty) {
-                    empty = false;
-                    state = t;
-                }
-                else {
-                    state = operator.applyAsLong(state, t);
-                }
-            }
-
-            @Override
-            public OptionalLong get() {
-                return empty ? OptionalLong.empty() : OptionalLong.of(state);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                if (!other.empty)
-                    accept(other.state);
-            }
-        }
-        return new ReduceOp<Long, OptionalLong, ReducingSink>(StreamShape.LONG_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a mutable reduce on long values
-     * @param collector A {@code Collector} defining the reduction
-     * @param <R> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static <R> TerminalOp<Long, R>
-    makeLong(Collector.OfLong<R> collector) {
-        Supplier<R> supplier = collector.resultSupplier();
-        ObjLongConsumer<R> accumulator = collector.longAccumulator();
-        BinaryOperator<R> combiner = collector.combiner();
-        class ReducingSink extends Box<R> implements AccumulatingSink<Long, R, ReducingSink>, Sink.OfLong {
-            @Override
-            public void begin(long size) {
-                state = supplier.get();
-            }
-
-            @Override
-            public void accept(long t) {
-                accumulator.accept(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                state = combiner.apply(state, other.state);
-            }
-        }
-        return new ReduceOp<Long, R, ReducingSink>(StreamShape.LONG_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on double values
-     * @param identity The identity for the combining function
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Double, Double>
-    makeDouble(double identity, DoubleBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Double, Double, ReducingSink>, Sink.OfDouble {
-            private double state;
-
-            @Override
-            public void begin(long size) {
-                state = identity;
-            }
-
-            @Override
-            public void accept(double t) {
-                state = operator.applyAsDouble(state, t);
-            }
-
-            @Override
-            public Double get() {
-                return state;
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                accept(other.state);
-            }
-        }
-        return new ReduceOp<Double, Double, ReducingSink>(StreamShape.DOUBLE_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a functional reduce on double values,
-     * producing an optional double result
-     * @param operator The combining function
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static TerminalOp<Double, OptionalDouble>
-    makeDouble(DoubleBinaryOperator operator) {
-        class ReducingSink implements AccumulatingSink<Double, OptionalDouble, ReducingSink>, Sink.OfDouble {
-            private boolean empty;
-            private double state;
-
-            public void begin(long size) {
-                empty = true;
-                state = 0;
-            }
-
-            @Override
-            public void accept(double t) {
-                if (empty) {
-                    empty = false;
-                    state = t;
-                }
-                else {
-                    state = operator.applyAsDouble(state, t);
-                }
-            }
-
-            @Override
-            public OptionalDouble get() {
-                return empty ? OptionalDouble.empty() : OptionalDouble.of(state);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                if (!other.empty)
-                    accept(other.state);
-            }
-        }
-        return new ReduceOp<Double, OptionalDouble, ReducingSink>(StreamShape.DOUBLE_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /**
-     * Construct a {@code ReduceOp} that implements a mutable reduce on double values
-     * @param collector A {@code Collector} defining the reduction
-     * @param <R> The type of the result
-     * @return A {@code ReduceOp} implementing the reduction
-     */
-    public static <R> TerminalOp<Double, R> makeDouble(Collector.OfDouble<R> collector) {
-        Supplier<R> supplier = collector.resultSupplier();
-        ObjDoubleConsumer<R> accumulator = collector.doubleAccumulator();
-        BinaryOperator<R> combiner = collector.combiner();
-        class ReducingSink extends Box<R> implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
-            @Override
-            public void begin(long size) {
-                state = supplier.get();
-            }
-
-            @Override
-            public void accept(double t) {
-                accumulator.accept(state, t);
-            }
-
-            @Override
-            public void combine(ReducingSink other) {
-                state = combiner.apply(state, other.state);
-            }
-        }
-        return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
-            @Override
-            public ReducingSink makeSink() {
-                return new ReducingSink();
-            }
-        };
-    }
-
-    /** A {@code ForkJoinTask} for performing a parallel reduce operation */
-    private static final class ReduceTask<P_IN, P_OUT, R, S extends AccumulatingSink<P_OUT, R, S>>
-            extends AbstractTask<P_IN, P_OUT, S, ReduceTask<P_IN, P_OUT, R, S>> {
-        private final ReduceOp<P_OUT, R, S> op;
-
-        ReduceTask(PipelineHelper<P_IN, P_OUT> helper, ReduceOp<P_OUT, R, S> op) {
-            super(helper);
-            this.op = op;
-        }
-
-        ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent, Spliterator<P_IN> spliterator) {
-            super(parent, spliterator);
-            this.op = parent.op;
-        }
-
-        @Override
-        protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator<P_IN> spliterator) {
-            return new ReduceTask<>(this, spliterator);
-        }
-
-        @Override
-        protected S doLeaf() {
-            return helper.into(op.makeSink(), spliterator);
-        }
-
-        @Override
-        public void onCompletion(CountedCompleter caller) {
-            if (!isLeaf()) {
-                ReduceTask<P_IN, P_OUT, R, S> child = children;
-                S result = child.getLocalResult();
-                child = child.nextSibling;
-                for (; child != null; child = child.nextSibling) {
-                    S otherResult = child.getLocalResult();
-                    result.combine(otherResult);
-                    child.setLocalResult(null); // GC otherResult
-                }
-                setLocalResult(result);
-            }
-            super.onCompletion(caller);
-        }
-    }
-}
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/share/classes/java/util/stream/ReduceOps.java	Mon Mar 04 18:51:07 2013 +0100
@@ -0,0 +1,705 @@
+/*
+ * Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package java.util.stream;
+
+import java.util.Optional;
+import java.util.OptionalDouble;
+import java.util.OptionalInt;
+import java.util.OptionalLong;
+import java.util.Spliterator;
+import java.util.concurrent.CountedCompleter;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.BinaryOperator;
+import java.util.function.DoubleBinaryOperator;
+import java.util.function.IntBinaryOperator;
+import java.util.function.LongBinaryOperator;
+import java.util.function.ObjDoubleConsumer;
+import java.util.function.ObjIntConsumer;
+import java.util.function.ObjLongConsumer;
+import java.util.function.Supplier;
+
+/**
+ * A factory for the creating instances of {@code TerminalOp) that implement
+ * reductions.
+ *
+ * @since 1.8
+ */
+final class ReduceOps {
+
+    private ReduceOps() { }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * reference values
+     *
+     * @param seed The identity element for the reduction
+     * @param reducer The accumulating function that incorporates an additional
+     *        input element into the result
+     * @param combiner The combining function that combines two intermediate
+     *        results
+     * @param <T> The type of the input elements
+     * @param <U> The type of the result
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static<T, U> TerminalOp<T, U>
+    makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
+        class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
+            @Override
+            public void begin(long size) {
+                state = seed;
+            }
+
+            @Override
+            public void accept(T t) {
+                state = reducer.apply(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                state = combiner.apply(state, other.state);
+            }
+        }
+        return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * reference values producing an optional reference result
+     *
+     * @param operator The reducing function
+     * @param <T> The type of the input elements, and the type of the result
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static<T> TerminalOp<T, Optional<T>>
+    makeRef(BinaryOperator<T> operator) {
+        class ReducingSink implements AccumulatingSink<T, Optional<T>, ReducingSink> {
+            private boolean empty;
+            private T state;
+
+            public void begin(long size) {
+                empty = true;
+                state = null;
+            }
+
+            @Override
+            public void accept(T t) {
+                if (empty) {
+                    empty = false;
+                    state = t;
+                } else {
+                    state = operator.apply(state, t);
+                }
+            }
+
+            @Override
+            public Optional<T> get() {
+                return empty ? Optional.empty() : Optional.of(state);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                if (!other.empty)
+                    accept(other.state);
+            }
+        }
+        return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a mutable reduce on
+     * reference values
+     *
+     * @param collector A {@code Collector} defining the reduction
+     * @param <T> The type of the input elements
+     * @param <R> The type of the result
+     * @return A {@code ReduceOp} implementing the reduction
+     */
+    public static<T,R> TerminalOp<T, R>
+    makeRef(Collector<? super T,R> collector) {
+        Supplier<R> supplier = collector.resultSupplier();
+        BiConsumer<R, ? super T> accumulator = collector.accumulator();
+        BinaryOperator<R> combiner = collector.combiner();
+        class ReducingSink extends Box<R> implements AccumulatingSink<T, R, ReducingSink> {
+            @Override
+            public void begin(long size) {
+                state = supplier.get();
+            }
+
+            @Override
+            public void accept(T t) {
+                accumulator.accept(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                state = combiner.apply(state, other.state);
+            }
+        }
+        return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a mutable reduce on
+     * reference values
+     *
+     * @param seedFactory A factory to produce a new base accumulator
+     * @param accumulator A function to incorporate an element into an
+     *        accumulator
+     * @param reducer A function to combine an accumulator into another
+     * @param <T> The type of the input elements
+     * @param <R> The type of the result
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static<T, R> TerminalOp<T, R>
+    makeRef(Supplier<R> seedFactory, BiConsumer<R, ? super T> accumulator, BiConsumer<R,R> reducer) {
+        class ReducingSink extends Box<R> implements AccumulatingSink<T, R, ReducingSink> {
+            @Override
+            public void begin(long size) {
+                state = seedFactory.get();
+            }
+
+            @Override
+            public void accept(T t) {
+                accumulator.accept(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                reducer.accept(state, other.state);
+            }
+        }
+        return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code int} values
+     *
+     * @param identity The identity for the combining function
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Integer, Integer>
+    makeInt(int identity, IntBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Integer, Integer, ReducingSink>, Sink.OfInt {
+            private int state;
+
+            @Override
+            public void begin(long size) {
+                state = identity;
+            }
+
+            @Override
+            public void accept(int t) {
+                state = operator.applyAsInt(state, t);
+            }
+
+            @Override
+            public Integer get() {
+                return state;
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                accept(other.state);
+            }
+        }
+        return new ReduceOp<Integer, Integer, ReducingSink>(StreamShape.INT_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code int} values, producing an optional integer result
+     *
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Integer, OptionalInt>
+    makeInt(IntBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Integer, OptionalInt, ReducingSink>, Sink.OfInt {
+            private boolean empty;
+            private int state;
+
+            public void begin(long size) {
+                empty = true;
+                state = 0;
+            }
+
+            @Override
+            public void accept(int t) {
+                if (empty) {
+                    empty = false;
+                    state = t;
+                }
+                else {
+                    state = operator.applyAsInt(state, t);
+                }
+            }
+
+            @Override
+            public OptionalInt get() {
+                return empty ? OptionalInt.empty() : OptionalInt.of(state);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                if (!other.empty)
+                    accept(other.state);
+            }
+        }
+        return new ReduceOp<Integer, OptionalInt, ReducingSink>(StreamShape.INT_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a mutable reduce on
+     * {@code int} values
+     *
+     * @param collector A {@code Collector} defining the reduction
+     * @param <R> The type of the result
+     * @return A {@code ReduceOp} implementing the reduction
+     */
+    public static <R> TerminalOp<Integer, R>
+    makeInt(Collector.OfInt<R> collector) {
+        Supplier<R> supplier = collector.resultSupplier();
+        ObjIntConsumer<R> accumulator = collector.intAccumulator();
+        BinaryOperator<R> combiner = collector.combiner();
+        class ReducingSink extends Box<R> implements AccumulatingSink<Integer, R, ReducingSink>, Sink.OfInt {
+            @Override
+            public void begin(long size) {
+                state = supplier.get();
+            }
+
+            @Override
+            public void accept(int t) {
+                accumulator.accept(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                state = combiner.apply(state, other.state);
+            }
+        }
+        return new ReduceOp<Integer, R, ReducingSink>(StreamShape.INT_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code long} values
+     *
+     * @param identity The identity for the combining function
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Long, Long>
+    makeLong(long identity, LongBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Long, Long, ReducingSink>, Sink.OfLong {
+            private long state;
+
+            @Override
+            public void begin(long size) {
+                state = identity;
+            }
+
+            @Override
+            public void accept(long t) {
+                state = operator.applyAsLong(state, t);
+            }
+
+            @Override
+            public Long get() {
+                return state;
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                accept(other.state);
+            }
+        }
+        return new ReduceOp<Long, Long, ReducingSink>(StreamShape.LONG_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code long} values, producing an optional long result
+     *
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Long, OptionalLong>
+    makeLong(LongBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Long, OptionalLong, ReducingSink>, Sink.OfLong {
+            private boolean empty;
+            private long state;
+
+            public void begin(long size) {
+                empty = true;
+                state = 0;
+            }
+
+            @Override
+            public void accept(long t) {
+                if (empty) {
+                    empty = false;
+                    state = t;
+                }
+                else {
+                    state = operator.applyAsLong(state, t);
+                }
+            }
+
+            @Override
+            public OptionalLong get() {
+                return empty ? OptionalLong.empty() : OptionalLong.of(state);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                if (!other.empty)
+                    accept(other.state);
+            }
+        }
+        return new ReduceOp<Long, OptionalLong, ReducingSink>(StreamShape.LONG_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a mutable reduce on
+     * {@code long} values
+     *
+     * @param collector A {@code Collector} defining the reduction
+     * @param <R> The type of the result
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static <R> TerminalOp<Long, R>
+    makeLong(Collector.OfLong<R> collector) {
+        Supplier<R> supplier = collector.resultSupplier();
+        ObjLongConsumer<R> accumulator = collector.longAccumulator();
+        BinaryOperator<R> combiner = collector.combiner();
+        class ReducingSink extends Box<R> implements AccumulatingSink<Long, R, ReducingSink>, Sink.OfLong {
+            @Override
+            public void begin(long size) {
+                state = supplier.get();
+            }
+
+            @Override
+            public void accept(long t) {
+                accumulator.accept(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                state = combiner.apply(state, other.state);
+            }
+        }
+        return new ReduceOp<Long, R, ReducingSink>(StreamShape.LONG_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code double} values
+     *
+     * @param identity The identity for the combining function
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Double, Double>
+    makeDouble(double identity, DoubleBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Double, Double, ReducingSink>, Sink.OfDouble {
+            private double state;
+
+            @Override
+            public void begin(long size) {
+                state = identity;
+            }
+
+            @Override
+            public void accept(double t) {
+                state = operator.applyAsDouble(state, t);
+            }
+
+            @Override
+            public Double get() {
+                return state;
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                accept(other.state);
+            }
+        }
+        return new ReduceOp<Double, Double, ReducingSink>(StreamShape.DOUBLE_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a functional reduce on
+     * {@code double} values, producing an optional double result
+     *
+     * @param operator The combining function
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static TerminalOp<Double, OptionalDouble>
+    makeDouble(DoubleBinaryOperator operator) {
+        class ReducingSink implements AccumulatingSink<Double, OptionalDouble, ReducingSink>, Sink.OfDouble {
+            private boolean empty;
+            private double state;
+
+            public void begin(long size) {
+                empty = true;
+                state = 0;
+            }
+
+            @Override
+            public void accept(double t) {
+                if (empty) {
+                    empty = false;
+                    state = t;
+                }
+                else {
+                    state = operator.applyAsDouble(state, t);
+                }
+            }
+
+            @Override
+            public OptionalDouble get() {
+                return empty ? OptionalDouble.empty() : OptionalDouble.of(state);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                if (!other.empty)
+                    accept(other.state);
+            }
+        }
+        return new ReduceOp<Double, OptionalDouble, ReducingSink>(StreamShape.DOUBLE_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * Constructs a {@code TerminalOp} that implements a mutable reduce on
+     * {@code double} values
+     *
+     * @param collector A {@code Collector} defining the reduction
+     * @param <R> The type of the result
+     * @return A {@code TerminalOp} implementing the reduction
+     */
+    public static <R> TerminalOp<Double, R> makeDouble(Collector.OfDouble<R> collector) {
+        Supplier<R> supplier = collector.resultSupplier();
+        ObjDoubleConsumer<R> accumulator = collector.doubleAccumulator();
+        BinaryOperator<R> combiner = collector.combiner();
+        class ReducingSink extends Box<R> implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
+            @Override
+            public void begin(long size) {
+                state = supplier.get();
+            }
+
+            @Override
+            public void accept(double t) {
+                accumulator.accept(state, t);
+            }
+
+            @Override
+            public void combine(ReducingSink other) {
+                state = combiner.apply(state, other.state);
+            }
+        }
+        return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
+            @Override
+            public ReducingSink makeSink() {
+                return new ReducingSink();
+            }
+        };
+    }
+
+    /**
+     * A type of {@code TerminalSink} that implements an associative reducing
+     * operation on elements of type {@code T} and producing a result of type
+     * {@code R}.
+     *
+     * @param <T> The type of input element to the combining operation
+     * @param <R> The result type
+     * @param <K> The type of the {@code AccumulatingSink}.
+     */
+    private static interface AccumulatingSink<T, R, K extends AccumulatingSink<T, R, K>> extends TerminalSink<T, R> {
+        public void combine(K other);
+    }
+
+    /**
+     * State box for a single state element, used as a base class for
+     * {@code AccumulatingSink} instances
+     *
+     * @param <U> The type of the state element
+     */
+    private static abstract class Box<U> {
+        U state;
+
+        Box() {} // Avoid creation of special accessor
+
+        public U get() {
+            return state;
+        }
+    }
+
+    /**
+     * A {@code TerminalOp} that evaluates a stream pipeline and sends the
+     * output into an {@code AccumulatingSink}, which performs a reduce
+     * operation. The {@code AccumulatingSink} must represent an associative
+     * reducing operation.
+     *
+     * @param <T> The output type of the stream pipeline
+     * @param <R> The result type of the reducing operation
+     * @param <S> The type of the {@code AccumulatingSink}
+     */
+    static abstract class ReduceOp<T, R, S extends AccumulatingSink<T, R, S>> implements TerminalOp<T, R> {
+        private final StreamShape inputShape;
+
+        /**
+         * Create a {@code ReduceOp} of the specified stream shape which uses
+         * the specified {@code Supplier} to create accumulating sinks
+         *
+         * @param shape The shape of the stream pipeline
+         */
+        ReduceOp(StreamShape shape) {
+            inputShape = shape;
+        }
+
+        public abstract S makeSink();
+
+        @Override
+        public StreamShape inputShape() {
+            return inputShape;
+        }
+
+        @Override
+        public <P_IN> R evaluateSequential(PipelineHelper<P_IN, T> helper) {
+            return helper.into(makeSink(), helper.sourceSpliterator()).get();
+        }
+
+        @Override
+        public <P_IN> R evaluateParallel(PipelineHelper<P_IN, T> helper) {
+            return new ReduceTask<>(helper, this).invoke().get();
+        }
+    }
+
+    /** A {@code ForkJoinTask} for performing a parallel reduce operation */
+    private static final class ReduceTask<P_IN, P_OUT, R, S extends AccumulatingSink<P_OUT, R, S>>
+            extends AbstractTask<P_IN, P_OUT, S, ReduceTask<P_IN, P_OUT, R, S>> {
+        private final ReduceOp<P_OUT, R, S> op;
+
+        ReduceTask(PipelineHelper<P_IN, P_OUT> helper, ReduceOp<P_OUT, R, S> op) {
+            super(helper);
+            this.op = op;
+        }
+
+        ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent, Spliterator<P_IN> spliterator) {
+            super(parent, spliterator);
+            this.op = parent.op;
+        }
+
+        @Override
+        protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator<P_IN> spliterator) {
+            return new ReduceTask<>(this, spliterator);
+        }
+
+        @Override
+        protected S doLeaf() {
+            return helper.into(op.makeSink(), spliterator);
+        }
+
+        @Override
+        public void onCompletion(CountedCompleter caller) {
+            if (!isLeaf()) {
+                ReduceTask<P_IN, P_OUT, R, S> child = children;
+                S result = child.getLocalResult();
+                child = child.nextSibling;
+                for (; child != null; child = child.nextSibling) {
+                    S otherResult = child.getLocalResult();
+                    result.combine(otherResult);
+                    child.setLocalResult(null); // GC otherResult
+                }
+                setLocalResult(result);
+            }
+            super.onCompletion(caller);
+        }
+    }
+}
--- a/src/share/classes/java/util/stream/ReferencePipeline.java	Mon Mar 04 12:10:03 2013 -0500
+++ b/src/share/classes/java/util/stream/ReferencePipeline.java	Mon Mar 04 18:51:07 2013 +0100
@@ -340,27 +340,27 @@
 
     @Override
     public U reduce(final U identity, final BinaryOperator<U> reducer) {
-        return pipeline(ReduceOp.makeRef(identity, reducer, reducer));
+        return pipeline(ReduceOps.makeRef(identity, reducer, reducer));
     }
 
     @Override
     public Optional<U> reduce(BinaryOperator<U> reducer) {
-        return pipeline(ReduceOp.makeRef(reducer));
+        return pipeline(ReduceOps.makeRef(reducer));
     }
 
     @Override
     public <R> R reduce(R identity, BiFunction<R, ? super U, R> reducer, BinaryOperator<R> combiner) {
-        return pipeline(ReduceOp.makeRef(identity, reducer, combiner));
+        return pipeline(ReduceOps.makeRef(identity, reducer, combiner));
     }
 
     @Override
     public <R> R collect(Collector<? super U, R> collector) {
-        return pipeline(ReduceOp.makeRef(collector));
+        return pipeline(ReduceOps.makeRef(collector));
     }
 
     @Override
     public <R> R collect(Supplier<R> resultFactory, BiConsumer<R, ? super U> accumulator, BiConsumer<R, R> combiner) {
-        return pipeline(ReduceOp.makeRef(resultFactory, accumulator, combiner));
+        return pipeline(ReduceOps.makeRef(resultFactory, accumulator, combiner));
     }
 
     @Override