Wait on execute threads. Closes #798 and closes #659

Author: dcwhite

src/Dataflow/Engine/Scheduler/DynamicExecutor/WorkUnitConsumer.h

@@ -47,8 +47,11 @@ namespace DynamicExecutor {
   class SCISHARE ModuleConsumer : boost::noncopyable
   {
   public:
-    explicit ModuleConsumer(ModuleWorkQueuePtr workQueue, const Networks::ExecutableLookup* lookup, ProducerInterfacePtr producer) :
-    work_(workQueue), producer_(producer), lookup_(lookup), shouldLog_(SCIRun::Core::Logging::Log::get().verbose())
+    explicit ModuleConsumer(ModuleWorkQueuePtr workQueue, const Networks::ExecutableLookup* lookup, ProducerInterfacePtr producer,
+      boost::thread_group& executeThreadGroup) :
+    work_(workQueue), producer_(producer), lookup_(lookup),
+    executeThreadGroup_(executeThreadGroup),
+    shouldLog_(SCIRun::Core::Logging::Log::get().verbose())
     {
       log_.setVerbose(shouldLog_);
       if (shouldLog_)
@@ -86,7 +89,7 @@ namespace DynamicExecutor {
 
             ModuleExecutor executor(unit, lookup_, producer_);
             /// @todo: thread pool
-            threads_.create_thread(boost::bind(&ModuleExecutor::run, executor));
+            executeThreadGroup_.create_thread(boost::bind(&ModuleExecutor::run, executor));
           }
           else
           {
@@ -107,7 +110,7 @@ namespace DynamicExecutor {
     ModuleWorkQueuePtr work_;
     ProducerInterfacePtr producer_;
     const Networks::ExecutableLookup* lookup_;
-    mutable boost::thread_group threads_;
+    boost::thread_group& executeThreadGroup_;
 
     static Core::Logging::Log& log_;
     bool shouldLog_;

src/Dataflow/Engine/Scheduler/DynamicMultithreadedNetworkExecutor.cc

@@ -52,12 +52,13 @@ namespace SCIRun {
       {
       public:
         DynamicMultithreadedNetworkExecutorImpl(const ExecutionContext& context, const NetworkInterface* network, Mutex* lock, size_t numModules) :
+          executeThreads_(new boost::thread_group),
           lookup_(&context.lookup),
           bounds_(&context.bounds()),
           work_(new DynamicExecutor::ModuleWorkQueue(numModules)),
           producer_(new DynamicExecutor::ModuleProducer(context.addAdditionalFilter(ModuleWaitingFilter::Instance()),
             network, lock, work_, numModules)),
-            consumer_(new DynamicExecutor::ModuleConsumer(work_, lookup_, producer_)),
+            consumer_(new DynamicExecutor::ModuleConsumer(work_, lookup_, producer_, *executeThreads_)),
           network_(network)
         {
         }
@@ -71,8 +72,10 @@ namespace SCIRun {
           boost::thread produce(boost::ref(*producer_));
           consume.join();
           produce.join();
+          executeThreads_->join_all();
         }
       private:
+        mutable boost::shared_ptr<boost::thread_group> executeThreads_;
         const Networks::ExecutableLookup* lookup_;
         const ExecutionBounds* bounds_;
         DynamicExecutor::ModuleWorkQueuePtr work_;