parallelism: streamline builder interfaces

Author: mtf90

CHANGELOG.md

@@ -9,8 +9,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ### Added
 
 * LearnLib now supports JPMS modules. All artifacts now provide a `module-info` descriptor except of the distribution artifacts (for Maven-less environments) which only provide an `Automatic-Module-Name` due to non-modular dependencies. Note that while this is a Java 9+ feature, LearnLib still supports Java 8 byte code for the remaining class files.
-* Added an `InterningMembershipOracle` (including refinements) to the `learnlib-cache` artifact that interns query responses to reduce memory consumption of large data structures. This exports the internal concepts of the DHC learner (which no longer interns query responses automatically).
 * The `ADTLearner` has been refactored to no longer use the (now-removed) `SymbolQueryOracle` but a new `AdaptiveMembershipOracle` instead which supports answering queries in parallel (thanks to [Leon Vitorovic](https://github.com/leonthalee)).
+* Added an `InterningMembershipOracle` (including refinements) to the `learnlib-cache` artifact that interns query responses to reduce memory consumption of large data structures. This exports the internal concepts of the DHC learner (which no longer interns query responses automatically).
+* `StaticParallelOracleBuilder` now supports custom executor services.
 
 ### Changed
 

filters/cache/src/test/java/de/learnlib/filter/cache/CacheConfig.java

@@ -32,7 +32,7 @@
 import de.learnlib.filter.cache.LearningCacheOracle.MooreLearningCacheOracle;
 import de.learnlib.oracle.MembershipOracle;
 import de.learnlib.oracle.ParallelOracle;
-import de.learnlib.oracle.parallelism.AbstractStaticBatchProcessor;
+import de.learnlib.oracle.parallelism.BatchProcessorDefaults;
 import de.learnlib.oracle.parallelism.ParallelOracleBuilders;
 import de.learnlib.sul.SUL;
 import de.learnlib.sul.StateLocalInputSUL;
@@ -188,9 +188,9 @@ public CacheConfig<I, D, C> apply(Alphabet<I> alphabet, M oracle) {
                 final List<C> oracles;
 
                 {
-                    List<M> suls = new ArrayList<>(AbstractStaticBatchProcessor.NUM_INSTANCES);
+                    List<M> suls = new ArrayList<>(BatchProcessorDefaults.POOL_SIZE);
 
-                    for (int i = 0; i < AbstractStaticBatchProcessor.NUM_INSTANCES; i++) {
+                    for (int i = 0; i < BatchProcessorDefaults.POOL_SIZE; i++) {
                         suls.add(oracle);
                     }
                     this.oracles = new ArrayList<>(provider.apply(alphabet, suls));

oracles/parallelism/src/main/java/de/learnlib/oracle/parallelism/AbstractDynamicBatchProcessor.java

@@ -26,13 +26,14 @@
 import de.learnlib.exception.BatchInterruptedException;
 import de.learnlib.oracle.BatchProcessor;
 import de.learnlib.oracle.ThreadPool;
-import de.learnlib.setting.LearnLibProperty;
-import de.learnlib.setting.LearnLibSettings;
 import net.automatalib.common.util.exception.ExceptionUtil;
 import org.checkerframework.checker.index.qual.NonNegative;
 
 /**
  * A batch processor that dynamically distributes queries to worker threads.
+ * <p>
+ * An incoming set of queries is split into batches of the given size. The number of batches may exceed the available
+ * threads so that they are dynamically scheduled once a job finishes.
  *
  * @param <Q>
  *         query type
@@ -42,20 +43,6 @@
 public abstract class AbstractDynamicBatchProcessor<Q, P extends BatchProcessor<Q>>
         implements ThreadPool, BatchProcessor<Q> {
 
-    public static final int BATCH_SIZE;
-    public static final int POOL_SIZE;
-    public static final PoolPolicy POOL_POLICY;
-
-    static {
-        LearnLibSettings settings = LearnLibSettings.getInstance();
-
-        int numProcessors = Runtime.getRuntime().availableProcessors();
-
-        BATCH_SIZE = settings.getInt(LearnLibProperty.PARALLEL_BATCH_SIZE_DYNAMIC, 1);
-        POOL_SIZE = settings.getInt(LearnLibProperty.PARALLEL_POOL_SIZE, numProcessors);
-        POOL_POLICY = settings.getEnumValue(LearnLibProperty.PARALLEL_POOL_POLICY, PoolPolicy.class, PoolPolicy.CACHED);
-    }
-
     private final ThreadLocal<P> threadLocalOracle;
     private final ExecutorService executor;
     private final @NonNegative int batchSize;

oracles/parallelism/src/main/java/de/learnlib/oracle/parallelism/AbstractDynamicBatchProcessorBuilder.java

@@ -15,6 +15,7 @@
  */
 package de.learnlib.oracle.parallelism;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -25,6 +26,7 @@
 
 import de.learnlib.oracle.BatchProcessor;
 import de.learnlib.oracle.ThreadPool.PoolPolicy;
+import net.automatalib.common.util.array.ArrayStorage;
 import net.automatalib.common.util.concurrent.ScalingThreadPoolExecutor;
 import org.checkerframework.checker.index.qual.NonNegative;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -45,77 +47,113 @@ public abstract class AbstractDynamicBatchProcessorBuilder<Q, P extends BatchPro
 
     private final @Nullable Supplier<? extends P> oracleSupplier;
     private final @Nullable Collection<? extends P> oracles;
+
     private ExecutorService customExecutor;
-    private @NonNegative int batchSize = AbstractDynamicBatchProcessor.BATCH_SIZE;
-    private @NonNegative int poolSize = AbstractDynamicBatchProcessor.POOL_SIZE;
-    private PoolPolicy poolPolicy = AbstractDynamicBatchProcessor.POOL_POLICY;
+    private @NonNegative int batchSize = BatchProcessorDefaults.BATCH_SIZE;
+    private @NonNegative int poolSize = BatchProcessorDefaults.POOL_SIZE;
+    private PoolPolicy poolPolicy = BatchProcessorDefaults.POOL_POLICY;
 
     public AbstractDynamicBatchProcessorBuilder(Supplier<? extends P> oracleSupplier) {
         this.oracleSupplier = oracleSupplier;
         this.oracles = null;
     }
 
     public AbstractDynamicBatchProcessorBuilder(Collection<? extends P> oracles) {
-        this(validateInputs(oracles), oracles);
-    }
-
-    // utility constructor to prevent finalizer attacks, see SEI CERT Rule OBJ-11
-    @SuppressWarnings("PMD.UnusedFormalParameter")
-    private AbstractDynamicBatchProcessorBuilder(boolean valid, Collection<? extends P> oracles) {
         this.oracles = oracles;
         this.oracleSupplier = null;
     }
 
-    private static boolean validateInputs(Collection<?> oracles) {
-        if (oracles.isEmpty()) {
-            throw new IllegalArgumentException("No oracles specified");
-        }
-        return true;
+    /**
+     * Sets the size of batches that are submitted.
+     *
+     * @param batchSize
+     *         the minimal size of batches
+     *
+     * @return {@code this}
+     */
+    public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withBatchSize(int batchSize) {
+        this.batchSize = batchSize;
+        return this;
     }
 
+    /**
+     * Sets the executor service to use for submitting batches. Note that if the builder is initialized with a
+     * collection of processors, an exception may be thrown if the thread pool tries to spawn more threads than oracles
+     * are available.
+     *
+     * @param executor
+     *         the executor to use
+     *
+     * @return {@code this}
+     */
     public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withCustomExecutor(ExecutorService executor) {
         this.customExecutor = executor;
         return this;
     }
 
-    public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withBatchSize(int batchSize) {
-        this.batchSize = batchSize;
+    /**
+     * Sets the pool policy in case the builder creates its own executor for processing batches.
+     *
+     * @param policy
+     *         the policy
+     *
+     * @return {@code this}
+     */
+    public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withPoolPolicy(PoolPolicy policy) {
+        this.poolPolicy = policy;
         return this;
     }
 
+    /**
+     * Sets the number of instances that should process batches. Note that this value is ignored if the builder has been
+     * initialized with a collection of processors in order to guarantee that no unavailable resources are accessed.
+     *
+     * @param poolSize
+     *         the number of instances to delegate batches to
+     *
+     * @return {@code this}
+     */
     public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withPoolSize(@NonNegative int poolSize) {
         this.poolSize = poolSize;
         return this;
     }
 
-    public AbstractDynamicBatchProcessorBuilder<Q, P, OR> withPoolPolicy(PoolPolicy policy) {
-        this.poolPolicy = policy;
-        return this;
-    }
-
+    /**
+     * Create the batch processor.
+     *
+     * @return the batch processor
+     */
     public OR create() {
-
         final Supplier<? extends P> supplier;
-        final ExecutorService executor;
+        final int size;
+
+        if (oracleSupplier == null) {
+            if (oracles == null || oracles.isEmpty()) {
+                throw new IllegalArgumentException("No oracles specified");
+            }
 
-        if (oracles != null) {
-            executor = Executors.newFixedThreadPool(oracles.size());
+            size = oracles.size();
             supplier = new StaticOracleProvider<>(oracles);
-        } else if (customExecutor != null) {
-            executor = customExecutor;
+        } else {
+            size = poolSize;
             supplier = oracleSupplier;
+        }
+
+        final ExecutorService executor;
+
+        if (customExecutor != null) {
+            executor = customExecutor;
         } else {
             switch (poolPolicy) {
                 case FIXED:
-                    executor = Executors.newFixedThreadPool(poolSize);
+                    executor = Executors.newFixedThreadPool(size);
                     break;
                 case CACHED:
-                    executor = new ScalingThreadPoolExecutor(0, poolSize, DEFAULT_KEEP_ALIVE_TIME, TimeUnit.SECONDS);
+                    executor = new ScalingThreadPoolExecutor(0, size, DEFAULT_KEEP_ALIVE_TIME, TimeUnit.SECONDS);
                     break;
                 default:
                     throw new IllegalStateException("Unknown pool policy: " + poolPolicy);
             }
-            supplier = oracleSupplier;
         }
 
         return buildOracle(supplier, batchSize, executor);
@@ -125,33 +163,33 @@ public OR create() {
 
     static class StaticOracleProvider<P extends BatchProcessor<?>> implements Supplier<P> {
 
-        private final P[] oracles;
+        private final ArrayStorage<P> oracles;
         private final Lock lock;
         private int idx;
 
-        @SuppressWarnings("unchecked")
-        StaticOracleProvider(Collection<? extends P> oracles) {
-            this(oracles.toArray((P[]) new BatchProcessor[oracles.size()]));
+        StaticOracleProvider(P[] oracles) {
+            this(Arrays.asList(oracles));
         }
 
-        StaticOracleProvider(P[] oracles) {
-            this.oracles = oracles;
+        StaticOracleProvider(Collection<? extends P> oracles) {
+            this.oracles = new ArrayStorage<>(oracles);
             this.lock = new ReentrantLock();
         }
 
         @Override
         public P get() {
             try {
                 lock.lock();
-                if (idx < oracles.length) {
-                    return oracles[idx++];
+                if (idx < oracles.size()) {
+                    return oracles.get(idx++);
                 }
             } finally {
                 lock.unlock();
             }
 
             throw new IllegalStateException(
-                    "The supplier should not have been called more than " + oracles.length + " times");
+                    "The executor service tried to spawn more threads than there are oracles available (" +
+                    oracles.size() + ')');
         }
     }
 }

oracles/parallelism/src/main/java/de/learnlib/oracle/parallelism/AbstractStaticBatchProcessor.java

@@ -21,15 +21,12 @@
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import de.learnlib.exception.BatchInterruptedException;
 import de.learnlib.oracle.BatchProcessor;
 import de.learnlib.oracle.ThreadPool;
-import de.learnlib.setting.LearnLibProperty;
-import de.learnlib.setting.LearnLibSettings;
-import net.automatalib.common.util.array.ArrayStorage;
+import net.automatalib.common.util.collection.CollectionUtil;
 import net.automatalib.common.util.exception.ExceptionUtil;
 import org.checkerframework.checker.index.qual.NonNegative;
 
@@ -48,37 +45,16 @@
 public abstract class AbstractStaticBatchProcessor<Q, P extends BatchProcessor<Q>>
         implements ThreadPool, BatchProcessor<Q> {
 
-    private static final int DEFAULT_MIN_BATCH_SIZE = 10;
-    public static final int MIN_BATCH_SIZE;
-    public static final int NUM_INSTANCES;
-    public static final PoolPolicy POOL_POLICY;
-
-    static {
-        LearnLibSettings settings = LearnLibSettings.getInstance();
-
-        int numCores = Runtime.getRuntime().availableProcessors();
-
-        MIN_BATCH_SIZE = settings.getInt(LearnLibProperty.PARALLEL_BATCH_SIZE_STATIC, DEFAULT_MIN_BATCH_SIZE);
-        NUM_INSTANCES = settings.getInt(LearnLibProperty.PARALLEL_POOL_SIZE, numCores);
-        POOL_POLICY = settings.getEnumValue(LearnLibProperty.PARALLEL_POOL_POLICY, PoolPolicy.class, PoolPolicy.CACHED);
-    }
-
     private final @NonNegative int minBatchSize;
-    private final ArrayStorage<P> oracles;
+    private final List<? extends P> oracles;
     private final ExecutorService executor;
 
     public AbstractStaticBatchProcessor(Collection<? extends P> oracles,
                                         @NonNegative int minBatchSize,
-                                        PoolPolicy policy) {
-
-        this.oracles = new ArrayStorage<>(oracles);
-
-        if (policy == PoolPolicy.FIXED) {
-            this.executor = Executors.newFixedThreadPool(this.oracles.size() - 1);
-        } else { // default is cached
-            this.executor = Executors.newCachedThreadPool();
-        }
+                                        ExecutorService executor) {
+        this.oracles = CollectionUtil.randomAccessList(oracles);
         this.minBatchSize = minBatchSize;
+        this.executor = executor;
     }
 
     @Override