From 65363b5f5fbe62a439a767f70a4e247318b24824 Mon Sep 17 00:00:00 2001 From: naftalmm Date: Sat, 5 Sep 2020 14:34:22 +0300 Subject: [PATCH 1/4] refactor WithFirstSpliterator; add couple of deprecated methods for the sake of benchmarking; add benchmarks to prove the absence of performance regression after refactoring; fix javaDocs for StreamEx.withFirst methods (actually this is not directly related to refactoring, as semantics was not changed) --- .../withFirst/WithFirstBenchmark.java | 81 ++++++++ src/main/java/one/util/streamex/StreamEx.java | 23 ++- .../util/streamex/WithFirstSpliterator.java | 184 ++++++++---------- .../streamex/WithFirstSpliteratorOld.java | 168 ++++++++++++++++ 4 files changed, 347 insertions(+), 109 deletions(-) create mode 100644 benchmark/src/java/main/one/util/streamex/benchmark/withFirst/WithFirstBenchmark.java create mode 100644 src/main/java/one/util/streamex/WithFirstSpliteratorOld.java diff --git a/benchmark/src/java/main/one/util/streamex/benchmark/withFirst/WithFirstBenchmark.java b/benchmark/src/java/main/one/util/streamex/benchmark/withFirst/WithFirstBenchmark.java new file mode 100644 index 00000000..b649ad3f --- /dev/null +++ b/benchmark/src/java/main/one/util/streamex/benchmark/withFirst/WithFirstBenchmark.java @@ -0,0 +1,81 @@ +/* + * Copyright 2015, 2020 StreamEx contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package one.util.streamex.benchmark.withFirst; + +import one.util.streamex.IntStreamEx; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; + +@State(Scope.Benchmark) +public class WithFirstBenchmark { + @Param({"100000"}) + private int N; + + @Benchmark + public Object[] parallelOld() { + return IntStreamEx.range(N).boxed().parallel().withFirstOld().toArray(); + } + + @Benchmark + public boolean parallelOldShortCircuit() { + return IntStreamEx.range(N).boxed().parallel().withFirstOld().anyMatch(x -> x.getKey() == -1); + } + + @Benchmark + public Object[] parallelNew() { + return IntStreamEx.range(N).boxed().parallel().withFirst().toArray(); + } + + @Benchmark + public boolean parallelNewShortCircuit() { + return IntStreamEx.range(N).boxed().parallel().withFirst().anyMatch(x -> x.getKey() == -1); + } + + @Benchmark + public Object[] sequentialOld() { + return IntStreamEx.range(N).boxed().withFirstOld().toArray(); + } + + @Benchmark + public boolean sequentialOldShortCircuit() { + return IntStreamEx.range(N).boxed().withFirstOld().anyMatch(x -> x.getKey() == -1); + } + + @Benchmark + public Object[] sequentialNew() { + return IntStreamEx.range(N).boxed().withFirst().toArray(); + } + + @Benchmark + public boolean sequentialNewShortCircuit() { + return IntStreamEx.range(N).boxed().withFirst().anyMatch(x -> x.getKey() == -1); + } + + public static void main(String[] args) throws RunnerException { + Options opt = new OptionsBuilder() + .include(WithFirstBenchmark.class.getSimpleName()) + .build(); + + new Runner(opt).run(); + } +} diff --git a/src/main/java/one/util/streamex/StreamEx.java b/src/main/java/one/util/streamex/StreamEx.java index 0a0b4688..559b80da 100644 --- a/src/main/java/one/util/streamex/StreamEx.java +++ b/src/main/java/one/util/streamex/StreamEx.java @@ -1764,11 +1764,6 @@ public StreamEx intervalMap(BiPredicate sameInterva *

* This is a quasi-intermediate * operation. - * - *

- * The size of the resulting stream is one element less than the input - * stream. If the input stream is empty or contains just one element, then - * the output stream will be empty. * * @param The element type of the new stream * @param mapper a non-interfering, stateless function to apply to the first @@ -1791,11 +1786,6 @@ public StreamEx withFirst(BiFunction m *

* This is a quasi-intermediate * operation. - * - *

- * The size of the resulting stream is one element less than the input - * stream. If the input stream is empty or contains just one element, then - * the output stream will be empty. * * @return the new stream * @see #withFirst(BiFunction) @@ -1807,6 +1797,19 @@ public EntryStream withFirst() { SimpleImmutableEntry::new); return new EntryStream<>(spliterator, context); } + + @Deprecated + public StreamEx withFirstOld(BiFunction mapper) { + WithFirstSpliteratorOld spliterator = new WithFirstSpliteratorOld<>(spliterator(), mapper); + return new StreamEx<>(spliterator, context); + } + + @Deprecated + public EntryStream withFirstOld() { + WithFirstSpliteratorOld> spliterator = new WithFirstSpliteratorOld<>(spliterator(), + SimpleImmutableEntry::new); + return new EntryStream<>(spliterator, context); + } /** * Creates a new {@link StreamEx} which is the result of applying of the diff --git a/src/main/java/one/util/streamex/WithFirstSpliterator.java b/src/main/java/one/util/streamex/WithFirstSpliterator.java index 295ce92e..9bd69ad4 100644 --- a/src/main/java/one/util/streamex/WithFirstSpliterator.java +++ b/src/main/java/one/util/streamex/WithFirstSpliterator.java @@ -16,153 +16,139 @@ package one.util.streamex; import java.util.Spliterator; -import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiFunction; import java.util.function.Consumer; import one.util.streamex.Internals.CloneableSpliterator; +import static one.util.streamex.Internals.NONE; + /** * @author Tagir Valeev */ -/* package */final class WithFirstSpliterator extends CloneableSpliterator> implements Consumer { - private static final int STATE_NONE = 0; - private static final int STATE_FIRST_READ = 1; - private static final int STATE_INIT = 2; - private static final int STATE_EMPTY = 3; - - private ReentrantLock lock; +/* package */final class WithFirstSpliterator extends CloneableSpliterator> { + private static final Object lock = new Object(); private Spliterator source; private WithFirstSpliterator prefix; - private volatile T first; - private volatile int state = STATE_NONE; + private T first = none(); + private T[] firstRefHolder; + boolean firstToBeConsumed; private final BiFunction mapper; - private Consumer action; WithFirstSpliterator(Spliterator source, BiFunction mapper) { this.source = source; this.mapper = mapper; } - private void acquire() { - if (lock != null && state == STATE_NONE) { - lock.lock(); - } + @SuppressWarnings("unchecked") + static T[] arrayOfNone() { + return (T[]) new Object[]{NONE}; } - private void release() { - if (lock != null && lock.isHeldByCurrentThread()) { - lock.unlock(); + @SuppressWarnings("unchecked") + static T none() { + return (T) NONE; + } + + @Override + public Spliterator trySplit() { + boolean isInit = first != NONE || (firstRefHolder != null && (first = firstRefHolder[0]) != NONE); + if (isInit) { + return null; + } + if (firstRefHolder == null) { + firstRefHolder = arrayOfNone(); + } + synchronized (lock) { + if (isInit()) { + return null; + } + Spliterator prefix = source.trySplit(); + if (prefix == null) { + return null; + } + WithFirstSpliterator result = doClone(); + result.source = prefix; + return this.prefix = result; } } - + + private boolean isInit() { + return first != NONE || (first = firstRefHolder[0]) != NONE; + } + @Override public boolean tryAdvance(Consumer action) { - if (state == STATE_NONE) { - acquire(); - try { + boolean isNotInit = first == NONE && (firstRefHolder == null || (first = firstRefHolder[0]) == NONE); + if (isNotInit) { + if (firstRefHolder == null) { + return source.tryAdvance(x -> { + first = x; + action.accept(mapper.apply(first, x)); + }); + } else { doInit(); - } finally { - release(); + if (first == NONE) { + return false; + } } } - if (state == STATE_FIRST_READ) { - state = STATE_INIT; + if (firstToBeConsumed) { + firstToBeConsumed = false; action.accept(mapper.apply(first, first)); return true; } - if (state != STATE_INIT) - return false; - this.action = action; - boolean hasNext = source.tryAdvance(this); - this.action = null; - return hasNext; + return source.tryAdvance(x -> action.accept(mapper.apply(first, x))); } - + private void doInit() { - int prefixState = state; - if (prefixState != STATE_NONE) - return; - if (prefix != null) { - prefix.doInit(); - prefixState = prefix.state; - } - if (prefixState == STATE_FIRST_READ || prefixState == STATE_INIT) { - first = prefix.first; - state = STATE_INIT; - return; + synchronized (lock) { + if (prefix != null) { + prefix.doInit(); + } + if (!isInit()) { + source.tryAdvance(x -> { + firstToBeConsumed = true; + first = x; + firstRefHolder[0] = x; + }); + } } - state = source.tryAdvance(x -> first = x) ? STATE_FIRST_READ : STATE_EMPTY; } - + @Override public void forEachRemaining(Consumer action) { - acquire(); - int myState = state; - this.action = action; - if (myState == STATE_FIRST_READ || myState == STATE_INIT) { - release(); - if (myState == STATE_FIRST_READ) { - state = STATE_INIT; - accept(first); - } - source.forEachRemaining(this); - this.action = null; - return; - } - try { + if (firstRefHolder == null) { Consumer init = x -> { - if (state == STATE_NONE) { - if (prefix != null) { - prefix.doInit(); - } - this.first = (prefix == null || prefix.state == STATE_EMPTY) ? x : prefix.first; - state = STATE_INIT; + if (first == NONE) { + first = x; } - release(); }; - source.forEachRemaining(init.andThen(this)); - this.action = null; - } finally { - release(); + source.forEachRemaining(init.andThen(x -> action.accept(mapper.apply(first, x)))); + } else { + init(); + if (firstToBeConsumed) { + firstToBeConsumed = false; + action.accept(mapper.apply(first, first)); + } + source.forEachRemaining(x -> action.accept(mapper.apply(first, x))); } } - - @Override - public Spliterator trySplit() { - if (state != STATE_NONE) - return null; - Spliterator prefix; - if (lock == null) - lock = new ReentrantLock(); - acquire(); - try { - if (state != STATE_NONE) - return null; - prefix = source.trySplit(); - if (prefix == null) - return null; - WithFirstSpliterator result = doClone(); - result.source = prefix; - return this.prefix = result; - } finally { - release(); + + private void init() { + if (!isInit()) { + doInit(); } } - + @Override public long estimateSize() { return source.estimateSize(); } - + @Override public int characteristics() { return NONNULL - | (source.characteristics() & (DISTINCT | IMMUTABLE | CONCURRENT | ORDERED | (lock == null ? SIZED : 0))); - } - - @Override - public void accept(T x) { - action.accept(mapper.apply(first, x)); + | (source.characteristics() & (DISTINCT | IMMUTABLE | CONCURRENT | ORDERED | (firstRefHolder == null ? SIZED : 0))); } } diff --git a/src/main/java/one/util/streamex/WithFirstSpliteratorOld.java b/src/main/java/one/util/streamex/WithFirstSpliteratorOld.java new file mode 100644 index 00000000..aa9c5f77 --- /dev/null +++ b/src/main/java/one/util/streamex/WithFirstSpliteratorOld.java @@ -0,0 +1,168 @@ +/* + * Copyright 2015, 2019 StreamEx contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package one.util.streamex; + +import java.util.Spliterator; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.BiFunction; +import java.util.function.Consumer; + +import one.util.streamex.Internals.CloneableSpliterator; + +/** + * @author Tagir Valeev + */ +/* package */final @Deprecated class WithFirstSpliteratorOld extends CloneableSpliterator> implements Consumer { + private static final int STATE_NONE = 0; + private static final int STATE_FIRST_READ = 1; + private static final int STATE_INIT = 2; + private static final int STATE_EMPTY = 3; + + private ReentrantLock lock; + private Spliterator source; + private WithFirstSpliteratorOld prefix; + private volatile T first; + private volatile int state = STATE_NONE; + private final BiFunction mapper; + private Consumer action; + + WithFirstSpliteratorOld(Spliterator source, BiFunction mapper) { + this.source = source; + this.mapper = mapper; + } + + private void acquire() { + if (lock != null && state == STATE_NONE) { + lock.lock(); + } + } + + private void release() { + if (lock != null && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + } + + @Override + public boolean tryAdvance(Consumer action) { + if (state == STATE_NONE) { + acquire(); + try { + doInit(); + } finally { + release(); + } + } + if (state == STATE_FIRST_READ) { + state = STATE_INIT; + action.accept(mapper.apply(first, first)); + return true; + } + if (state != STATE_INIT) + return false; + this.action = action; + boolean hasNext = source.tryAdvance(this); + this.action = null; + return hasNext; + } + + private void doInit() { + int prefixState = state; + if (prefixState != STATE_NONE) + return; + if (prefix != null) { + prefix.doInit(); + prefixState = prefix.state; + } + if (prefixState == STATE_FIRST_READ || prefixState == STATE_INIT) { + first = prefix.first; + state = STATE_INIT; + return; + } + state = source.tryAdvance(x -> first = x) ? STATE_FIRST_READ : STATE_EMPTY; + } + + @Override + public void forEachRemaining(Consumer action) { + acquire(); + int myState = state; + this.action = action; + if (myState == STATE_FIRST_READ || myState == STATE_INIT) { + release(); + if (myState == STATE_FIRST_READ) { + state = STATE_INIT; + accept(first); + } + source.forEachRemaining(this); + this.action = null; + return; + } + try { + Consumer init = x -> { + if (state == STATE_NONE) { + if (prefix != null) { + prefix.doInit(); + } + this.first = (prefix == null || prefix.state == STATE_EMPTY) ? x : prefix.first; + state = STATE_INIT; + } + release(); + }; + source.forEachRemaining(init.andThen(this)); + this.action = null; + } finally { + release(); + } + } + + @Override + public Spliterator trySplit() { + if (state != STATE_NONE) + return null; + Spliterator prefix; + if (lock == null) + lock = new ReentrantLock(); + acquire(); + try { + if (state != STATE_NONE) + return null; + prefix = source.trySplit(); + if (prefix == null) + return null; + WithFirstSpliteratorOld result = doClone(); + result.source = prefix; + return this.prefix = result; + } finally { + release(); + } + } + + @Override + public long estimateSize() { + return source.estimateSize(); + } + + @Override + public int characteristics() { + return NONNULL + | (source.characteristics() & (DISTINCT | IMMUTABLE | CONCURRENT | ORDERED | (lock == null ? SIZED : 0))); + } + + @Override + public void accept(T x) { + action.accept(mapper.apply(first, x)); + } +} From efbfd8d2a654d1ad449af6fc6bb03aa876d67dba Mon Sep 17 00:00:00 2001 From: naftalmm Date: Sat, 5 Sep 2020 14:36:43 +0300 Subject: [PATCH 2/4] update jmh to 1.25 --- .idea/benchmark.iml | 7 +++++-- .../Maven__net_sf_jopt_simple_jopt_simple_4_6.xml | 13 +++++++++++++ .../Maven__org_apache_commons_commons_math3_3_2.xml | 13 +++++++++++++ .../Maven__org_openjdk_jmh_jmh_core_1_21.xml | 12 ------------ .../Maven__org_openjdk_jmh_jmh_core_1_25.xml | 13 +++++++++++++ ...rg_openjdk_jmh_jmh_generator_annprocess_1_25.xml | 13 +++++++++++++ benchmark/pom.xml | 4 ++-- 7 files changed, 59 insertions(+), 16 deletions(-) create mode 100644 .idea/libraries/Maven__net_sf_jopt_simple_jopt_simple_4_6.xml create mode 100644 .idea/libraries/Maven__org_apache_commons_commons_math3_3_2.xml delete mode 100644 .idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_21.xml create mode 100644 .idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_25.xml create mode 100644 .idea/libraries/Maven__org_openjdk_jmh_jmh_generator_annprocess_1_25.xml diff --git a/.idea/benchmark.iml b/.idea/benchmark.iml index 7b25e1ba..e2209fda 100644 --- a/.idea/benchmark.iml +++ b/.idea/benchmark.iml @@ -4,13 +4,16 @@ - + - + + + + \ No newline at end of file diff --git a/.idea/libraries/Maven__net_sf_jopt_simple_jopt_simple_4_6.xml b/.idea/libraries/Maven__net_sf_jopt_simple_jopt_simple_4_6.xml new file mode 100644 index 00000000..18ea23b9 --- /dev/null +++ b/.idea/libraries/Maven__net_sf_jopt_simple_jopt_simple_4_6.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/libraries/Maven__org_apache_commons_commons_math3_3_2.xml b/.idea/libraries/Maven__org_apache_commons_commons_math3_3_2.xml new file mode 100644 index 00000000..dde81ab7 --- /dev/null +++ b/.idea/libraries/Maven__org_apache_commons_commons_math3_3_2.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_21.xml b/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_21.xml deleted file mode 100644 index 52e125ce..00000000 --- a/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_21.xml +++ /dev/null @@ -1,12 +0,0 @@ - - - - - - - - - - - - \ No newline at end of file diff --git a/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_25.xml b/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_25.xml new file mode 100644 index 00000000..5ada9b91 --- /dev/null +++ b/.idea/libraries/Maven__org_openjdk_jmh_jmh_core_1_25.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/libraries/Maven__org_openjdk_jmh_jmh_generator_annprocess_1_25.xml b/.idea/libraries/Maven__org_openjdk_jmh_jmh_generator_annprocess_1_25.xml new file mode 100644 index 00000000..769a50cd --- /dev/null +++ b/.idea/libraries/Maven__org_openjdk_jmh_jmh_generator_annprocess_1_25.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/benchmark/pom.xml b/benchmark/pom.xml index 46aebfe4..0f1a42a9 100644 --- a/benchmark/pom.xml +++ b/benchmark/pom.xml @@ -31,12 +31,12 @@ org.openjdk.jmh jmh-core - 1.21 + 1.25 org.openjdk.jmh jmh-generator-annprocess - 1.21 + 1.25 From f587da17570f81791e7120849a78a399a8c810c0 Mon Sep 17 00:00:00 2001 From: naftalmm Date: Sat, 5 Sep 2020 14:44:00 +0300 Subject: [PATCH 3/4] add test to increase coverage --- src/test/java/one/util/streamex/StreamExTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/test/java/one/util/streamex/StreamExTest.java b/src/test/java/one/util/streamex/StreamExTest.java index 532dd92c..d23dd61a 100644 --- a/src/test/java/one/util/streamex/StreamExTest.java +++ b/src/test/java/one/util/streamex/StreamExTest.java @@ -1953,6 +1953,8 @@ public void testWithFirst() { streamEx(() -> StreamEx.of(5, 10, 13, 12, 11), s -> assertEquals(asList("5+0", "5+5", "5+8", "5+7", "5+6"), s .get().withFirst((a, b) -> a + "+" + (b - a)).toList())); + + streamEx(StreamEx::empty, s -> assertEquals(Optional.empty(), s.get().withFirst().findAny())); } @Test From 45c60576494cab963ec501925842fcf0303429a5 Mon Sep 17 00:00:00 2001 From: naftalmm Date: Sun, 6 Sep 2020 14:06:07 +0300 Subject: [PATCH 4/4] fix global lock --- src/main/java/one/util/streamex/WithFirstSpliterator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/one/util/streamex/WithFirstSpliterator.java b/src/main/java/one/util/streamex/WithFirstSpliterator.java index 9bd69ad4..ad96d45f 100644 --- a/src/main/java/one/util/streamex/WithFirstSpliterator.java +++ b/src/main/java/one/util/streamex/WithFirstSpliterator.java @@ -27,7 +27,7 @@ * @author Tagir Valeev */ /* package */final class WithFirstSpliterator extends CloneableSpliterator> { - private static final Object lock = new Object(); + private final Object lock = new Object(); private Spliterator source; private WithFirstSpliterator prefix; private T first = none();