From c71f2874983b04d42136d9a2e779819dc0945d7b Mon Sep 17 00:00:00 2001
From: Richard Harrison <rjh@zaretto.com>
Date: Tue, 11 Jun 2019 12:42:16 +0200
Subject: [PATCH] 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 | 96 ++++++++++++++++++++++++++++++++++++
 simgear/threads/SGThread.hxx | 41 +++++++++++++++
 2 files changed, 137 insertions(+)

diff --git a/simgear/threads/SGThread.cxx b/simgear/threads/SGThread.cxx
index 0ba71f40..c5686e51 100644
--- a/simgear/threads/SGThread.cxx
+++ b/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;
+    }
diff --git a/simgear/threads/SGThread.hxx b/simgear/threads/SGThread.hxx
index d905c624..e5610f4b 100644
--- a/simgear/threads/SGThread.hxx
+++ b/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 */