Take 2: PipelineHelper and ParallelPipelineHelper

Paul Sandoz paul.sandoz at oracle.com
Fri Sep 28 09:14:44 PDT 2012


Here is another webrev:


I have tried to make the PipelineHelper classes invariant as much as possible to the source but this is trickier in the serial case where the helper encapsulates the knowledge to push or pull. I think it may take a couple more iterations to get this right (perhaps also re-thinking properties on ops).

The evaluateSequential/evaluateParallel both pass the StreamAccessor and the helper. The intent is that evaluateParallel should pass a Spliterator and both  Spliterator and StreamAccessor extend from a common common type, X say which currently does not exist, that is the parameter on evaluateSequential:

  evaluateSequential(X x, PipelineHelper ph) { ... }

  evaluateParallel(Spliterator x, ParallelPipelineHelper ph) {
    return evaluateSequential(x, ph);

so that the former can easily defer to the latter for common cases or for F/J leaf cases.

On Sep 24, 2012, at 9:16 PM, Paul Sandoz <Paul.Sandoz at oracle.com> wrote:

> Hi
> Here is a rough start refactoring to introduce a PipelineHelper for the serial case and a corresponding ParallelPipelineHelper for the parallel case:
>  http://cr.openjdk.java.net/~psandoz/lambda/ophelper/webrev/
> Needs more work before committing.
> All the tests pass.
> Part of the reason for this is to clean up TerminalOp and ShortCircuitTerminalOp. For serial evaluation the latter should always pull (even if a sink is used) while the former could pull/push depending on whether there is a intermediate short-circuiting op, or the stream has already been pulled from before evaluation, or because it is implementation chooses to pull rather than push.
> I am not sure i have the methods on PipelineHelper/ParallelPipelineHelper quite sorted out just yet. 
> In the parallel case ParallelPipelineHelper is most often used as follows: 
>  spliterator.into(helper.wrapSink(reduceStage)); // non-short circuiting
>  helper.wrapIterator(spliterator.iterator()) // short-circuiting
> So currently the only relationship between ParallelPipelineHelper and PipelineHelper is the requirement for parallel evaluation to defer to serial evaluation when there is no implementation for the former. Hence the inheritance relationship:
>    RES evaluate(PipelineHelper<E_RES> helper);
>    @Override
>    <E_SRC> RES evaluateParallel(ParallelPipelineHelper<E_SRC, E_RES> helper) default {
>        System.out.println(getClass().getSimpleName() + " using TerminalOp.computeParallel serial default");
>        return evaluate(helper);
>    }
> I also tried to make clearer the definition of the type variables.
> Paul.

More information about the lambda-dev mailing list