#80 CQuery: parallel spliterator

assira.core/src/main/java/net/ranides/assira/collection/query/support/BaseArrayParalell.java

@@ -34,8 +34,10 @@ public class BaseArrayParalell {
         int blocksize = Math.max(1, length / threads);
         List<Callable<Void>> tasks = new ArrayList<>(threads);
 
-        for (int offset = begin; offset < end; offset += blocksize) {
+        int x = 0;
+        for (int offset = begin, bound=end-blocksize; offset < bound; offset += blocksize) {
             int let_offset = offset;
+            System.out.printf(">>>>>    ARRAY TASK[%d]: %d-%d%n", ++x, let_offset, let_offset + blocksize);
             tasks.add(() -> {
                 loop(array, begin, let_offset, let_offset + blocksize, consumer);
                 return null;
@@ -44,11 +46,13 @@ public class BaseArrayParalell {
 
         int tail = length % threads;
         if (tail != 0) {
+            System.out.println(">>>>>    ARRAY TASK TAIL: " + (tail));
             tasks.add(() -> {
                 loop(array, begin, end - tail, end, consumer);
                 return null;
             });
         }
+        System.out.println(">>>>> ARRAY: " + tasks.size());
         ForkJoinPool.commonPool().invokeAll(tasks);
     }
 

assira.core/src/main/java/net/ranides/assira/collection/query/support/BaseListParalell.java

@@ -38,6 +38,7 @@ public class BaseListParalell {
                 return null;
             });
         }
+        System.out.println(">>>>> LIST: " + tasks.size());
         ForkJoinPool.commonPool().invokeAll(tasks);
     }
 

assira.core/src/main/java/net/ranides/assira/collection/query/support/BaseSpliteratorParalell.java

@@ -33,7 +33,7 @@ public class BaseSpliteratorParalell {
         }
 
         ForkJoinPool executor = ForkJoinPool.commonPool();
-        int splits = MathUtils.log2ceil(executor.getParallelism());
+        int splits = estimateSplits(executor, that);
         List<Callable<Void>> tasks = new ArrayList<>();
         try {
             forEachSplit(splits, tasks, that, 0, consumer);
@@ -73,7 +73,7 @@ public class BaseSpliteratorParalell {
         }
 
         ForkJoinPool executor = ForkJoinPool.commonPool();
-        int splits = MathUtils.log2ceil(executor.getParallelism());
+        int splits = estimateSplits(executor, that);
         List<Callable<Void>> tasks = new ArrayList<>();
         BaseState.Terminator terminator = BaseState.newTerminator(true);
         try {
@@ -119,7 +119,7 @@ public class BaseSpliteratorParalell {
         }
 
         ForkJoinPool executor = ForkJoinPool.commonPool();
-        int splits = MathUtils.log2ceil(executor.getParallelism());
+        int splits = estimateSplits(executor, that);
         List<Callable<Integer>> tasks = new ArrayList<>();
         try {
             sizeSplit(splits, tasks, that);
@@ -149,4 +149,28 @@ public class BaseSpliteratorParalell {
         }
     }
 
+    private static int estimateSplits(ForkJoinPool executor, Spliterator<?> that) {
+        // we can't split spliterators into groups with specified size
+        // which mean we can't distribute elements evenly
+        // we can do 2 things:
+        //  - split into number of tasks lower than executor
+        //  - split into number of tasks SIGNIFICANTLY bigger than executor
+
+        // this estimation assumes that 3 additional splits will make groups
+        // 8 times smaller, which will eliminate disproportion between groups
+        //
+        // effectively, it will introduce 1/8 overhead at most, which means ~12%
+
+        int splits = MathUtils.log2floor(executor.getParallelism());
+
+        long n = that.getExactSizeIfKnown();
+        for(int i = 0; i<3; i++, splits++) {
+            if(n / (1L << splits) <= 3) {
+                break;
+            }
+        }
+        return splits;
+    }
+
+
 }

assira.core/src/test/java/net/ranides/assira/collection/query/base/CQArrayBenchmark.java

@@ -0,0 +1,55 @@
+package net.ranides.assira.collection.query.base;
+
+import net.ranides.assira.collection.lists.IntRange;
+import net.ranides.assira.collection.query.CQuery;
+import net.ranides.assira.time.TimeMetric;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.stream.StreamSupport;
+
+public class CQArrayBenchmark {
+
+    public static void main(String[] args) {
+        Integer[] range = new IntRange(0, 1000).toArray(new Integer[0]);
+
+//        compare("CQArray", range, new CQArray<>(range));
+//        compare("CQArraySupplier", range, CQArraySupplier.from(() -> range));
+//
+//        compare("CQList", range, new CQList<>(Arrays.asList(range)));
+//        compare("CQListSupplier", range, CQListSupplier.from(() -> Arrays.asList(range)));
+//
+
+        compare("spliterator", range, CQSpliterator.from(() -> Arrays.asList(range).spliterator()));
+        compare("stream.spliterator", range, CQSpliterator.from(() -> Arrays.asList(range).stream().spliterator()));
+        compare("stream.parallel.spliterator", range, CQSpliterator.from(() -> Arrays.asList(range).stream().parallel().spliterator()));
+    }
+
+    private static <T> void compare(String label, T[] range, CQuery<T> query) {
+        TimeMetric tmJDK = new TimeMetric();
+
+        StreamSupport.stream(Arrays.spliterator(range), true).forEach(v -> {
+            sleep(100);
+        });
+        long timeJDK = tmJDK.stop();
+
+        TimeMetric tmQuery = new TimeMetric();
+        query.parallel().forEach((i,v) -> {
+            sleep(100);
+        });
+        long timeQuery = tmQuery.stop();
+
+        long timeDt = timeQuery - timeJDK;
+        System.out.printf("%-30s Time = %8dms %8dms %8dms %8d%%%n",
+            label, timeJDK, timeQuery, timeDt, 100* (timeDt) / timeQuery
+        );
+    }
+
+    private static void sleep(int ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}