1 /* 2 * Copyright (c) 2014 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. 8 * 9 * This code is distributed in the hope that it will be useful, but WITHOUT 10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 12 * version 2 for more details (a copy is included in the LICENSE file that 13 * accompanied this code). 14 * 15 * You should have received a copy of the GNU General Public License version 16 * 2 along with this work; if not, write to the Free Software Foundation, 17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 18 * 19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 20 * or visit www.oracle.com if you need additional information or have any 21 * questions. 22 */ 23 package org.openjdk.bench.java.util.stream.tasks.IntegerDuplicate; 24 25 import org.openjdk.jmh.annotations.Benchmark; 26 import org.openjdk.jmh.annotations.BenchmarkMode; 27 import org.openjdk.jmh.annotations.Mode; 28 import org.openjdk.jmh.annotations.OutputTimeUnit; 29 import org.openjdk.jmh.annotations.Scope; 30 import org.openjdk.jmh.annotations.Setup; 31 import org.openjdk.jmh.annotations.State; 32 33 import java.util.Arrays; 34 import java.util.Collections; 35 import java.util.concurrent.RecursiveAction; 36 import java.util.concurrent.TimeUnit; 37 import java.util.concurrent.atomic.LongAdder; 38 import java.util.function.Function; 39 import java.util.stream.Stream; 40 41 /** 42 * This benchmark assesses flatMap() performance. 43 */ 44 @BenchmarkMode(Mode.Throughput) 45 @OutputTimeUnit(TimeUnit.SECONDS) 46 @State(Scope.Thread) 47 public class Bulk { 48 49 /** 50 * Implementation notes: 51 * - parallel versions need to use special sink to get the values 52 */ 53 54 private IntegerDuplicateProblem problem; 55 private LongAdder adder; 56 57 @Setup 58 public void setup() { 59 problem = new IntegerDuplicateProblem(); 60 adder = new LongAdder(); 61 } 62 63 @Benchmark 64 public long hm_seq() { 65 for (Integer i : problem.get()) { 66 adder.add(i); 67 adder.add(i); 68 } 69 return adder.sum(); 70 } 71 72 @Benchmark 73 public long hm_par() { 74 new Task(problem.get(), adder, 0, problem.get().length).invoke(); 75 return adder.sum(); 76 } 77 78 @Benchmark 79 public long bulk_seq_inner() { 80 return Arrays.stream(problem.get()).flatMap(new Function<Integer, Stream<Integer>>() { 81 @Override 82 public Stream<Integer> apply(Integer integer) { 83 return Collections.nCopies(2, integer).stream(); 84 } 85 }).collect(LongAdder::new, LongAdder::add, (la1, la2) -> la1.add(la2.sum())).sum(); 86 } 87 88 @Benchmark 89 public long bulk_par_inner() { 90 return Arrays.stream(problem.get()).parallel().flatMap(new Function<Integer, Stream<Integer>>() { 91 @Override 92 public Stream<Integer> apply(Integer integer) { 93 return Collections.nCopies(2, integer).stream(); 94 } 95 }).collect(LongAdder::new, LongAdder::add, (la1, la2) -> la1.add(la2.sum())).sum(); 96 } 97 98 public static class Task extends RecursiveAction { 99 private static final int FORK_LIMIT = 500; 100 101 private final Integer[] data; 102 private final LongAdder sink; 103 private final int start; 104 private final int end; 105 106 Task(Integer[] w, LongAdder sink, int start, int end) { 107 this.data = w; 108 this.sink = sink; 109 this.start = start; 110 this.end = end; 111 } 112 113 @Override 114 protected void compute() { 115 int size = end - start; 116 if (size > FORK_LIMIT) { 117 int mid = start + size / 2; 118 Task t1 = new Task(data, sink, start, mid); 119 Task t2 = new Task(data, sink, mid, end); 120 t1.fork(); 121 t2.invoke(); 122 t1.join(); 123 } else { 124 for (int i = start; i < end; i++) { 125 sink.add(i); 126 sink.add(i); 127 } 128 } 129 } 130 } 131 132 }