1 /*
2 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25 package java.util.stream;
26
27 import java.util.Objects;
28 import java.util.Spliterator;
29 import java.util.function.IntFunction;
30 import java.util.function.Supplier;
31
32 /**
33 * Abstract base class for "pipeline" classes, which are the core
34 * implementations of the Stream interface and its primitive specializations.
35 * Manages construction and evaluation of stream pipelines.
36 *
37 * <p>An {@code AbstractPipeline} represents an initial portion of a stream
38 * pipeline, encapsulating a stream source and zero or more intermediate
39 * operations. The individual {@code AbstractPipeline} objects are often
40 * referred to as <em>stages</em>, where each stage describes either the stream
41 * source or an intermediate operation.
42 *
43 * <p>A concrete intermediate stage is generally built from an
44 * {@code AbstractPipeline}, a shape-specific pipeline class which extends it
45 * (e.g., {@code IntPipeline}) which is also abstract, and an operation-specific
46 * concrete class which extends that. {@code AbstractPipeline} contains most of
47 * the mechanics of evaluating the pipeline, and implements methods that will be
48 * used by the operation; the shape-specific classes add helper methods for
49 * dealing with collection of results into the appropriate shape-specific
50 * containers.
51 *
52 * <p>After chaining a new intermediate operation, or executing a terminal
53 * operation, the stream is considered to be consumed, and no more intermediate
54 * or terminal operations are permitted on this stream instance.
55 *
56 * @implNote
57 * <p>For sequential streams, and parallel streams without
58 * <a href="package-summary.html#StreamOps">stateful intermediate
59 * operations</a>, parallel streams, pipeline evaluation is done in a single
60 * pass that "jams" all the operations together. For parallel streams with
61 * stateful operations, execution is divided into segments, where each
62 * stateful operations marks the end of a segment, and each segment is
63 * evaluated separately and the result used as the input to the next
64 * segment. In all cases, the source data is not consumed until a terminal
65 * operation begins.
66 *
67 * @param <E_IN> type of input elements
68 * @param <E_OUT> type of output elements
69 * @param <S> type of the subclass implementing {@code BaseStream}
70 * @since 1.8
71 */
72 abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
73 extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {
74 /**
75 * Backlink to the head of the pipeline chain (self if this is the source
76 * stage).
77 */
78 private final AbstractPipeline sourceStage;
79
80 /**
81 * The "upstream" pipeline, or null if this is the source stage.
82 */
83 private final AbstractPipeline previousStage;
84
85 /**
86 * The operation flags for the intermediate operation represented by this
87 * pipeline object.
88 */
89 protected final int sourceOrOpFlags;
90
91 /**
92 * The next stage in the pipeline, or null if this is the last stage.
93 * Effectively final at the point of linking to the next pipeline.
94 */
95 private AbstractPipeline nextStage;
96
97 /**
98 * The number of intermediate operations between this pipeline object
99 * and the stream source if sequential, or the previous stateful if parallel.
100 * Valid at the point of pipeline preparation for evaluation.
101 */
102 private int depth;
103
104 /**
105 * The combined source and operation flags for the source and all operations
106 * up to and including the operation represented by this pipeline object.
107 * Valid at the point of pipeline preparation for evaluation.
108 */
109 private int combinedFlags;
110
111 /**
112 * The source spliterator. Only valid for the head pipeline.
113 * Before the pipeline is consumed if non-null then {@code sourceSupplier}
114 * must be null. After the pipeline is consumed if non-null then is set to
115 * null.
116 */
117 private Spliterator<?> sourceSpliterator;
118
119 /**
120 * The source supplier. Only valid for the head pipeline. Before the
121 * pipeline is consumed if non-null then {@code sourceSpliterator} must be
122 * null. After the pipeline is consumed if non-null then is set to null.
123 */
124 private Supplier<? extends Spliterator<?>> sourceSupplier;
125
126 /**
127 * True if this pipeline has been linked or consumed
128 */
129 private boolean linkedOrConsumed;
130
131 /**
132 * True if there are any stateful ops in the pipeline; only valid for the
133 * source stage.
134 */
135 private boolean sourceAnyStateful;
136
137 /**
138 * True if pipeline is parallel, otherwise the pipeline is sequential; only
139 * valid for the source stage.
140 */
141 private boolean parallel;
142
143 /**
144 * Constructor for the head of a stream pipeline.
145 *
146 * @param source {@code Supplier<Spliterator>} describing the stream source
147 * @param sourceFlags The source flags for the stream source, described in
148 * {@link StreamOpFlag}
149 * @param parallel True if the pipeline is parallel
150 */
151 AbstractPipeline(Supplier<? extends Spliterator<?>> source,
152 int sourceFlags, boolean parallel) {
153 this.previousStage = null;
154 this.sourceSupplier = source;
155 this.sourceStage = this;
156 this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
157 // The following is an optimization of:
158 // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
159 this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
160 this.depth = 0;
161 this.parallel = parallel;
162 }
163
164 /**
165 * Constructor for the head of a stream pipeline.
166 *
167 * @param source {@code Spliterator} describing the stream source
168 * @param sourceFlags the source flags for the stream source, described in
169 * {@link StreamOpFlag}
170 * @param parallel {@code true} if the pipeline is parallel
171 */
172 AbstractPipeline(Spliterator<?> source,
173 int sourceFlags, boolean parallel) {
174 this.previousStage = null;
175 this.sourceSpliterator = source;
176 this.sourceStage = this;
177 this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
178 // The following is an optimization of:
179 // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
180 this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
181 this.depth = 0;
182 this.parallel = parallel;
183 }
184
185 /**
186 * Constructor for appending an intermediate operation stage onto an
187 * existing pipeline.
188 *
189 * @param previousStage the upstream pipeline stage
190 * @param opFlags the operation flags for the new stage, described in
191 * {@link StreamOpFlag}
192 */
193 AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
194 if (previousStage.linkedOrConsumed)
195 throw new IllegalStateException("stream has already been operated upon");
196 previousStage.linkedOrConsumed = true;
197 previousStage.nextStage = this;
198
199 this.previousStage = previousStage;
200 this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
201 this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
202 this.sourceStage = previousStage.sourceStage;
203 if (opIsStateful())
204 sourceStage.sourceAnyStateful = true;
205 this.depth = previousStage.depth + 1;
206 }
207
208
209 // Terminal evaluation methods
210
211 /**
212 * Evaluate the pipeline with a terminal operation to produce a result.
213 *
214 * @param <R> the type of result
215 * @param terminalOp the terminal operation to be applied to the pipeline.
216 * @return the result
217 */
218 final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
219 assert getOutputShape() == terminalOp.inputShape();
220 if (linkedOrConsumed)
221 throw new IllegalStateException("stream has already been operated upon");
222 linkedOrConsumed = true;
223
224 return isParallel()
225 ? (R) terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
226 : (R) terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
227 }
228
229 /**
230 * Collect the elements output from the pipeline stage.
231 *
232 * @param generator the array generator to be used to create array instances
233 * @return a flat array-backed Node that holds the collected output elements
234 */
235 final Node<E_OUT> evaluateToArrayNode(IntFunction<E_OUT[]> generator) {
236 if (linkedOrConsumed)
237 throw new IllegalStateException("stream has already been operated upon");
238 linkedOrConsumed = true;
239
240 // If the last intermediate operation is stateful then
241 // evaluate directly to avoid an extra collection step
242 if (isParallel() && previousStage != null && opIsStateful()) {
243 return opEvaluateParallel(previousStage, previousStage.sourceSpliterator(0), generator);
244 }
245 else {
246 return evaluate(sourceSpliterator(0), true, generator);
247 }
248 }
249
250 /**
251 * Gets the source stage spliterator if this pipeline stage is the source
252 * stage. The pipeline is consumed after this method is called and
253 * returns successfully.
254 *
255 * @return the source stage spliterator
256 * @throws IllegalStateException if this pipeline stage is not the source
257 * stage.
258 */
259 final Spliterator<E_OUT> sourceStageSpliterator() {
260 if (this != sourceStage)
261 throw new IllegalStateException();
262
263 if (linkedOrConsumed)
264 throw new IllegalStateException("stream has already been operated upon");
265 linkedOrConsumed = true;
266
267 if (sourceStage.sourceSpliterator != null) {
268 Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
269 sourceStage.sourceSpliterator = null;
270 return s;
271 }
272 else if (sourceStage.sourceSupplier != null) {
273 Spliterator<E_OUT> s = (Spliterator<E_OUT>) sourceStage.sourceSupplier.get();
274 sourceStage.sourceSupplier = null;
275 return s;
276 }
277 else {
278 throw new IllegalStateException("source already consumed");
279 }
280 }
281
282 // BaseStream
283
284 @Override
285 public final S sequential() {
286 sourceStage.parallel = false;
287 return (S) this;
288 }
289
290 @Override
291 public final S parallel() {
292 sourceStage.parallel = true;
293 return (S) this;
294 }
295
296 // Primitive specialization use co-variant overrides, hence is not final
297 @Override
298 public Spliterator<E_OUT> spliterator() {
299 if (linkedOrConsumed)
300 throw new IllegalStateException("stream has already been operated upon");
301 linkedOrConsumed = true;
302
303 if (this == sourceStage) {
304 if (sourceStage.sourceSpliterator != null) {
305 Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
306 sourceStage.sourceSpliterator = null;
307 return s;
308 }
309 else if (sourceStage.sourceSupplier != null) {
310 Supplier<Spliterator<E_OUT>> s = sourceStage.sourceSupplier;
311 sourceStage.sourceSupplier = null;
312 return lazySpliterator(s);
313 }
314 else {
315 throw new IllegalStateException("source already consumed");
316 }
317 }
318 else {
319 return wrap(this, () -> sourceSpliterator(0), isParallel());
320 }
321 }
322
323 @Override
324 public final boolean isParallel() {
325 return sourceStage.parallel;
326 }
327
328
329 /**
330 * Returns the composition of stream flags of the stream source and all
331 * intermediate operations.
332 *
333 * @return the composition of stream flags of the stream source and all
334 * intermediate operations
335 * @see StreamOpFlag
336 */
337 final int getStreamFlags() {
338 return StreamOpFlag.toStreamFlags(combinedFlags);
339 }
340
341 /**
342 * Prepare the pipeline for a parallel execution. As the pipeline is built,
343 * the flags and depth indicators are set up for a sequential execution.
344 * If the execution is parallel, and there are any stateful operations, then
345 * some of these need to be adjusted, as well as adjusting for flags from
346 * the terminal operation (such as back-propagating UNORDERED).
347 * Need not be called for a sequential execution.
348 *
349 * @param terminalFlags Operation flags for the terminal operation
350 */
351 private void parallelPrepare(int terminalFlags) {
352 AbstractPipeline backPropagationHead = sourceStage;
353 if (sourceStage.sourceAnyStateful) {
354 int depth = 1;
355 for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage;
356 p != null;
357 u = p, p = p.nextStage) {
358 int thisOpFlags = p.sourceOrOpFlags;
359 if (p.opIsStateful()) {
360 // If the stateful operation is a short-circuit operation
361 // then move the back propagation head forwards
362 // NOTE: there are no size-injecting ops
363 if (StreamOpFlag.SHORT_CIRCUIT.isKnown(thisOpFlags)) {
364 backPropagationHead = p;
365 // Clear the short circuit flag for next pipeline stage
366 // This stage encapsulates short-circuiting, the next
367 // stage may not have any short-circuit operations, and
368 // if so spliterator.forEachRemaining should be be used
369 // for traversal
370 thisOpFlags = thisOpFlags & ~StreamOpFlag.IS_SHORT_CIRCUIT;
371 }
372
373 depth = 0;
374 // The following injects size, it is equivalent to:
375 // StreamOpFlag.combineOpFlags(StreamOpFlag.IS_SIZED, p.combinedFlags);
376 thisOpFlags = (thisOpFlags & ~StreamOpFlag.NOT_SIZED) | StreamOpFlag.IS_SIZED;
377 }
378 p.depth = depth++;
379 p.combinedFlags = StreamOpFlag.combineOpFlags(thisOpFlags, u.combinedFlags);
380 }
381 }
382
383 // Apply the upstream terminal flags
384 if (terminalFlags != 0) {
385 int upstreamTerminalFlags = terminalFlags & StreamOpFlag.UPSTREAM_TERMINAL_OP_MASK;
386 for (AbstractPipeline p = backPropagationHead; p.nextStage != null; p = p.nextStage) {
387 p.combinedFlags = StreamOpFlag.combineOpFlags(upstreamTerminalFlags, p.combinedFlags);
388 }
389
390 combinedFlags = StreamOpFlag.combineOpFlags(terminalFlags, combinedFlags);
391 }
392 }
393
394 /**
395 * Get the source spliterator for this pipeline stage. For a sequential or
396 * stateless parallel pipeline, this is the source spliterator. For a
397 * stateful parallel pipeline, this is a spliterator describing the results
398 * of all computations up to and including the most recent stateful
399 * operation.
400 */
401 private Spliterator<?> sourceSpliterator(int terminalFlags) {
402 // Get the source spliterator of the pipeline
403 Spliterator<?> spliterator = null;
404 if (sourceStage.sourceSpliterator != null) {
405 spliterator = sourceStage.sourceSpliterator;
406 sourceStage.sourceSpliterator = null;
407 }
408 else if (sourceStage.sourceSupplier != null) {
409 spliterator = (Spliterator<?>) sourceStage.sourceSupplier.get();
410 sourceStage.sourceSupplier = null;
411 }
412 else {
413 throw new IllegalStateException("source already consumed");
414 }
415
416 if (isParallel()) {
417 // @@@ Merge parallelPrepare with the loop below and use the
418 // spliterator characteristics to determine if SIZED
419 // should be injected
420 parallelPrepare(terminalFlags);
421
422 // Adapt the source spliterator, evaluating each stateful op
423 // in the pipeline up to and including this pipeline stage
424 for (AbstractPipeline u = sourceStage, p = sourceStage.nextStage, e = this;
425 u != e;
426 u = p, p = p.nextStage) {
427
428 if (p.opIsStateful()) {
429 spliterator = p.opEvaluateParallelLazy(u, spliterator);
430 }
431 }
432 }
433 else if (terminalFlags != 0) {
434 combinedFlags = StreamOpFlag.combineOpFlags(terminalFlags, combinedFlags);
435 }
436
437 return spliterator;
438 }
439
440
441 // PipelineHelper
442
443 @Override
444 final StreamShape getSourceShape() {
445 AbstractPipeline p = AbstractPipeline.this;
446 while (p.depth > 0) {
447 p = p.previousStage;
448 }
449 return p.getOutputShape();
450 }
451
452 @Override
453 final <P_IN> long exactOutputSizeIfKnown(Spliterator<P_IN> spliterator) {
454 return StreamOpFlag.SIZED.isKnown(getStreamAndOpFlags()) ? spliterator.getExactSizeIfKnown() : -1;
455 }
456
457 @Override
458 final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
459 copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
460 return sink;
461 }
462
463 @Override
464 final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
465 Objects.requireNonNull(wrappedSink);
466
467 if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
468 wrappedSink.begin(spliterator.getExactSizeIfKnown());
469 spliterator.forEachRemaining(wrappedSink);
470 wrappedSink.end();
471 }
472 else {
473 copyIntoWithCancel(wrappedSink, spliterator);
474 }
475 }
476
477 @Override
478 final <P_IN> void copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
479 AbstractPipeline p = AbstractPipeline.this;
480 while (p.depth > 0) {
481 p = p.previousStage;
482 }
483 wrappedSink.begin(spliterator.getExactSizeIfKnown());
484 p.forEachWithCancel(spliterator, wrappedSink);
485 wrappedSink.end();
486 }
487
488 @Override
489 final int getStreamAndOpFlags() {
490 return combinedFlags;
491 }
492
493 final boolean isOrdered() {
494 return StreamOpFlag.ORDERED.isKnown(combinedFlags);
495 }
496
497 @Override
498 final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
499 Objects.requireNonNull(sink);
500
501 for (AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
502 sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
503 }
504 return (Sink<P_IN>) sink;
505 }
506
507 @Override
508 final <P_IN> Spliterator<E_OUT> wrapSpliterator(Spliterator<P_IN> sourceSpliterator) {
509 if (depth == 0) {
510 return (Spliterator<E_OUT>) sourceSpliterator;
511 }
512 else {
513 return wrap(this, () -> sourceSpliterator, isParallel());
514 }
515 }
516
517 @Override
518 @SuppressWarnings("unchecked")
519 final <P_IN> Node<E_OUT> evaluate(Spliterator<P_IN> spliterator,
520 boolean flatten,
521 IntFunction<E_OUT[]> generator) {
522 if (isParallel()) {
523 // @@@ Optimize if op of this pipeline stage is a stateful op
524 return evaluateToNode(this, spliterator, flatten, generator);
525 }
526 else {
527 Node.Builder<E_OUT> nb = makeNodeBuilder(
528 exactOutputSizeIfKnown(spliterator), generator);
529 return wrapAndCopyInto(nb, spliterator).build();
530 }
531 }
532
533
534 // Shape-specific abstract methods, implemented by XxxPipeline classes
535
536 /**
537 * Get the output shape of the pipeline. If the pipeline is the head,
538 * then it's output shape corresponds to the shape of the source.
539 * Otherwise, it's output shape corresponds to the output shape of the
540 * associated operation.
541 *
542 * @return the output shape
543 */
544 abstract StreamShape getOutputShape();
545
546 /**
547 * Collect elements output from a pipeline into a Node that holds elements
548 * of this shape.
549 *
550 * @param helper the pipeline helper describing the pipeline stages
551 * @param spliterator the source spliterator
552 * @param flattenTree true if the returned node should be flattened
553 * @param generator the array generator
554 * @return a Node holding the output of the pipeline
555 */
556 abstract <P_IN> Node<E_OUT> evaluateToNode(PipelineHelper<E_OUT> helper,
557 Spliterator<P_IN> spliterator,
558 boolean flattenTree,
559 IntFunction<E_OUT[]> generator);
560
561 /**
562 * Create a spliterator that wraps a source spliterator, compatible with
563 * this stream shape, and operations associated with a {@link
564 * PipelineHelper}.
565 *
566 * @param ph the pipeline helper describing the pipeline stages
567 * @param supplier the supplier of a spliterator
568 * @return a wrapping spliterator compatible with this shape
569 */
570 abstract <P_IN> Spliterator<E_OUT> wrap(PipelineHelper<E_OUT> ph,
571 Supplier<Spliterator<P_IN>> supplier,
572 boolean isParallel);
573
574 /**
575 * Create a lazy spliterator that wraps and obtains the supplied the
576 * spliterator when a method is invoked on the lazy spliterator.
577 * @param supplier the supplier of a spliterator
578 */
579 abstract Spliterator<E_OUT> lazySpliterator(Supplier<? extends Spliterator<E_OUT>> supplier);
580
581 /**
582 * Traverse the elements of a spliterator compatible with this stream shape,
583 * pushing those elements into a sink. If the sink requests cancellation,
584 * no further elements will be pulled or pushed.
585 *
586 * @param spliterator the spliterator to pull elements from
587 * @param sink the sink to push elements to
588 */
589 abstract void forEachWithCancel(Spliterator<E_OUT> spliterator, Sink<E_OUT> sink);
590
591 /**
592 * Make a node builder compatible with this stream shape.
593 *
594 * @param exactSizeIfKnown if {@literal >=0}, then a node builder will be created that
595 * has a fixed capacity of at most sizeIfKnown elements. If {@literal < 0},
596 * then the node builder has an unfixed capacity. A fixed capacity node
597 * builder will throw exceptions if an element is added after builder has
598 * reached capacity, or is built before the builder has reached capacity.
599 * @param generator the array generator to be used to create instances of a
600 * T[] array. For implementations supporting primitive nodes, this parameter
601 * may be ignored.
602 * @return a node builder
603 */
604 abstract Node.Builder<E_OUT> makeNodeBuilder(long exactSizeIfKnown,
605 IntFunction<E_OUT[]> generator);
606
607
608 // Op-specific abstract methods, implemented by the operation class
609
610 /**
611 * Returns whether this operation is stateful or not. If it is stateful,
612 * then the method
613 * {@link #opEvaluateParallel(PipelineHelper, java.util.Spliterator, java.util.function.IntFunction)}
614 * must be overridden.
615 *
616 * @return {@code true} if this operation is stateful
617 */
618 abstract boolean opIsStateful();
619
620 /**
621 * Accepts a {@code Sink} which will receive the results of this operation,
622 * and return a {@code Sink} which accepts elements of the input type of
623 * this operation and which performs the operation, passing the results to
624 * the provided {@code Sink}.
625 *
626 * @apiNote
627 * The implementation may use the {@code flags} parameter to optimize the
628 * sink wrapping. For example, if the input is already {@code DISTINCT},
629 * the implementation for the {@code Stream#distinct()} method could just
630 * return the sink it was passed.
631 *
632 * @param flags The combined stream and operation flags up to, but not
633 * including, this operation
634 * @param sink sink to which elements should be sent after processing
635 * @return a sink which accepts elements, perform the operation upon
636 * each element, and passes the results (if any) to the provided
637 * {@code Sink}.
638 */
639 abstract Sink<E_IN> opWrapSink(int flags, Sink<E_OUT> sink);
640
641 /**
642 * Performs a parallel evaluation of the operation using the specified
643 * {@code PipelineHelper} which describes the upstream intermediate
644 * operations. Only called on stateful operations. If {@link
645 * #opIsStateful()} returns true then implementations must override the
646 * default implementation.
647 *
648 * @implSpec The default implementation always throw
649 * {@code UnsupportedOperationException}.
650 *
651 * @param helper the pipeline helper describing the pipeline stages
652 * @param spliterator the source {@code Spliterator}
653 * @param generator the array generator
654 * @return a {@code Node} describing the result of the evaluation
655 */
656 <P_IN> Node<E_OUT> opEvaluateParallel(PipelineHelper<E_OUT> helper,
657 Spliterator<P_IN> spliterator,
658 IntFunction<E_OUT[]> generator) {
659 throw new UnsupportedOperationException("Parallel evaluation is not supported");
660 }
661
662 /**
663 * Returns a {@code Spliterator} describing a parallel evaluation of the
664 * operation, using the specified {@code PipelineHelper} which describes the
665 * upstream intermediate operations. Only called on stateful operations.
666 * It is not necessary (though acceptable) to do a full computation of the
667 * result here; it is preferable, if possible, to describe the result via a
668 * lazily evaluated spliterator.
669 *
670 * @implSpec The default implementation behaves as if:
671 * <pre>{@code
672 * return evaluateParallel(helper, i -> (E_OUT[]) new
673 * Object[i]).spliterator();
674 * }</pre>
675 * and is suitable for implementations that cannot do better than a full
676 * synchronous evaluation.
677 *
678 * @param helper the pipeline helper
679 * @param spliterator the source {@code Spliterator}
680 * @return a {@code Spliterator} describing the result of the evaluation
681 */
682 <P_IN> Spliterator<E_OUT> opEvaluateParallelLazy(PipelineHelper<E_OUT> helper,
683 Spliterator<P_IN> spliterator) {
684 return opEvaluateParallel(helper, spliterator, i -> (E_OUT[]) new Object[i]).spliterator();
685 }
686 }