Merge pull request #392 from JLLeitschuh/feat/pipelineInOneThread

Pipeline run in own thread

Author: ThomasJClark

build.gradle

@@ -57,6 +57,8 @@ allprojects {
         testCompile group: 'net.jodah', name: 'concurrentunit', version: '0.4.2'
         testCompile group: 'org.hamcrest', name: 'hamcrest-all', version: '1.3'
         testCompile group: 'junit', name: 'junit', version: '4.12'
+        testCompile group: 'com.google.truth', name: 'truth', version: '0.28'
+        testCompile group: 'com.google.guava', name: 'guava-testlib', version: '19.0'
     }
 
     version = getVersionName()
@@ -122,6 +124,7 @@ project(":core") {
     }
 
     dependencies {
+        compile group: 'com.google.code.findbugs', name: 'jsr305', version: '3.0.0'
         compile group: 'org.bytedeco', name: 'javacv', version: '1.1'
         compile group: 'org.bytedeco.javacpp-presets', name: 'opencv', version: '3.0.0-1.1'
         compile group: 'org.bytedeco.javacpp-presets', name: 'opencv', version: '3.0.0-1.1', classifier: os

core/src/main/java/edu/wpi/grip/core/Main.java

@@ -26,6 +26,8 @@ public class Main {
     @Inject
     private Project project;
     @Inject
+    private PipelineRunner pipelineRunner;
+    @Inject
     private EventBus eventBus;
     @Inject
     private Logger logger;
@@ -58,6 +60,7 @@ public void start(String[] args) throws IOException, InterruptedException {
         // Open a project from a .grip file specified on the command line
         project.open(new File(projectPath));
 
+        pipelineRunner.startAsync();
 
         // This is done in order to indicate to the user using the deployment UI that this is running
         logger.log(Level.INFO, "SUCCESS! The project is running in headless mode!");

core/src/main/java/edu/wpi/grip/core/Operation.java

@@ -64,4 +64,18 @@ default void perform(InputSocket<?>[] inputs, OutputSocket<?>[] outputs, Optiona
     default void perform(InputSocket<?>[] inputs, OutputSocket<?>[] outputs) {
         throw new UnsupportedOperationException("Perform was not overridden");
     }
+
+    /**
+     * Allows the step to clean itself up when removed from the pipeline.
+     * This should only be called by {@link Step#setRemoved()} to ensure correct synchronization.
+     *
+     * @param inputs  An array obtained from {@link #createInputSockets(EventBus)}. The caller can set the value of
+     *                each socket to an actual parameter for the operation.
+     * @param outputs An array obtained from {@link #createOutputSockets(EventBus)}. The outputs of the operation will
+     *                be stored in these sockets.
+     * @param data    Optional data to be passed to the operation
+     */
+    default void cleanUp(InputSocket<?>[] inputs, OutputSocket<?>[] outputs, Optional<?> data) {
+        /* no-op */
+    }
 }

core/src/main/java/edu/wpi/grip/core/Pipeline.java

@@ -1,5 +1,6 @@
 package edu.wpi.grip.core;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.eventbus.EventBus;
 import com.google.common.eventbus.Subscribe;
 import com.google.inject.Singleton;
@@ -11,6 +12,11 @@
 
 import javax.inject.Inject;
 import java.util.*;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -35,7 +41,14 @@ public class Pipeline {
     @XStreamOmitField
     private NTManager ntManager;
 
+    /*
+     * We have separate locks for sources and steps because we don't want to
+     * block access to both resources when only one is in use.
+     */
+
+    private transient final ReadWriteLock sourceLock = new ReentrantReadWriteLock();
     private final List<Source> sources = new ArrayList<>();
+    private transient ReadWriteLock stepLock = new ReentrantReadWriteLock();
     private final List<Step> steps = new ArrayList<>();
     private final Set<Connection> connections = new HashSet<>();
     private ProjectSettings settings = new ProjectSettings();
@@ -44,27 +57,108 @@ public class Pipeline {
      * Remove everything in the pipeline
      */
     public void clear() {
-        this.steps.stream().collect(Collectors.toList()).forEach(this::removeStep);
-
+        getSteps().forEach(this::removeStep);
+        // We collect the list first because the event modifies the list
         this.sources.stream()
                 .map(SourceRemovedEvent::new)
                 .collect(Collectors.toList())
                 .forEach(this.eventBus::post);
     }
 
+    private final <R> R readSourcesSafely(Function<List<Source>, R> sourceListFunction) {
+        return accessSafely(sourceLock.readLock(), Collections.unmodifiableList(sources), sourceListFunction);
+    }
+
+    /**
+     * Returns a snapshot of all of the sources in the pipeline.
+     *
+     * @return an Immutable copy of the sources at the current point in the pipeline.
+     * @see <a href="https://youtu.be/ZeO_J2OcHYM?t=16m35s">Why we use ImmutableList return type</a>
+     */
+    public final ImmutableList<Source> getSources() {
+        return readSourcesSafely(ImmutableList::copyOf);
+    }
+
     /**
-     * @return The unmodifiable list of sources for inputs to the algorithm
-     * @see Source
+     * @param stepListFunction The function to read the steps with.
+     * @param <R>              The return type of the function
+     * @return The value returned by the function.
      */
-    public List<Source> getSources() {
-        return Collections.unmodifiableList(this.sources);
+    private final <R> R readStepsSafely(Function<List<Step>, R> stepListFunction) {
+        return accessSafely(stepLock.readLock(), Collections.unmodifiableList(steps), stepListFunction);
     }
 
     /**
-     * @return The unmodifiable list of steps in the computer vision algorithm
+     * Returns a snapshot of all of the steps in the pipeline.
+     *
+     * @return an Immutable copy of the steps at the current point in the pipeline.
+     * @see <a href="https://youtu.be/ZeO_J2OcHYM?t=16m35s">Why we use ImmutableList return type</a>
      */
-    public List<Step> getSteps() {
-        return Collections.unmodifiableList(this.steps);
+    public final ImmutableList<Step> getSteps() {
+        return readStepsSafely(ImmutableList::copyOf);
+    }
+
+    /*
+     * These methods should not be made public.
+     * If you do so you are making a poor design decision and should move whatever you are trying to do into
+     * this class.
+     */
+
+    /**
+     * @param stepListWriterFunction A function that modifies the step list passed to the operation.
+     * @param <R>                    The return type of the function
+     * @return The value returned by the function.
+     */
+    private <R> R writeStepsSafely(Function<List<Step>, R> stepListWriterFunction) {
+        return accessSafely(stepLock.writeLock(), steps, stepListWriterFunction);
+    }
+
+    /**
+     * @param stepListWriterConsumer A consumer that can modify the list that is passed to it.
+     */
+    private void writeStepsSafelyConsume(Consumer<List<Step>> stepListWriterConsumer) {
+        writeStepsSafely(stepList -> {
+            stepListWriterConsumer.accept(stepList);
+            return null;
+        });
+    }
+
+    private <R> R writeSourcesSafely(Function<List<Source>, R> sourceListWriterFunction) {
+        return accessSafely(sourceLock.writeLock(), sources, sourceListWriterFunction);
+    }
+
+    private void writeSourcesSafelyConsume(Consumer<List<Source>> sourceListWriterFunction) {
+        writeSourcesSafely(sources -> {
+            sourceListWriterFunction.accept(sources);
+            return null;
+        });
+    }
+
+    /*
+     * End of methods that should not be made public
+     */
+
+    /**
+     * Locks the resource with the specified lock and performs the function.
+     * When the function is complete then the lock unlocked again.
+     *
+     * @param lock         The lock for the given resource
+     * @param list         The list that will be accessed while the resource is locked
+     * @param listFunction The function that either modifies or accesses the list
+     * @param <T>          The type of list
+     * @param <R>          The return value for the function
+     * @return The value returned by the list function
+     */
+    private static <T, R> R accessSafely(Lock lock, List<T> list, Function<List<T>, R> listFunction) {
+        final R returnValue;
+        lock.lock();
+        try {
+            returnValue = listFunction.apply(list);
+        } finally {
+            // Ensure that no matter what may get thrown while reading the steps we unlock
+            lock.unlock();
+        }
+        return returnValue;
     }
 
     /**
@@ -132,43 +226,53 @@ public boolean canConnect(Socket socket1, Socket socket2) {
     /**
      * @return true if the step1 is before step2 in the pipeline
      */
-    private synchronized boolean isBefore(Step step1, Step step2) {
-        return this.steps.indexOf(step1) < this.steps.indexOf(step2);
+    private boolean isBefore(Step step1, Step step2) {
+        return readStepsSafely(steps -> steps.indexOf(step1) < steps.indexOf(step2));
     }
 
     @Subscribe
     public void onSourceAdded(SourceAddedEvent event) {
-        this.sources.add(event.getSource());
+        writeSourcesSafelyConsume(sources -> {
+            sources.add(event.getSource());
+        });
     }
 
     @Subscribe
     public void onSourceRemoved(SourceRemovedEvent event) {
-        this.sources.remove(event.getSource());
+        writeSourcesSafelyConsume(sources -> {
+            sources.remove(event.getSource());
+        });
 
         // Sockets of deleted sources should not be previewed
         for (OutputSocket<?> socket : event.getSource().getOutputSockets()) {
             socket.setPreviewed(false);
         }
     }
 
-    public synchronized void addStep(int index, Step step) {
+    public void addStep(int index, Step step) {
         checkNotNull(step, "The step can not be null");
-        this.steps.add(index, step);
+        checkArgument(!step.removed(), "The step must not have been disabled already");
+
+        writeStepsSafelyConsume(steps -> steps.add(index, step));
+
         this.eventBus.register(step);
         this.eventBus.post(new StepAddedEvent(step, index));
     }
 
-    public synchronized void addStep(Step step) {
+    public void addStep(Step step) {
         addStep(this.steps.size(), step);
     }
 
-    public synchronized void removeStep(Step step) {
+    public void removeStep(Step step) {
         checkNotNull(step, "The step can not be null");
-        this.steps.remove(step);
+
+        writeStepsSafelyConsume(steps -> steps.remove(step));
+
         // Sockets of deleted steps should not be previewed
         for (OutputSocket<?> socket : step.getOutputSockets()) {
             socket.setPreviewed(false);
         }
+        step.setRemoved();
         this.eventBus.unregister(step);
         this.eventBus.post(new StepRemovedEvent(step));
     }
@@ -177,13 +281,19 @@ public synchronized void moveStep(Step step, int delta) {
         checkNotNull(step, "The step can not be null");
         checkArgument(this.steps.contains(step), "The step must exist in the pipeline to be moved");
 
-        final int oldIndex = this.steps.indexOf(step);
-        this.steps.remove(oldIndex);
+        // We are modifying the steps array
+        writeStepsSafelyConsume(steps -> {
+            final int oldIndex = this.steps.indexOf(step);
+            this.steps.remove(oldIndex);
 
-        // Compute the new index of the step, clamping to the beginning or end of pipeline if it goes past either end
-        final int newIndex = Math.min(Math.max(oldIndex + delta, 0), this.steps.size());
-        this.steps.add(newIndex, step);
+            // Compute the new index of the step, clamping to the beginning or end of pipeline if it goes past either end
+            final int newIndex = Math.min(Math.max(oldIndex + delta, 0), this.steps.size());
+            this.steps.add(newIndex, step);
+        });
+
+        // Do not lock while posting the event
         eventBus.post(new StepMovedEvent(step, delta));
+
     }
 
     @Subscribe