1 /* 2 * Copyright (c) 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.DoubleSummaryStatistics; 28 import java.util.Objects; 29 import java.util.OptionalDouble; 30 import java.util.PrimitiveIterator; 31 import java.util.Spliterator; 32 import java.util.Spliterators; 33 import java.util.function.BiConsumer; 34 import java.util.function.BinaryOperator; 35 import java.util.function.DoubleBinaryOperator; 36 import java.util.function.DoubleConsumer; 37 import java.util.function.DoubleFunction; 38 import java.util.function.DoublePredicate; 39 import java.util.function.DoubleToIntFunction; 40 import java.util.function.DoubleToLongFunction; 41 import java.util.function.DoubleUnaryOperator; 42 import java.util.function.IntFunction; 43 import java.util.function.ObjDoubleConsumer; 44 import java.util.function.Supplier; 45 46 /** 47 * Abstract base class for an intermediate pipeline stage or pipeline source 48 * stage implementing whose elements are of type {@code double}. 49 * 50 * @param <E_IN> type of elements in the upstream source 51 * 52 * @since 1.8 53 */ 54 abstract class DoublePipeline<E_IN> 55 extends AbstractPipeline<E_IN, Double, DoubleStream> 56 implements DoubleStream { 57 58 /** 59 * Constructor for the head of a stream pipeline. 60 * 61 * @param source {@code Supplier<Spliterator>} describing the stream source 62 * @param sourceFlags the source flags for the stream source, described in 63 * {@link StreamOpFlag} 64 */ 65 DoublePipeline(Supplier<? extends Spliterator<Double>> source, 66 int sourceFlags, boolean parallel) { 67 super(source, sourceFlags, parallel); 68 } 69 70 /** 71 * Constructor for the head of a stream pipeline. 72 * 73 * @param source {@code Spliterator} describing the stream source 74 * @param sourceFlags the source flags for the stream source, described in 75 * {@link StreamOpFlag} 76 */ 77 DoublePipeline(Spliterator<Double> source, 78 int sourceFlags, boolean parallel) { 79 super(source, sourceFlags, parallel); 80 } 81 82 /** 83 * Constructor for appending an intermediate operation onto an existing 84 * pipeline. 85 * 86 * @param upstream the upstream element source. 87 * @param opFlags the operation flags 88 */ 89 DoublePipeline(AbstractPipeline<?, E_IN, ?> upstream, int opFlags) { 90 super(upstream, opFlags); 91 } 92 93 /** 94 * Adapt a {@code Sink<Double> to a {@code DoubleConsumer}, ideally simply 95 * by casting. 96 */ 97 private static DoubleConsumer adapt(Sink<Double> sink) { 98 if (sink instanceof DoubleConsumer) { 99 return (DoubleConsumer) sink; 100 } else { 101 if (Tripwire.ENABLED) 102 Tripwire.trip(AbstractPipeline.class, 103 "using DoubleStream.adapt(Sink<Double> s)"); 104 return sink::accept; 105 } 106 } 107 108 /** 109 * Adapt a {@code Spliterator<Double>} to a {@code Spliterator.OfDouble}. 110 * 111 * @implNote 112 * The implementation attempts to cast to a Spliterator.OfDouble, and throws 113 * an exception if this cast is not possible. 114 */ 115 private static Spliterator.OfDouble adapt(Spliterator<Double> s) { 116 if (s instanceof Spliterator.OfDouble) { 117 return (Spliterator.OfDouble) s; 118 } else { 119 if (Tripwire.ENABLED) 120 Tripwire.trip(AbstractPipeline.class, 121 "using DoubleStream.adapt(Spliterator<Double> s)"); 122 throw new UnsupportedOperationException("DoubleStream.adapt(Spliterator<Double> s)"); 123 } 124 } 125 126 127 // Shape-specific methods 128 129 @Override 130 final StreamShape getOutputShape() { 131 return StreamShape.DOUBLE_VALUE; 132 } 133 134 @Override 135 final <P_IN> Node<Double> evaluateToNode(PipelineHelper<Double> helper, 136 Spliterator<P_IN> spliterator, 137 boolean flattenTree, 138 IntFunction<Double[]> generator) { 139 return Nodes.collectDouble(helper, spliterator, flattenTree); 140 } 141 142 @Override 143 final <P_IN> Spliterator<Double> wrap(PipelineHelper<Double> ph, 144 Supplier<Spliterator<P_IN>> supplier, 145 boolean isParallel) { 146 return new StreamSpliterators.DoubleWrappingSpliterator<>(ph, supplier, isParallel); 147 } 148 149 @Override 150 @SuppressWarnings("unchecked") 151 final Spliterator.OfDouble lazySpliterator(Supplier<? extends Spliterator<Double>> supplier) { 152 return new StreamSpliterators.DelegatingSpliterator.OfDouble((Supplier<Spliterator.OfDouble>) supplier); 153 } 154 155 @Override 156 final void forEachWithCancel(Spliterator<Double> spliterator, Sink<Double> sink) { 157 Spliterator.OfDouble spl = adapt(spliterator); 158 DoubleConsumer adaptedSink = adapt(sink); 159 do { } while (!sink.cancellationRequested() && spl.tryAdvance(adaptedSink)); 160 } 161 162 @Override 163 final Node.Builder<Double> makeNodeBuilder(long exactSizeIfKnown, IntFunction<Double[]> generator) { 164 return Nodes.doubleBuilder(exactSizeIfKnown); 165 } 166 167 168 // DoubleStream 169 170 @Override 171 public final PrimitiveIterator.OfDouble iterator() { 172 return Spliterators.iterator(spliterator()); 173 } 174 175 @Override 176 public final Spliterator.OfDouble spliterator() { 177 return adapt(super.spliterator()); 178 } 179 180 // Stateless intermediate ops from DoubleStream 181 182 @Override 183 public final Stream<Double> boxed() { 184 return mapToObj(Double::valueOf); 185 } 186 187 @Override 188 public final DoubleStream map(DoubleUnaryOperator mapper) { 189 Objects.requireNonNull(mapper); 190 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 191 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { 192 @Override 193 Sink<Double> opWrapSink(int flags, Sink<Double> sink) { 194 return new Sink.ChainedDouble<Double>(sink) { 195 @Override 196 public void accept(double t) { 197 downstream.accept(mapper.applyAsDouble(t)); 198 } 199 }; 200 } 201 }; 202 } 203 204 @Override 205 public final <U> Stream<U> mapToObj(DoubleFunction<? extends U> mapper) { 206 Objects.requireNonNull(mapper); 207 return new ReferencePipeline.StatelessOp<Double, U>(this, StreamShape.DOUBLE_VALUE, 208 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { 209 @Override 210 Sink<Double> opWrapSink(int flags, Sink<U> sink) { 211 return new Sink.ChainedDouble<U>(sink) { 212 @Override 213 public void accept(double t) { 214 downstream.accept(mapper.apply(t)); 215 } 216 }; 217 } 218 }; 219 } 220 221 @Override 222 public final IntStream mapToInt(DoubleToIntFunction mapper) { 223 Objects.requireNonNull(mapper); 224 return new IntPipeline.StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 225 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { 226 @Override 227 Sink<Double> opWrapSink(int flags, Sink<Integer> sink) { 228 return new Sink.ChainedDouble<Integer>(sink) { 229 @Override 230 public void accept(double t) { 231 downstream.accept(mapper.applyAsInt(t)); 232 } 233 }; 234 } 235 }; 236 } 237 238 @Override 239 public final LongStream mapToLong(DoubleToLongFunction mapper) { 240 Objects.requireNonNull(mapper); 241 return new LongPipeline.StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 242 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { 243 @Override 244 Sink<Double> opWrapSink(int flags, Sink<Long> sink) { 245 return new Sink.ChainedDouble<Long>(sink) { 246 @Override 247 public void accept(double t) { 248 downstream.accept(mapper.applyAsLong(t)); 249 } 250 }; 251 } 252 }; 253 } 254 255 @Override 256 public final DoubleStream flatMap(DoubleFunction<? extends DoubleStream> mapper) { 257 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 258 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) { 259 @Override 260 Sink<Double> opWrapSink(int flags, Sink<Double> sink) { 261 return new Sink.ChainedDouble<Double>(sink) { 262 @Override 263 public void begin(long size) { 264 downstream.begin(-1); 265 } 266 267 @Override 268 public void accept(double t) { 269 try (DoubleStream result = mapper.apply(t)) { 270 // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it 271 if (result != null) 272 result.sequential().forEach(i -> downstream.accept(i)); 273 } 274 } 275 }; 276 } 277 }; 278 } 279 280 @Override 281 public DoubleStream unordered() { 282 if (!isOrdered()) 283 return this; 284 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, StreamOpFlag.NOT_ORDERED) { 285 @Override 286 Sink<Double> opWrapSink(int flags, Sink<Double> sink) { 287 return sink; 288 } 289 }; 290 } 291 292 @Override 293 public final DoubleStream filter(DoublePredicate predicate) { 294 Objects.requireNonNull(predicate); 295 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 296 StreamOpFlag.NOT_SIZED) { 297 @Override 298 Sink<Double> opWrapSink(int flags, Sink<Double> sink) { 299 return new Sink.ChainedDouble<Double>(sink) { 300 @Override 301 public void begin(long size) { 302 downstream.begin(-1); 303 } 304 305 @Override 306 public void accept(double t) { 307 if (predicate.test(t)) 308 downstream.accept(t); 309 } 310 }; 311 } 312 }; 313 } 314 315 @Override 316 public final DoubleStream peek(DoubleConsumer action) { 317 Objects.requireNonNull(action); 318 return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, 319 0) { 320 @Override 321 Sink<Double> opWrapSink(int flags, Sink<Double> sink) { 322 return new Sink.ChainedDouble<Double>(sink) { 323 @Override 324 public void accept(double t) { 325 action.accept(t); 326 downstream.accept(t); 327 } 328 }; 329 } 330 }; 331 } 332 333 // Stateful intermediate ops from DoubleStream 334 335 @Override 336 public final DoubleStream limit(long maxSize) { 337 if (maxSize < 0) 338 throw new IllegalArgumentException(Long.toString(maxSize)); 339 return SliceOps.makeDouble(this, (long) 0, maxSize); 340 } 341 342 @Override 343 public final DoubleStream skip(long n) { 344 if (n < 0) 345 throw new IllegalArgumentException(Long.toString(n)); 346 if (n == 0) 347 return this; 348 else { 349 long limit = -1; 350 return SliceOps.makeDouble(this, n, limit); 351 } 352 } 353 354 @Override 355 public final DoubleStream sorted() { 356 return SortedOps.makeDouble(this); 357 } 358 359 @Override 360 public final DoubleStream distinct() { 361 // While functional and quick to implement, this approach is not very efficient. 362 // An efficient version requires a double-specific map/set implementation. 363 return boxed().distinct().mapToDouble(i -> (double) i); 364 } 365 366 // Terminal ops from DoubleStream 367 368 @Override 369 public void forEach(DoubleConsumer consumer) { 370 evaluate(ForEachOps.makeDouble(consumer, false)); 371 } 372 373 @Override 374 public void forEachOrdered(DoubleConsumer consumer) { 375 evaluate(ForEachOps.makeDouble(consumer, true)); 376 } 377 378 @Override 379 public final double sum() { 380 /* 381 * In the arrays allocated for the collect operation, index 0 382 * holds the high-order bits of the running sum, index 1 holds 383 * the low-order bits of the sum computed via compensated 384 * summation, and index 2 holds the simple sum used to compute 385 * the proper result if the stream contains infinite values of 386 * the same sign. 387 */ 388 double[] summation = collect(() -> new double[3], 389 (ll, d) -> { 390 Collectors.sumWithCompensation(ll, d); 391 ll[2] += d; 392 }, 393 (ll, rr) -> { 394 Collectors.sumWithCompensation(ll, rr[0]); 395 Collectors.sumWithCompensation(ll, rr[1]); 396 ll[2] += rr[2]; 397 }); 398 399 return Collectors.computeFinalSum(summation); 400 } 401 402 @Override 403 public final OptionalDouble min() { 404 return reduce(Math::min); 405 } 406 407 @Override 408 public final OptionalDouble max() { 409 return reduce(Math::max); 410 } 411 412 /** 413 * {@inheritDoc} 414 * 415 * @implNote The {@code double} format can represent all 416 * consecutive integers in the range -2<sup>53</sup> to 417 * 2<sup>53</sup>. If the pipeline has more than 2<sup>53</sup> 418 * values, the divisor in the average computation will saturate at 419 * 2<sup>53</sup>, leading to additional numerical errors. 420 */ 421 @Override 422 public final OptionalDouble average() { 423 /* 424 * In the arrays allocated for the collect operation, index 0 425 * holds the high-order bits of the running sum, index 1 holds 426 * the low-order bits of the sum computed via compensated 427 * summation, index 2 holds the number of values seen, index 3 428 * holds the simple sum. 429 */ 430 double[] avg = collect(() -> new double[4], 431 (ll, d) -> { 432 ll[2]++; 433 Collectors.sumWithCompensation(ll, d); 434 ll[3] += d; 435 }, 436 (ll, rr) -> { 437 Collectors.sumWithCompensation(ll, rr[0]); 438 Collectors.sumWithCompensation(ll, rr[1]); 439 ll[2] += rr[2]; 440 ll[3] += rr[3]; 441 }); 442 return avg[2] > 0 443 ? OptionalDouble.of(Collectors.computeFinalSum(avg) / avg[2]) 444 : OptionalDouble.empty(); 445 } 446 447 @Override 448 public final long count() { 449 return mapToObj(e -> null).mapToInt(e -> 1).sum(); 450 } 451 452 @Override 453 public final DoubleSummaryStatistics summaryStatistics() { 454 return collect(DoubleSummaryStatistics::new, DoubleSummaryStatistics::accept, 455 DoubleSummaryStatistics::combine); 456 } 457 458 @Override 459 public final double reduce(double identity, DoubleBinaryOperator op) { 460 return evaluate(ReduceOps.makeDouble(identity, op)); 461 } 462 463 @Override 464 public final OptionalDouble reduce(DoubleBinaryOperator op) { 465 return evaluate(ReduceOps.makeDouble(op)); 466 } 467 468 @Override 469 public final <R> R collect(Supplier<R> supplier, 470 ObjDoubleConsumer<R> accumulator, 471 BiConsumer<R, R> combiner) { 472 BinaryOperator<R> operator = (left, right) -> { 473 combiner.accept(left, right); 474 return left; 475 }; 476 return evaluate(ReduceOps.makeDouble(supplier, accumulator, operator)); 477 } 478 479 @Override 480 public final boolean anyMatch(DoublePredicate predicate) { 481 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ANY)); 482 } 483 484 @Override 485 public final boolean allMatch(DoublePredicate predicate) { 486 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ALL)); 487 } 488 489 @Override 490 public final boolean noneMatch(DoublePredicate predicate) { 491 return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.NONE)); 492 } 493 494 @Override 495 public final OptionalDouble findFirst() { 496 return evaluate(FindOps.makeDouble(true)); 497 } 498 499 @Override 500 public final OptionalDouble findAny() { 501 return evaluate(FindOps.makeDouble(false)); 502 } 503 504 @Override 505 public final double[] toArray() { 506 return Nodes.flattenDouble((Node.OfDouble) evaluateToArrayNode(Double[]::new)) 507 .asPrimitiveArray(); 508 } 509 510 // 511 512 /** 513 * Source stage of a DoubleStream 514 * 515 * @param <E_IN> type of elements in the upstream source 516 */ 517 static class Head<E_IN> extends DoublePipeline<E_IN> { 518 /** 519 * Constructor for the source stage of a DoubleStream. 520 * 521 * @param source {@code Supplier<Spliterator>} describing the stream 522 * source 523 * @param sourceFlags the source flags for the stream source, described 524 * in {@link StreamOpFlag} 525 * @param parallel {@code true} if the pipeline is parallel 526 */ 527 Head(Supplier<? extends Spliterator<Double>> source, 528 int sourceFlags, boolean parallel) { 529 super(source, sourceFlags, parallel); 530 } 531 532 /** 533 * Constructor for the source stage of a DoubleStream. 534 * 535 * @param source {@code Spliterator} describing the stream source 536 * @param sourceFlags the source flags for the stream source, described 537 * in {@link StreamOpFlag} 538 * @param parallel {@code true} if the pipeline is parallel 539 */ 540 Head(Spliterator<Double> source, 541 int sourceFlags, boolean parallel) { 542 super(source, sourceFlags, parallel); 543 } 544 545 @Override 546 final boolean opIsStateful() { 547 throw new UnsupportedOperationException(); 548 } 549 550 @Override 551 final Sink<E_IN> opWrapSink(int flags, Sink<Double> sink) { 552 throw new UnsupportedOperationException(); 553 } 554 555 // Optimized sequential terminal operations for the head of the pipeline 556 557 @Override 558 public void forEach(DoubleConsumer consumer) { 559 if (!isParallel()) { 560 adapt(sourceStageSpliterator()).forEachRemaining(consumer); 561 } 562 else { 563 super.forEach(consumer); 564 } 565 } 566 567 @Override 568 public void forEachOrdered(DoubleConsumer consumer) { 569 if (!isParallel()) { 570 adapt(sourceStageSpliterator()).forEachRemaining(consumer); 571 } 572 else { 573 super.forEachOrdered(consumer); 574 } 575 } 576 577 } 578 579 /** 580 * Base class for a stateless intermediate stage of a DoubleStream. 581 * 582 * @param <E_IN> type of elements in the upstream source 583 * @since 1.8 584 */ 585 abstract static class StatelessOp<E_IN> extends DoublePipeline<E_IN> { 586 /** 587 * Construct a new DoubleStream by appending a stateless intermediate 588 * operation to an existing stream. 589 * 590 * @param upstream the upstream pipeline stage 591 * @param inputShape the stream shape for the upstream pipeline stage 592 * @param opFlags operation flags for the new stage 593 */ 594 StatelessOp(AbstractPipeline<?, E_IN, ?> upstream, 595 StreamShape inputShape, 596 int opFlags) { 597 super(upstream, opFlags); 598 assert upstream.getOutputShape() == inputShape; 599 } 600 601 @Override 602 final boolean opIsStateful() { 603 return false; 604 } 605 } 606 607 /** 608 * Base class for a stateful intermediate stage of a DoubleStream. 609 * 610 * @param <E_IN> type of elements in the upstream source 611 * @since 1.8 612 */ 613 abstract static class StatefulOp<E_IN> extends DoublePipeline<E_IN> { 614 /** 615 * Construct a new DoubleStream by appending a stateful intermediate 616 * operation to an existing stream. 617 * 618 * @param upstream the upstream pipeline stage 619 * @param inputShape the stream shape for the upstream pipeline stage 620 * @param opFlags operation flags for the new stage 621 */ 622 StatefulOp(AbstractPipeline<?, E_IN, ?> upstream, 623 StreamShape inputShape, 624 int opFlags) { 625 super(upstream, opFlags); 626 assert upstream.getOutputShape() == inputShape; 627 } 628 629 @Override 630 final boolean opIsStateful() { 631 return true; 632 } 633 634 @Override 635 abstract <P_IN> Node<Double> opEvaluateParallel(PipelineHelper<Double> helper, 636 Spliterator<P_IN> spliterator, 637 IntFunction<Double[]> generator); 638 } 639 }