Condition check before ending the main thread.

Make sure the main thread waits for the child thread with the Run() method.

* Add this at the end of your Run() method for each device:
```
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
```
* Then you must replace the `char ch; cin.get(ch);` in your main() function with:
```
boost::unique_lock<boost::mutex> lock(processor.fRunningMutex);
while (!processor.fRunningFinished)
{
    processor.fRunningCondition.wait(lock);
}
```

fairmq/FairMQStateMachine.cxx

@@ -16,6 +16,7 @@
 #include "FairMQLogger.h"
 
 FairMQStateMachine::FairMQStateMachine()
+    : fRunningFinished(false)
 {
     start();
 }

fairmq/FairMQStateMachine.h

@@ -16,13 +16,16 @@
 #define FAIRMQSTATEMACHINE_H_
 
 #include <boost/thread.hpp>
+#include <boost/thread/mutex.hpp>
+#include <boost/thread/condition_variable.hpp>
 #include <boost/bind.hpp>
+#include <boost/function.hpp>
+
 #include <boost/msm/back/state_machine.hpp>
 #include <boost/msm/front/state_machine_def.hpp>
 #include <boost/msm/front/functor_row.hpp>
 #include <boost/msm/front/euml/common.hpp>
 #include <boost/msm/front/euml/operator.hpp>
-#include <boost/function.hpp>
 
 #include "FairMQLogger.h"
 
@@ -192,6 +195,11 @@ class FairMQStateMachine : public FairMQFSM::FairMQFSM
     FairMQStateMachine();
     virtual ~FairMQStateMachine();
     void ChangeState(int event);
+
+    // condition variable to notify parent thread about end of running state.
+    boost::condition_variable fRunningCondition;
+    boost::mutex fRunningMutex;
+    bool fRunningFinished;
 };
 
 #endif /* FAIRMQSTATEMACHINE_H_ */

fairmq/devices/FairMQBenchmarkSampler.cxx

@@ -76,6 +76,11 @@ void FairMQBenchmarkSampler::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }
 
 void FairMQBenchmarkSampler::ResetEventCounter()

fairmq/devices/FairMQBuffer.cxx

@@ -54,6 +54,11 @@ void FairMQBuffer::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }
 
 FairMQBuffer::~FairMQBuffer()

fairmq/devices/FairMQMerger.cxx

@@ -69,4 +69,9 @@ void FairMQMerger::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }

fairmq/devices/FairMQProxy.cxx

@@ -56,4 +56,9 @@ void FairMQProxy::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }

fairmq/devices/FairMQSink.cxx

@@ -47,6 +47,11 @@ void FairMQSink::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }
 
 FairMQSink::~FairMQSink()

fairmq/devices/FairMQSplitter.cxx

@@ -63,4 +63,9 @@ void FairMQSplitter::Run()
     }
 
     FairMQDevice::Shutdown();
+
+    // notify parent thread about end of processing.
+    boost::lock_guard<boost::mutex> lock(fRunningMutex);
+    fRunningFinished = true;
+    fRunningCondition.notify_one();
 }

fairmq/run/runBenchmarkSampler.cxx

@@ -113,8 +113,12 @@ int main(int argc, char** argv)
     sampler.ChangeState(FairMQBenchmarkSampler::SETINPUT);
     sampler.ChangeState(FairMQBenchmarkSampler::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sampler.fRunningMutex);
+    while (!sampler.fRunningFinished)
+    {
+        sampler.fRunningCondition.wait(lock);
+    }
 
     sampler.ChangeState(FairMQBenchmarkSampler::STOP);
     sampler.ChangeState(FairMQBenchmarkSampler::END);

fairmq/run/runBinSampler.cxx

@@ -113,8 +113,12 @@ int main(int argc, char** argv)
     sampler.ChangeState(FairMQBinSampler::SETINPUT);
     sampler.ChangeState(FairMQBinSampler::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sampler.fRunningMutex);
+    while (!sampler.fRunningFinished)
+    {
+        sampler.fRunningCondition.wait(lock);
+    }
 
     sampler.ChangeState(FairMQBinSampler::STOP);
     sampler.ChangeState(FairMQBinSampler::END);

fairmq/run/runBinSink.cxx

@@ -103,8 +103,12 @@ int main(int argc, char** argv)
     sink.ChangeState(FairMQBinSink::SETINPUT);
     sink.ChangeState(FairMQBinSink::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sink.fRunningMutex);
+    while (!sink.fRunningFinished)
+    {
+        sink.fRunningCondition.wait(lock);
+    }
 
     sink.ChangeState(FairMQBinSink::STOP);
     sink.ChangeState(FairMQBinSink::END);

fairmq/run/runBuffer.cxx

@@ -114,8 +114,12 @@ int main(int argc, char** argv)
     buffer.ChangeState(FairMQBuffer::SETINPUT);
     buffer.ChangeState(FairMQBuffer::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(buffer.fRunningMutex);
+    while (!buffer.fRunningFinished)
+    {
+        buffer.fRunningCondition.wait(lock);
+    }
 
     buffer.ChangeState(FairMQBuffer::STOP);
     buffer.ChangeState(FairMQBuffer::END);

fairmq/run/runMerger.cxx

@@ -124,8 +124,12 @@ int main(int argc, char** argv)
     merger.ChangeState(FairMQMerger::SETINPUT);
     merger.ChangeState(FairMQMerger::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(merger.fRunningMutex);
+    while (!merger.fRunningFinished)
+    {
+        merger.fRunningCondition.wait(lock);
+    }
 
     merger.ChangeState(FairMQMerger::STOP);
     merger.ChangeState(FairMQMerger::END);

fairmq/run/runProtoSampler.cxx

@@ -113,8 +113,12 @@ int main(int argc, char** argv)
     sampler.ChangeState(FairMQProtoSampler::SETINPUT);
     sampler.ChangeState(FairMQProtoSampler::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sampler.fRunningMutex);
+    while (!sampler.fRunningFinished)
+    {
+        sampler.fRunningCondition.wait(lock);
+    }
 
     sampler.ChangeState(FairMQProtoSampler::STOP);
     sampler.ChangeState(FairMQProtoSampler::END);

fairmq/run/runProtoSink.cxx

@@ -103,8 +103,12 @@ int main(int argc, char** argv)
     sink.ChangeState(FairMQProtoSink::SETINPUT);
     sink.ChangeState(FairMQProtoSink::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sink.fRunningMutex);
+    while (!sink.fRunningFinished)
+    {
+        sink.fRunningCondition.wait(lock);
+    }
 
     sink.ChangeState(FairMQProtoSink::STOP);
     sink.ChangeState(FairMQProtoSink::END);

fairmq/run/runProxy.cxx

@@ -115,8 +115,12 @@ int main(int argc, char** argv)
     proxy.ChangeState(FairMQProxy::SETINPUT);
     proxy.ChangeState(FairMQProxy::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(proxy.fRunningMutex);
+    while (!proxy.fRunningFinished)
+    {
+        proxy.fRunningCondition.wait(lock);
+    }
 
     proxy.ChangeState(FairMQProxy::STOP);
     proxy.ChangeState(FairMQProxy::END);

fairmq/run/runSink.cxx

@@ -103,8 +103,12 @@ int main(int argc, char** argv)
     sink.ChangeState(FairMQSink::SETINPUT);
     sink.ChangeState(FairMQSink::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(sink.fRunningMutex);
+    while (!sink.fRunningFinished)
+    {
+        sink.fRunningCondition.wait(lock);
+    }
 
     sink.ChangeState(FairMQSink::STOP);
     sink.ChangeState(FairMQSink::END);

fairmq/run/runSplitter.cxx

@@ -124,8 +124,12 @@ int main(int argc, char** argv)
     splitter.ChangeState(FairMQSplitter::SETINPUT);
     splitter.ChangeState(FairMQSplitter::RUN);
 
-    char ch;
+    // wait until the running thread has finished processing.
-    cin.get(ch);
+    boost::unique_lock<boost::mutex> lock(splitter.fRunningMutex);
+    while (!splitter.fRunningFinished)
+    {
+        splitter.fRunningCondition.wait(lock);
+    }
 
     splitter.ChangeState(FairMQSplitter::STOP);
     splitter.ChangeState(FairMQSplitter::END);