Push execution launching to separate thread

Author: dcwhite

src/Dataflow/Engine/Controller/NetworkEditorController.cc

@@ -64,6 +64,7 @@ NetworkEditorController::NetworkEditorController(ModuleFactoryHandle mf, ModuleS
   reexFactory_(reex),
   executorFactory_(executorFactory),
   serializationManager_(nesm),
+  executionManager_(currentExecutor_), //TODO: pass in factory instead
   signalSwitch_(true)
 {
   dynamicPortManager_.reset(new DynamicPortManager(connectionAdded_, connectionRemoved_, this));
@@ -76,7 +77,7 @@ NetworkEditorController::NetworkEditorController(ModuleFactoryHandle mf, ModuleS
 }
 
 NetworkEditorController::NetworkEditorController(SCIRun::Dataflow::Networks::NetworkHandle network, ExecutionStrategyFactoryHandle executorFactory, NetworkEditorSerializationManager* nesm)
-  : theNetwork_(network), executorFactory_(executorFactory), serializationManager_(nesm)
+  : theNetwork_(network), executorFactory_(executorFactory), serializationManager_(nesm), executionManager_(currentExecutor_) //TODO: pass in factory instead
 {
 }
 
@@ -367,7 +368,31 @@ void NetworkEditorController::executeGeneric(const ExecutableLookup* lookup, Mod
 {
   initExecutor();
   auto context = createExecutionContext(lookup, filter);
-  currentExecutor_->execute(*context);
+
+  executionManager_.enqueueContext(context);
+}
+
+ExecutionQueueManager::ExecutionQueueManager(ExecutionStrategyHandle& currentExecutor) : contexts_(1), currentExecutor_(currentExecutor),
+  executionLaunchThread_([this]() {executeTopContext(); } )
+{
+}
+
+void ExecutionQueueManager::enqueueContext(ExecutionContextHandle context)
+{
+  if (contexts_.push(context))
+    std::cout << "ctx queued" << std::endl;
+  else
+    std::cout << "ctx queue is full" << std::endl;
+}
+
+void ExecutionQueueManager::executeTopContext()
+{
+  while (true)
+  {
+    if (contexts_.read_available())
+      contexts_.consume_one([&](ExecutionContextHandle ctx) { currentExecutor_->execute(*ctx); });
+    boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
+  }
 }
 
 NetworkHandle NetworkEditorController::getNetwork() const

src/Dataflow/Engine/Controller/NetworkEditorController.h

@@ -83,13 +83,13 @@ namespace Engine {
   class SCISHARE NetworkEditorController : public NetworkIOInterface<Networks::NetworkFileHandle>, public Networks::NetworkEditorControllerInterface
   {
   public:
-    explicit NetworkEditorController(Networks::ModuleFactoryHandle mf,
+    NetworkEditorController(Networks::ModuleFactoryHandle mf,
       Networks::ModuleStateFactoryHandle sf,
       ExecutionStrategyFactoryHandle executorFactory,
       Core::Algorithms::AlgorithmFactoryHandle algoFactory,
       Networks::ReexecuteStrategyFactoryHandle reexFactory,
       Networks::NetworkEditorSerializationManager* nesm = 0);
-    explicit NetworkEditorController(Networks::NetworkHandle network, ExecutionStrategyFactoryHandle executorFactory, Networks::NetworkEditorSerializationManager* nesm = 0);
+    NetworkEditorController(Networks::NetworkHandle network, ExecutionStrategyFactoryHandle executorFactory, Networks::NetworkEditorSerializationManager* nesm = 0);
 
 //////////////////////////////////////////////////////////////////////////
 //////////////////////Start: To be Pythonized/////////////////////////////
@@ -159,6 +159,8 @@ namespace Engine {
     ExecutionStrategyFactoryHandle executorFactory_;
     Networks::NetworkEditorSerializationManager* serializationManager_;
 
+    ExecutionQueueManager executionManager_;
+
     ModuleAddedSignalType moduleAdded_;
     ModuleRemovedSignalType moduleRemoved_; //not used yet
     ConnectionAddedSignalType connectionAdded_;

src/Dataflow/Engine/Scheduler/ExecutionStrategy.h

@@ -30,6 +30,8 @@
 #define ENGINE_SCHEDULER_EXECUTION_STRATEGY_H
 
 #include <Dataflow/Engine/Scheduler/SchedulerInterfaces.h>
+#include <Dataflow/Engine/Scheduler/DynamicExecutor/WorkQueue.h>
+#include <boost/thread.hpp>
 #include <Dataflow/Engine/Scheduler/share.h>
 
 namespace SCIRun {
@@ -64,6 +66,22 @@ namespace Engine {
 
   typedef boost::shared_ptr<ExecutionStrategyFactory> ExecutionStrategyFactoryHandle;
 
+
+  class SCISHARE ExecutionQueueManager
+  {
+  public:
+    explicit ExecutionQueueManager(ExecutionStrategyHandle& currentExecutor);
+    void enqueueContext(ExecutionContextHandle context);
+  private:
+    typedef DynamicExecutor::WorkQueue<ExecutionContextHandle>::Impl ExecutionContextQueue;
+    ExecutionContextQueue contexts_;
+
+    ExecutionStrategyHandle& currentExecutor_;
+
+    boost::thread executionLaunchThread_;
+
+    void executeTopContext();
+  };
 }
 }}