Fixes memory leak of grouped iterator (#1907)

* Fixes memory leak of grouped iterator * Improved by throwing away unnecessary elements and by cleaning buffer before adding new elements. * Used Iterator instead of AbstractIterator * List.take/drop optimizations. See also #1910 * Added Iterator.sliding() benchmark * Based benchmark Iterators of element array * Fixed a bug/uncovered case * low-level optimization
vavr-io · Mar 12, 2017 · c8b67e8 · c8b67e8
1 parent dc18105
commit c8b67e8
Show file tree

Hide file tree

Showing 4 changed files with 174 additions and 38 deletions.
diff --git a/javaslang-benchmark/src/test/java/javaslang/collection/IteratorBenchmark.java b/javaslang-benchmark/src/test/java/javaslang/collection/IteratorBenchmark.java
@@ -0,0 +1,70 @@
+/*     / \____  _    _  ____   ______  / \ ____  __    _______
+ *    /  /    \/ \  / \/    \ /  /\__\/  //    \/  \  //  /\__\   JΛVΛSLΛNG
+ *  _/  /  /\  \  \/  /  /\  \\__\\  \  //  /\  \ /\\/ \ /__\ \   Copyright 2014-2017 Javaslang, http://javaslang.io
+ * /___/\_/  \_/\____/\_/  \_/\__\/__/\__\_/  \_//  \__/\_____/   Licensed under the Apache License, Version 2.0
+ */
+package javaslang.collection;
+
+import javaslang.JmhRunner;
+import org.junit.Test;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.Random;
+
+import static javaslang.JmhRunner.Includes.*;
+import static javaslang.JmhRunner.getRandomValues;
+
+@SuppressWarnings({ "ALL", "unchecked", "rawtypes" })
+public class IteratorBenchmark {
+
+    static final Array<Class<?>> CLASSES = Array.of(
+            Sliding.class
+    );
+
+    @Test
+    public void testAsserts() { JmhRunner.runDebugWithAsserts(CLASSES); }
+
+    public static void main(String... args) {
+        JmhRunner.runDebugWithAsserts(CLASSES);
+        JmhRunner.runNormalNoAsserts(CLASSES, JAVA, SCALA, JAVASLANG);
+    }
+
+    @State(Scope.Benchmark)
+    public static class Base {
+        @Param({ "10", "100", "1000" })
+        public int CONTAINER_SIZE;
+
+        Integer[] ELEMENTS;
+
+        scala.collection.Iterator<Integer> scalaIterator;
+        javaslang.collection.Iterator<Integer> slangIterator;
+
+        @Setup
+        public void setup() {
+            final Random random = new Random(0);
+            ELEMENTS = getRandomValues(CONTAINER_SIZE, false, random);
+            scalaIterator = (scala.collection.Iterator<Integer>) (Object) scala.collection.mutable.WrappedArray$.MODULE$.make(ELEMENTS).iterator();
+            slangIterator = Iterator.of(ELEMENTS);
+        }
+    }
+
+    public static class Sliding extends Base {
+
+        @Benchmark
+        public void scala_persistent(Blackhole bh) {
+            final scala.collection.Iterator.GroupedIterator values = scalaIterator.sliding(3, 1);
+            while (values.hasNext()) {
+                bh.consume(values.next());
+            }
+        }
+
+        @Benchmark
+        public void slang_persistent(Blackhole bh) {
+            final Iterator<Seq<Integer>> values = slangIterator.sliding(3);
+            while (values.hasNext()) {
+                bh.consume(values.next());
+            }
+        }
+    }
+}
diff --git a/javaslang/src/main/java/javaslang/collection/Iterator.java b/javaslang/src/main/java/javaslang/collection/Iterator.java
@@ -11,6 +11,7 @@
 import javaslang.Tuple3;
 import javaslang.collection.IteratorModule.ConcatIterator;
 import javaslang.collection.IteratorModule.DistinctIterator;
+import javaslang.collection.IteratorModule.GroupedIterator;
 import javaslang.control.Option;
 
 import java.math.BigDecimal;
@@ -20,6 +21,7 @@
 import static java.lang.Double.NEGATIVE_INFINITY;
 import static java.lang.Double.POSITIVE_INFINITY;
 import static java.math.RoundingMode.HALF_UP;
+import static javaslang.API.*;
 import static javaslang.collection.IteratorModule.BigDecimalHelper.areEqual;
 import static javaslang.collection.IteratorModule.BigDecimalHelper.asDecimal;
 import static javaslang.collection.IteratorModule.EmptyIterator;
@@ -1535,9 +1537,9 @@ default <C> Map<C, Iterator<T>> groupBy(Function<? super T, ? extends C> classif
 
     @Override
     default Iterator<Seq<T>> grouped(int size) {
-        return sliding(size, size);
+        return new GroupedIterator<>(this, size, size);
     }
-
+    
     @Override
     default boolean hasDefiniteSize() {
         return false;
@@ -1827,37 +1829,9 @@ default Iterator<Seq<T>> sliding(int size) {
 
     @Override
     default Iterator<Seq<T>> sliding(int size, int step) {
-        if (size <= 0 || step <= 0) {
-            throw new IllegalArgumentException("size: " + size + " or step: " + step + " not positive");
-        }
-        if (!hasNext()) {
-            return empty();
-        } else {
-            final Stream<T> source = Stream.ofAll(this);
-            return new AbstractIterator<Seq<T>>() {
-                private Stream<T> that = source;
-                private IndexedSeq<T> next = null;
-
-                @Override
-                public boolean hasNext() {
-                    while (next == null && !that.isEmpty()) {
-                        final Tuple2<Stream<T>, Stream<T>> split = that.splitAt(size);
-                        next = split._1.toVector();
-                        that = split._2.isEmpty() ? Stream.empty() : that.drop(step);
-                    }
-                    return next != null;
-                }
-
-                @Override
-                public IndexedSeq<T> getNext() {
-                    final IndexedSeq<T> result = next;
-                    next = null;
-                    return result;
-                }
-            };
-        }
+        return new GroupedIterator<>(this, size, step);
     }
-
+    
     @Override
     default Tuple2<Iterator<T>, Iterator<T>> span(Predicate<? super T> predicate) {
         Objects.requireNonNull(predicate, "predicate is null");
@@ -2089,6 +2063,78 @@ public String toString() {
         }
     }
 
+    final class GroupedIterator<T> implements Iterator<Seq<T>> {
+
+        private final Iterator<T> that;
+        private final int size;
+        private final int step;
+        private final int gap;
+        private final int preserve;
+
+        private Object[] buffer;
+
+        GroupedIterator(Iterator<T> that, int size, int step) {
+            if (size < 1 || step < 1) {
+                throw new IllegalArgumentException("size (" + size + ") and step (" + step + ") must both be positive");
+            }
+            this.that = that;
+            this.size = size;
+            this.step = step;
+            this.gap = Math.max(step - size, 0);
+            this.preserve = Math.max(size - step, 0);
+            this.buffer = take(that, new Object[size], 0, size);
+        }
+
+        @Override
+        public boolean hasNext() {
+            return buffer.length > 0;
+        }
+
+        @Override
+        public Seq<T> next() {
+            if (buffer.length == 0) {
+                throw new NoSuchElementException();
+            }
+            final Object[] result = buffer;
+            if (that.hasNext()) {
+                buffer = new Object[size];
+                if (preserve > 0) {
+                    System.arraycopy(result, step, buffer, 0, preserve);
+                }
+                if (gap > 0) {
+                    drop(that, gap);
+                    buffer = take(that, buffer, preserve, size);
+                } else {
+                    buffer = take(that, buffer, preserve, step);
+                }
+            } else {
+                buffer = new Object[0];
+            }
+            return Array.wrap(result);
+        }
+
+        private static void drop(Iterator<?> source, int count) {
+            for (int i = 0; i < count && source.hasNext(); i++) {
+                source.next();
+            }
+        }
+
+        private static Object[] take(Iterator<?> source, Object[] target, int offset, int count) {
+            int i = offset;
+            while (i < count + offset && source.hasNext()) {
+                target[i] = source.next();
+                i++;
+            }
+            if (i < target.length) {
+                final Object[] result = new Object[i];
+                System.arraycopy(target, 0, result, 0, i);
+                return result;
+            } else {
+                return target;
+            }
+        }
+    }
+
     final class BigDecimalHelper {
 
         @GwtIncompatible("Math::nextDown is not implemented")

diff --git a/javaslang/src/main/java/javaslang/collection/List.java b/javaslang/src/main/java/javaslang/collection/List.java
@@ -720,6 +720,12 @@ default <U> List<T> distinctBy(Function<? super T, ? extends U> keyExtractor) {
 
     @Override
     default List<T> drop(int n) {
+        if (n <= 0) {
+            return this;
+        }
+        if (n >= size()) {
+            return empty();
+        }
         List<T> list = this;
         for (long i = n; i > 0 && !list.isEmpty(); i--) {
             list = list.tail();
@@ -1390,12 +1396,12 @@ default Option<List<T>> tailOption() {
 
     @Override
     default List<T> take(int n) {
-        if (n >= length()) {
-            return this;
-        }
         if (n <= 0) {
             return empty();
         }
+        if (n >= length()) {
+            return this;
+        }
         List<T> result = Nil.instance();
         List<T> list = this;
         for (int i = 0; i < n; i++, list = list.tail()) {
@@ -1406,12 +1412,12 @@ default List<T> take(int n) {
 
     @Override
     default List<T> takeRight(int n) {
-        if (n >= length()) {
-            return this;
-        }
         if (n <= 0) {
             return empty();
         }
+        if (n >= length()) {
+            return this;
+        }
         return reverse().take(n).reverse();
     }
 

diff --git a/javaslang/src/test/java/javaslang/collection/AbstractTraversableTest.java b/javaslang/src/test/java/javaslang/collection/AbstractTraversableTest.java
@@ -1797,6 +1797,20 @@ public void shouldSlide4ElementsBySize5AndStep3() {
         assertThat(actual).isEqualTo(expected);
     }
 
+    @Test
+    public void shouldSlide7ElementsBySize1AndStep3() {
+        final List<Traversable<Integer>> actual = of(1, 2, 3, 4, 5, 6 ,7).sliding(1, 3).toList().map(Vector::ofAll);
+        final List<Traversable<Integer>> expected = List.of(Vector.of(1), Vector.of(4), Vector.of(7));
+        assertThat(actual).isEqualTo(expected);
+    }
+
+    @Test
+    public void shouldSlide7ElementsBySize2AndStep3() {
+        final List<Traversable<Integer>> actual = of(1, 2, 3, 4, 5, 6 ,7).sliding(2, 3).toList().map(Vector::ofAll);
+        final List<Traversable<Integer>> expected = List.of(Vector.of(1, 2), Vector.of(4, 5), Vector.of(7));
+        assertThat(actual).isEqualTo(expected);
+    }
+
     // -- span
 
     @Test