PairTest.java

package org.sample;

import java.util.concurrent.TimeUnit;
import javax.util.streamex.*;
import java.util.stream.*;
import java.util.*;

import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.annotations.*;

@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@Fork(2)
@State(Scope.Benchmark)
public class PairTest {
	@Param({ "10000", "100000", "1000000" })
	private int n;

	private List<Integer> input;

	@Setup
	public void setUp() {
		input = IntStreamEx.of(new Random(1), n, 1, 10000).boxed().toList();
	}

	@Benchmark
	public void naiveIterator(Blackhole bh) {
		List<Integer> result = new ArrayList<>();
		Integer last = null;
		for (Integer cur : input) {
			if (last != null && last < cur)
				result.add(last);
			last = cur;
		}
		bh.consume(result);
	}

	@Benchmark
	public void naiveGet(Blackhole bh) {
		List<Integer> result = new ArrayList<>();
		for (int i = 0; i < input.size() - 1; i++) {
			Integer cur = input.get(i), next = input.get(i + 1);
			if (cur < next)
				result.add(cur);
		}
		bh.consume(result);
	}

	@Benchmark
	public void stream(Blackhole bh) {
		List<Integer> result = IntStream.range(0, input.size() - 1)
				.filter(i -> input.get(i) < input.get(i + 1))
				.mapToObj(input::get).collect(Collectors.toList());
		bh.consume(result);
	}

	@Benchmark
	public void streamParallel(Blackhole bh) {
		List<Integer> result = IntStream.range(0, input.size() - 1).parallel()
				.filter(i -> input.get(i) < input.get(i + 1))
				.mapToObj(input::get).collect(Collectors.toList());
		bh.consume(result);
	}

	@Benchmark
	public void streamEx(Blackhole bh) {
		List<Integer> result = StreamEx.of(input)
				.pairMap((a, b) -> a < b ? a : null).nonNull().toList();
		bh.consume(result);
	}

	@Benchmark
	public void streamExParallel(Blackhole bh) {
		List<Integer> result = StreamEx.of(input).parallel()
				.pairMap((a, b) -> a < b ? a : null).nonNull().toList();
		bh.consume(result);
	}

	@Benchmark
	public void reduce(Blackhole bh) {
		List<Integer> result = new ArrayList<>();
		input.stream().reduce((a, b) -> {
			if (a < b)
				result.add(a);
			return b;
		});
		bh.consume(result);
	}

	public static Collector<Integer, ?, List<Integer>> collectPrecedingValues() {
		int[] holder = { Integer.MAX_VALUE };
		return Collector.of(ArrayList::new, (l, elem) -> {
			if (holder[0] < elem)
				l.add(holder[0]);
			holder[0] = elem;
		}, (l1, l2) -> {
			throw new UnsupportedOperationException("Don't run in parallel");
		});
	}

	@Benchmark
	public void collector(Blackhole bh) {
		List<Integer> result = input.stream().collect(collectPrecedingValues());
		bh.consume(result);
	}

	public static Collector<Integer, ?, List<Integer>> collectPrecedingValuesParallel() {
		return Collectors.collectingAndThen(
				Collector.of(() -> new ArrayList<int[]>(), (l, elem) -> {
					if (l.isEmpty())
						l.add(new int[] { Integer.MAX_VALUE, elem });
					else
						l.add(new int[] { l.get(l.size() - 1)[1], elem });
				}, (l1, l2) -> {
					if (l1.isEmpty())
						return l2;
					if (l2.isEmpty())
						return l1;
					l2.get(0)[0] = l1.get(l1.size() - 1)[1];
					l1.addAll(l2);
					return l1;
				}),
				l -> l.parallelStream().filter(arr -> arr[0] < arr[1])
						.map(arr -> arr[0]).collect(Collectors.toList()));
	}

	@Benchmark
	public void collectorParallel(Blackhole bh) {
		List<Integer> result = input.stream().parallel()
				.collect(collectPrecedingValuesParallel());
		bh.consume(result);
	}
}

results.txt

Benchmark                       (n)  Mode  Cnt      Score      Error  Units
PairTest.collector            10000  avgt   20    126.179 ±    1.826  us/op
PairTest.collector           100000  avgt   20   1518.903 ±   15.395  us/op
PairTest.collector          1000000  avgt   20  17341.463 ±  212.640  us/op
PairTest.collectorParallel    10000  avgt   20    259.592 ±    6.603  us/op
PairTest.collectorParallel   100000  avgt   20   2997.632 ±   21.827  us/op
PairTest.collectorParallel  1000000  avgt   20  42631.489 ± 2359.500  us/op
PairTest.naiveGet             10000  avgt   20    109.022 ±    6.375  us/op
PairTest.naiveGet            100000  avgt   20   1212.518 ±   19.594  us/op
PairTest.naiveGet           1000000  avgt   20  13023.481 ±  123.957  us/op
PairTest.naiveIterator        10000  avgt   20    102.385 ±    6.677  us/op
PairTest.naiveIterator       100000  avgt   20   1156.980 ±   69.579  us/op
PairTest.naiveIterator      1000000  avgt   20  13304.636 ±  238.012  us/op
PairTest.reduce               10000  avgt   20    115.380 ±    3.281  us/op
PairTest.reduce              100000  avgt   20   1328.418 ±   21.185  us/op
PairTest.reduce             1000000  avgt   20  14438.555 ±  579.884  us/op
PairTest.stream               10000  avgt   20    165.663 ±    1.661  us/op
PairTest.stream              100000  avgt   20   1831.555 ±  155.805  us/op
PairTest.stream             1000000  avgt   20  19195.204 ±  109.736  us/op
PairTest.streamEx             10000  avgt   20    142.501 ±   11.263  us/op
PairTest.streamEx            100000  avgt   20   1502.853 ±   14.691  us/op
PairTest.streamEx           1000000  avgt   20  16220.175 ±  677.324  us/op
PairTest.streamExParallel     10000  avgt   20     64.650 ±    2.188  us/op
PairTest.streamExParallel    100000  avgt   20    677.746 ±   47.541  us/op
PairTest.streamExParallel   1000000  avgt   20   8706.159 ±   36.649  us/op
PairTest.streamParallel       10000  avgt   20     70.382 ±    4.120  us/op
PairTest.streamParallel      100000  avgt   20    681.116 ±    9.002  us/op
PairTest.streamParallel     1000000  avgt   20   8887.987 ±   84.066  us/op