SG Threading added new class for an exclusive thread

An exclusive thread is one that is suited to being used where the thread needs to be activated at a certain point, and also that the code activating the thread may also need to wait for thread completion.

Example of this is the new background garbage collection for Nasal. The thread will be activated at the end of the frame processing and at the start of the next frame the thread can be awaited - thus allowing the thread to work in parallel with the rendering.

simgear/threads/SGThread.cxx

@@ -25,6 +25,7 @@
 #endif
 
 #include <simgear/compiler.h>
+#include <simgear/debug/logstream.hxx>
 
 #include "SGThread.hxx"
 
@@ -416,3 +417,98 @@ SGWaitCondition::broadcast()
 {
     _privateData->broadcast();
 }
+
+
+SGExclusiveThread::SGExclusiveThread() :
+        _started(false), _terminated(false), last_await_time(0),
+        dataReady(false), complete(true), process_ran(false), process_running(false)
+    {
+    }
+
+    SGExclusiveThread::~SGExclusiveThread()
+    {
+
+    }
+
+    void SGExclusiveThread::release() {
+        std::unique_lock<std::mutex> lck(mutex_);
+        if (!complete) {
+            SG_LOG(SG_NASAL, SG_ALERT, "[SGExclusiveThread]  not finished - skipping");
+            return;
+        }
+        if (!complete.exchange(false))
+            SG_LOG(SG_NASAL, SG_ALERT, "[SGExclusiveThread]  concurrent failure (2)");
+        if (dataReady.exchange(true))
+            SG_LOG(SG_NASAL, SG_ALERT, "[SGExclusiveThread]  concurrent failure (1)");
+        condVar.notify_one();
+    }
+    void SGExclusiveThread::wait() {
+        std::unique_lock<std::mutex> lck(mutex_);
+        if (!dataReady)
+        {
+            do
+            {
+                condVar.wait(lck);
+            } while (!dataReady);
+        }
+    }
+    void SGExclusiveThread::clearAwaitCompletionTime() {
+        last_await_time = 0;
+    }
+    void SGExclusiveThread::awaitCompletion() {
+        timestamp.stamp();
+        std::unique_lock<std::mutex> lck(Cmutex_);
+        if (!complete)
+        {
+            do {
+                CcondVar.wait(lck);
+            } while (!complete.load());
+        }
+
+        if (process_ran) {
+            last_await_time = timestamp.elapsedUSec();
+            process_ran = 0;
+        }
+    }
+
+    void SGExclusiveThread::setCompletion() {
+        std::unique_lock<std::mutex> lck(Cmutex_);
+        if (!dataReady.exchange(false))
+            SG_LOG(SG_NASAL, SG_ALERT, "[SGExclusiveThread]  atomic operation on dataReady failed (5)\n");
+
+        if (complete.exchange(true))
+            SG_LOG(SG_NASAL, SG_ALERT, "[SGExclusiveThread]  atomic operation on complete failed (5)\n");
+        CcondVar.notify_one();
+    }
+    void SGExclusiveThread::run()
+    {
+        process_running = true;
+        while (!_terminated) {
+            wait();
+            process_ran = process();
+            setCompletion();
+        }
+        process_running = false;
+        _terminated = false;
+        _started = false;
+    }
+
+    void SGExclusiveThread::terminate() {
+        _terminated = true;
+    }
+    bool SGExclusiveThread::stop()
+    {
+        return true;
+    }
+    void SGExclusiveThread::ensure_running()
+    {
+        if (!_started)
+        {
+            _started = true;
+            start();
+        }
+    }
+    bool SGExclusiveThread::is_running()
+    {
+        return process_running;
+    }

simgear/threads/SGThread.hxx

@@ -23,7 +23,11 @@
 #ifndef SGTHREAD_HXX_INCLUDED
 #define SGTHREAD_HXX_INCLUDED 1
 
+#include <mutex>
+#include <condition_variable>
+#include <atomic>
 #include <simgear/compiler.h>
+#include <simgear/timing/timestamp.hxx>
 
 /**
  * Encapsulate generic threading methods.
@@ -184,4 +188,41 @@ private:
     PrivateData* _privateData;
 };
 
+///
+/// an exclusive thread is one that is designed for frame processing;
+/// it has the ability to synchronise such that the caller can await
+/// the processing to finish.
+class SGExclusiveThread : public SGThread{
+private:
+    std::mutex mutex_;
+    std::condition_variable condVar;
+    SGTimeStamp timestamp;
+    std::mutex Cmutex_;
+    std::condition_variable CcondVar;
+
+    bool _started;
+    bool _terminated;
+    int last_await_time;
+
+    std::atomic<bool> dataReady;
+    std::atomic<bool> complete;
+    std::atomic<bool> process_ran;
+    std::atomic<bool> process_running;
+
+public:
+    SGExclusiveThread();
+    virtual ~SGExclusiveThread();
+    void release();
+    void wait();
+    void clearAwaitCompletionTime();
+    virtual void awaitCompletion();
+    void setCompletion();
+    virtual int process() = 0;
+    virtual void run();
+    void terminate();
+    bool stop();
+    void ensure_running();
+    bool is_running();
+};
+
 #endif /* SGTHREAD_HXX_INCLUDED */