Refactors Fulgora and ScanJob (plus executors) to correspond to TP3 s…

…etup and parallel execution semantics. Specifically, each thread now uses its own instance of ScanJob or VertexScanJob - cloning if necessary. Rather than using a ThreadPool the StandardScannerExecutor maintains the threads specifically.

There is now the notion of a workBlockSize - i.e. the number of rows that are worked on in one block of work. We use one instance of ScanJob per block and call workerIterationStart/End before/after such block. This provides better control over resources needed to execute a particular ScanJob.

Fixes #857. Fixes #865.

titan-core/src/main/java/com/thinkaurelius/titan/diskstorage/Backend.java

@@ -349,7 +349,8 @@ private StandardScanner.Builder buildStoreIndexScanJob(String storeName) {
                 .setTimestampProvider(provider)
                 .setJobConfiguration(jobConfig)
                 .setGraphConfiguration(configuration)
-                .setNumProcessingThreads(1);
+                .setNumProcessingThreads(1)
+                .setWorkBlockSize(10000);
     }
 
     public StandardScanner.ScanResult getScanJobStatus(Object jobId) {

titan-core/src/main/java/com/thinkaurelius/titan/diskstorage/keycolumnvalue/scan/ScanJob.java

@@ -10,30 +10,40 @@
 import java.util.function.Predicate;
 
 /**
- * A computation over edgestore entries.
+ * A global computation over
  *
  * @author Matthias Broecheler (me@matthiasb.com)
  */
-public interface ScanJob {
+public interface ScanJob extends Cloneable {
 
     /**
-     * Invoked prior to any other method on this {@code ScanJob} instance.
-     * This will only be called once per instance.
+     * Invoked before a block of computation (i.e. multiple process() calls) is handed to this particular ScanJob.
+     * Can be used to initialize the iteration. This method is called exactly once for each before a block of computation.
+     * This method is semantically aligned with {@link com.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationStart()}
      *
-     * @param config
-     * @param metrics
+     * This method may not be called if there is no data to be processed. Correspondingly, the end method won't be called either.
+     *
+     * No-op default implementation.
+     *
+     * @param jobConfiguration configuration for this particular job
+     * @param graphConfiguration configuration options for the entire graph against which this job is executed
+     * @param metrics {@link com.thinkaurelius.titan.diskstorage.keycolumnvalue.scan.ScanMetrics} for this job
      */
-    public default void setup(Configuration jobConfiguration,
-                  Configuration graphConfiguration, ScanMetrics metrics) {}
+    public default void workerIterationStart(Configuration jobConfiguration,
+                                             Configuration graphConfiguration, ScanMetrics metrics) {}
 
     /**
-     * After this method is invoked, no additional method calls on this
-     * {@code ScanJob} instance are permitted.
-     * This will only be called once per instance.
+     * Invoked after a block of computation (i.e. multiple process() calls) is handed to this particular ScanJob.
+     * Can be used to close any resources held by this job. This method is called exactly once for each after a block of computation.
+     * This method is semantically aligned with {@link com.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationEnd()}
      *
-     * @param metrics
+     * This method may not be called if there is no data to be processed. Correspondingly, the start method won't be called either.
+     *
+     * No-op default implementation.
+     *
+     * @param metrics {@link com.thinkaurelius.titan.diskstorage.keycolumnvalue.scan.ScanMetrics} for this job
      */
-    public default void teardown(ScanMetrics metrics) {}
+    public default void workerIterationEnd(ScanMetrics metrics) {}
 
     /**
      * Run this {@code ScanJob}'s computation on the supplied row-key and entries.
@@ -109,4 +119,12 @@ public default Predicate<StaticBuffer> getKeyFilter() {
         return b -> true; //No filter by default
     }
 
+    /**
+     * Returns a clone of this ScanJob. The clone will not yet be initialized for computation but all of
+     * its internal state (if any) must match that of the original copy.
+     *
+     * @return A clone of this {@link com.thinkaurelius.titan.diskstorage.keycolumnvalue.scan.ScanJob}
+     */
+    public ScanJob clone();
+
 }

titan-core/src/main/java/com/thinkaurelius/titan/diskstorage/keycolumnvalue/scan/StandardScanner.java

@@ -72,8 +72,11 @@ public ScanResult getRunningJob(Object jobId) {
 
     public class Builder {
 
+        private static final int DEFAULT_WORKBLOCK_SIZE = 10000;
+
         private ScanJob job;
         private int numProcessingThreads;
+        private int workBlockSize;
         private TimestampProvider times;
         private Configuration graphConfiguration;
         private Configuration jobConfiguration;
@@ -83,6 +86,7 @@ public class Builder {
 
         private Builder() {
             numProcessingThreads = 1;
+            workBlockSize = DEFAULT_WORKBLOCK_SIZE;
             job = null;
             times = null;
             graphConfiguration = Configuration.EMPTY;
@@ -99,6 +103,12 @@ public Builder setNumProcessingThreads(int numThreads) {
             return this;
         }
 
+        public Builder setWorkBlockSize(int size) {
+            Preconditions.checkArgument(size>0, "Need to specify a positive work block size: %s",size);
+            this.workBlockSize = size;
+            return this;
+        }
+
         public Builder setTimestampProvider(TimestampProvider times) {
             Preconditions.checkArgument(times!=null);
             this.times=times;
@@ -178,7 +188,7 @@ public ScanResult execute() throws BackendException {
             openStores.add(kcvs);
             try {
                 StandardScannerExecutor executor = new StandardScannerExecutor(job, finishJob, kcvs, storeTx,
-                        manager.getFeatures(), numProcessingThreads, jobConfiguration, graphConfiguration);
+                        manager.getFeatures(), numProcessingThreads, workBlockSize, jobConfiguration, graphConfiguration);
                 addJob(jobId,executor);
                 new Thread(executor).start();
                 return executor;

titan-core/src/main/java/com/thinkaurelius/titan/diskstorage/keycolumnvalue/scan/StandardScannerExecutor.java

@@ -9,6 +9,7 @@
 import com.thinkaurelius.titan.diskstorage.util.RecordIterator;
 import com.thinkaurelius.titan.diskstorage.util.StaticArrayEntry;
 import com.thinkaurelius.titan.diskstorage.util.StaticArrayEntryList;
+import com.thinkaurelius.titan.util.system.Threads;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,6 +37,7 @@ class StandardScannerExecutor extends AbstractFuture<ScanMetrics> implements Sta
     private final StoreTransaction storeTx;
     private final KeyColumnValueStore store;
     private final int numProcessors;
+    private final int workBlockSize;
     private final Configuration jobConfiguration;
     private final Configuration graphConfiguration;
     private final ScanMetrics metrics;
@@ -48,18 +50,19 @@ class StandardScannerExecutor extends AbstractFuture<ScanMetrics> implements Sta
     private List<BlockingQueue<SliceResult>> dataQueues;
     private DataPuller[] pullThreads;
 
-
     StandardScannerExecutor(final ScanJob job, final Consumer<ScanMetrics> finishJob,
                             final KeyColumnValueStore store, final StoreTransaction storeTx,
                             final StoreFeatures storeFeatures,
-                            final int numProcessors, final Configuration jobConfiguration,
+                            final int numProcessors, final int workBlockSize,
+                            final Configuration jobConfiguration,
                             final Configuration graphConfiguration) throws BackendException {
         this.job = job;
         this.finishJob = finishJob;
         this.store = store;
         this.storeTx = storeTx;
         this.storeFeatures = storeFeatures;
         this.numProcessors = numProcessors;
+        this.workBlockSize = workBlockSize;
         this.jobConfiguration = jobConfiguration;
         this.graphConfiguration = graphConfiguration;
 
@@ -80,7 +83,7 @@ private final DataPuller addDataPuller(SliceQuery sq, StoreTransaction stx) thro
     @Override
     public void run() {
         try {
-            job.setup(jobConfiguration, graphConfiguration, metrics);
+            job.workerIterationStart(jobConfiguration, graphConfiguration, metrics);
 
             queries = job.getQueries();
             numQueries = queries.size();
@@ -104,13 +107,19 @@ public void run() {
         }  catch (Throwable e) {
             log.error("Exception trying to setup the job:", e);
             cleanupSilent();
-            job.teardown(metrics);
+            job.workerIterationEnd(metrics);
             setException(e);
             return;
         }
 
-        ThreadPoolExecutor processor = new ThreadPoolExecutor(numProcessors, numProcessors, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(QUEUE_SIZE));
-        processor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        BlockingQueue<Row> processorQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
+
+        Processor[] processors = new Processor[numProcessors];
+        for (int i=0;i<processors.length;i++) {
+            processors[i]= new Processor(job.clone(),processorQueue);
+            processors[i].start();
+        }
+
         try {
             SliceResult[] currentResults = new SliceResult[numQueries];
             while (!interrupted) {
@@ -142,8 +151,7 @@ public void run() {
                     }
                     queryResults.put(query,entries);
                 }
-                processor.submit(new RowProcessor(key, queryResults));
-
+                processorQueue.put(new Row(key, queryResults));
             }
 
             for (int i = 0; i < pullThreads.length; i++) {
@@ -154,12 +162,13 @@ public void run() {
                 }
             }
 
-            processor.shutdown();
-            processor.awaitTermination(TIMEOUT_MS,TimeUnit.MILLISECONDS);
-            if (!processor.isTerminated()) log.error("Processor did not terminate in time");
+            for (int i=0; i<processors.length;i++) {
+                processors[i].finish();
+            }
+            if (!Threads.waitForCompletion(processors,TIMEOUT_MS)) log.error("Processor did not terminate in time");
 
             cleanup();
-            job.teardown(metrics);
+            job.workerIterationEnd(metrics);
 
             if (interrupted) {
                 setException(new InterruptedException("Scanner got interrupted"));
@@ -169,10 +178,10 @@ public void run() {
             }
         } catch (Throwable e) {
             log.error("Exception occured during job execution: {}",e);
-            job.teardown(metrics);
+            job.workerIterationEnd(metrics);
             setException(e);
         } finally {
-            processor.shutdownNow();
+            Threads.terminate(processors);
             cleanupSilent();
         }
     }
@@ -209,28 +218,75 @@ public ScanMetrics getIntermediateResult() {
         return metrics;
     }
 
-    private class RowProcessor implements Runnable {
+    private static class Row {
 
-        private final StaticBuffer key;
-        private final Map<SliceQuery,EntryList> entries;
+        final StaticBuffer key;
+        final Map<SliceQuery,EntryList> entries;
 
-        private RowProcessor(StaticBuffer key, Map<SliceQuery, EntryList> entries) {
+        private Row(StaticBuffer key, Map<SliceQuery, EntryList> entries) {
             this.key = key;
             this.entries = entries;
         }
+    }
+
+
+
+    private class Processor extends Thread {
+
+        private ScanJob job;
+        private final BlockingQueue<Row> processorQueue;
+
+        private volatile boolean finished;
+        private int numProcessed;
+
+
+        private Processor(ScanJob job, BlockingQueue<Row> processorQueue) {
+            this.job = job;
+            this.processorQueue = processorQueue;
+
+            this.finished = false;
+            this.numProcessed = 0;
+        }
 
         @Override
         public void run() {
             try {
-                job.process(key,entries,metrics);
-                metrics.increment(ScanMetrics.Metric.SUCCESS);
-            } catch (Throwable ex) {
-                log.error("Exception processing row ["+key+"]: ",ex);
-                metrics.increment(ScanMetrics.Metric.FAILURE);
+                job.workerIterationStart(jobConfiguration, graphConfiguration, metrics);
+                while (!finished) {
+                    Row row;
+                    while ((row=processorQueue.poll(100,TimeUnit.MILLISECONDS))!=null) {
+                        if (numProcessed>=workBlockSize) {
+                            //Setup new chunk of work
+                            job.workerIterationEnd(metrics);
+                            job = job.clone();
+                            job.workerIterationStart(jobConfiguration, graphConfiguration, metrics);
+                            numProcessed=0;
+                        }
+                        try {
+                            job.process(row.key,row.entries,metrics);
+                            metrics.increment(ScanMetrics.Metric.SUCCESS);
+                        } catch (Throwable ex) {
+                            log.error("Exception processing row ["+row.key+"]: ",ex);
+                            metrics.increment(ScanMetrics.Metric.FAILURE);
+                        }
+                        numProcessed++;
+                    }
+                }
+            } catch (InterruptedException e) {
+                log.error("Processing thread interrupted while waiting on queue or processing data", e);
+            } catch (Throwable e) {
+                log.error("Unexpected error processing data: {}",e);
+            } finally {
+                job.workerIterationEnd(metrics);
             }
         }
+
+        public void finish() {
+            this.finished=true;
+        }
     }
 
+
     private static class DataPuller extends Thread {
 
         private final BlockingQueue<SliceResult> queue;

titan-core/src/main/java/com/thinkaurelius/titan/graphdb/olap/VertexJobConverter.java

@@ -53,10 +53,16 @@ public class VertexJobConverter implements ScanJob {
     protected VertexJobConverter(TitanGraph graph, VertexScanJob job) {
         Preconditions.checkArgument(job!=null);
         this.graph = new GraphProvider();
-        this.graph.setGraph(graph);
+        if (graph!=null) this.graph.setGraph(graph);
         this.job = job;
     }
 
+    protected VertexJobConverter(VertexJobConverter copy) {
+        this.graph = new GraphProvider();
+        if (copy.graph.isProvided()) this.graph.setGraph(copy.graph.get());
+        this.job = copy.job.clone();
+    }
+
     public static ScanJob convert(TitanGraph graph, VertexScanJob vertexJob) {
         return new VertexJobConverter(graph,vertexJob);
     }
@@ -66,7 +72,7 @@ public static ScanJob convert(VertexScanJob vertexJob) {
     }
 
     @Override
-    public void setup(Configuration jobConfig, Configuration graphConfig, ScanMetrics metrics) {
+    public void workerIterationStart(Configuration jobConfig, Configuration graphConfig, ScanMetrics metrics) {
         graph.initializeGraph(graphConfig);
         idManager = graph.get().getIDManager();
         StandardTransactionBuilder txb = graph.get().buildTransaction().readOnly();
@@ -76,7 +82,7 @@ public void setup(Configuration jobConfig, Configuration graphConfig, ScanMetric
         txb.vertexCacheSize(500);
         try {
             tx = (StandardTitanTx)txb.start();
-            job.setup(graph.get(), jobConfig, metrics);
+            job.workerIterationStart(graph.get(), jobConfig, metrics);
         } catch (Throwable e) {
             close();
             throw e;
@@ -90,8 +96,8 @@ private void close() {
     }
 
     @Override
-    public void teardown(ScanMetrics metrics) {
-        job.teardown(metrics);
+    public void workerIterationEnd(ScanMetrics metrics) {
+        job.workerIterationEnd(metrics);
         close();
     }
 
@@ -154,6 +160,11 @@ public Predicate<StaticBuffer> getKeyFilter() {
         };
     }
 
+    @Override
+    public VertexJobConverter clone() {
+        return new VertexJobConverter(this);
+    }
+
     protected long getVertexId(StaticBuffer key) {
         return idManager.getKeyID(key);
     }
@@ -182,6 +193,10 @@ public void close() {
             }
         }
 
+        public boolean isProvided() {
+            return provided;
+        }
+
         public final StandardTitanGraph get() {
             Preconditions.checkState(graph!=null);
             return graph;