Implement generic async task runner

Author: jushar

Client/mods/deathmatch/logic/CClientGame.cpp

@@ -301,6 +301,9 @@ CClientGame::CClientGame ( bool bLocalPlay )
     CLuaFunctionDefs::Initialize ( m_pLuaManager, m_pScriptDebugging, this );
     CLuaDefs::Initialize ( this, m_pLuaManager, m_pScriptDebugging );
 
+    // Start async task scheduler
+    m_pAsyncTaskScheduler = new SharedUtil::CAsyncTaskScheduler(2);
+
     // Disable the enter/exit vehicle key button (we want to handle this button ourselves)
     g_pMultiplayer->DisableEnterExitVehicleKey ( true );
 
@@ -391,6 +394,9 @@ CClientGame::~CClientGame ( void )
     m_pTransferBox->Hide();
     m_pBigPacketTransferBox->Hide();
 
+    // Stop async task scheduler
+    SAFE_DELETE(m_pAsyncTaskScheduler);
+
     SAFE_DELETE( m_pVoiceRecorder );
 
     // Singular file download manager
@@ -1305,6 +1311,9 @@ void CClientGame::DoPulses ( void )
     // Send screen shot data
     ProcessDelayedSendList ();
 
+    // Collect async task scheduler results
+    m_pAsyncTaskScheduler->CollectResults();
+
     TIMING_CHECKPOINT( "-CClientGame::DoPulses" );
 }
 

Client/mods/deathmatch/logic/CClientGame.h

@@ -306,6 +306,8 @@ class CClientGame
     CRemoteCalls*                       GetRemoteCalls                  ( void )        { return m_pRemoteCalls; }
     CResourceFileDownloadManager*       GetResourceFileDownloadManager  ( void )        { return m_pResourceFileDownloadManager; }
 
+    inline SharedUtil::CAsyncTaskScheduler* GetAsyncTaskScheduler       ( void )        { return m_pAsyncTaskScheduler; }
+
     // Status toggles
     void                                ShowNetstat                     ( int iCmd );
     void                                ShowEaeg                        ( bool bShow );
@@ -793,6 +795,8 @@ class CClientGame
     std::set < SString >                m_AllowKeyUpMap;
     uint                                m_uiPrecisionCallDepth;
     SString                             m_strFileCacheRoot;
+
+    SharedUtil::CAsyncTaskScheduler*    m_pAsyncTaskScheduler;
 };
 
 extern CClientGame* g_pClientGame;

Server/mods/deathmatch/logic/CGame.cpp

@@ -274,6 +274,9 @@ CGame::~CGame ( void )
     // Stop networking
     Stop ();
 
+    // Stop async task scheduler
+    SAFE_DELETE(m_pAsyncTaskScheduler);
+
      // Destroy our stuff
     SAFE_DELETE( m_pResourceManager );
 
@@ -483,6 +486,8 @@ void CGame::DoPulse ( void )
 
     CLOCK_CALL1( m_pLatentTransferManager->DoPulse (); );
 
+    CLOCK_CALL1(m_pAsyncTaskScheduler->CollectResults());
+
     PrintLogOutputFromNetModule();
     m_pScriptDebugging->UpdateLogOutput();
 
@@ -597,6 +602,9 @@ bool CGame::Start ( int iArgumentCount, char* szArguments [] )
     unsigned short usServerPort = m_pMainConfig->GetServerPort ();
     unsigned int uiMaxPlayers = m_pMainConfig->GetMaxPlayers ();
 
+    // Start async task scheduler
+    m_pAsyncTaskScheduler = new SharedUtil::CAsyncTaskScheduler(2);
+
     // Create the account manager
     strBuffer = g_pServerInterface->GetModManager ()->GetAbsolutePath ( "internal.db" );
     m_pDatabaseManager = NewDatabaseManager ();

Server/mods/deathmatch/logic/CGame.h

@@ -266,6 +266,7 @@ class CGame
     inline CCustomWeaponManager*    GetCustomWeaponManager      ( void )        { return m_pCustomWeaponManager; }
     inline CFunctionUseLogger*      GetFunctionUseLogger        ( void )        { return m_pFunctionUseLogger; }
     inline CMasterServerAnnouncer*  GetMasterServerAnnouncer    ( void )        { return m_pMasterServerAnnouncer; }
+    inline SharedUtil::CAsyncTaskScheduler* GetAsyncTaskScheduler(void )        { return m_pAsyncTaskScheduler; }
 
     void                        JoinPlayer                  ( CPlayer& Player );
     void                        InitialDataStream           ( CPlayer& Player );
@@ -588,6 +589,8 @@ class CGame
     SString                     m_strPrevMinClientKickRequirement;
     SString                     m_strPrevMinClientConnectRequirement;
     SString                     m_strPrevLowestConnectedPlayerVersion;
+
+    SharedUtil::CAsyncTaskScheduler* m_pAsyncTaskScheduler;
 };
 
 #endif

Shared/sdk/SharedUtil.AsyncTaskScheduler.h

@@ -0,0 +1,98 @@
+#pragma once
+#include <queue>
+#include <functional>
+#include <memory>
+#include <thread>
+#include <mutex>
+
+namespace SharedUtil
+{
+    ///////////////////////////////////////////////////////////////
+    //
+    // CAsyncTaskScheduler class
+    //
+    // Asynchronously executes tasks in secondary worker threads
+    // and returns the result back to the main thread
+    //
+    ///////////////////////////////////////////////////////////////
+    class CAsyncTaskScheduler
+    {
+        struct SBaseTask
+        {
+            virtual void Execute() = 0;
+            virtual void ProcessResult() = 0;
+        };
+
+        template<typename ResultType>
+        struct STask : public SBaseTask
+        {
+            using TaskFunction_t = std::function<ResultType()>;
+            using ReadyFunction_t = std::function<void(const ResultType&)>;
+
+            TaskFunction_t m_TaskFunction;
+            ReadyFunction_t m_ReadyFunction;
+            ResultType m_Result;
+
+            STask(const TaskFunction_t& taskFunc, const ReadyFunction_t& readyFunc) : m_TaskFunction(taskFunc), m_ReadyFunction(readyFunc)
+            {}
+
+            void Execute() override
+            {
+                m_Result = std::move(m_TaskFunction());
+            }
+
+            void ProcessResult() override
+            {
+                m_ReadyFunction(m_Result);
+            }
+        };
+
+    public:
+        //
+        // Creates a new async task scheduler
+        // with a fixed number of worker threads
+        //
+        CAsyncTaskScheduler(std::size_t numWorkers);
+
+        //
+        // Ends all worker threads (waits for the last task to finish)
+        //
+        ~CAsyncTaskScheduler();
+
+        //
+        // Pushes a new task for execution once a worker is free
+        // (Template Parameter) ResultType: The type of the result
+        //
+        // taskFunc: Time-consuming function that is executed on the secondary thread (be aware of thread safety!)
+        // readyFunc: Function that is called once the result is ready (called on the main thread)
+        //
+        template<typename ResultType>
+        void PushTask(const std::function<ResultType()>& taskFunc, const std::function<void(const ResultType&)>& readyFunc)
+        {
+            std::unique_ptr<SBaseTask> pTask{ new STask<ResultType>{ taskFunc, readyFunc } };
+
+            std::lock_guard<std::mutex>{ m_TasksMutex };
+            m_Tasks.push(std::move(pTask));
+        }
+
+        //
+        // Collects (polls) the results of the finished tasks
+        // and invokes its ready-functions on the main thread
+        // THIS FUNCTION MUST BE CALLED ON THE MAIN THREAD
+        //
+        void CollectResults();
+
+    protected:
+        void DoWork();
+
+    private:
+        std::vector<std::thread> m_Workers;
+        bool m_Running = true;
+
+        std::queue<std::unique_ptr<SBaseTask>> m_Tasks;
+        std::mutex m_TasksMutex;
+
+        std::vector<std::unique_ptr<SBaseTask>> m_TaskResults;
+        std::mutex m_TaskResultsMutex;
+    };
+}

Shared/sdk/SharedUtil.AsyncTaskScheduler.hpp

@@ -0,0 +1,67 @@
+#include "SharedUtil.AsyncTaskScheduler.h"
+
+namespace SharedUtil
+{
+    CAsyncTaskScheduler::CAsyncTaskScheduler(std::size_t numWorkers)
+    {
+        for (std::size_t i = 0; i < numWorkers; ++i)
+        {
+            m_Workers.emplace_back(&CAsyncTaskScheduler::DoWork, this);
+        }
+    }
+
+    CAsyncTaskScheduler::~CAsyncTaskScheduler()
+    {
+        m_Running = false;
+
+        // Wait for all threads to end
+        for (auto& thread : m_Workers)
+        {
+            if (thread.joinable())
+                thread.join();
+        }
+    }
+
+    void CAsyncTaskScheduler::CollectResults()
+    {
+        std::lock_guard<std::mutex>{ m_TaskResultsMutex };
+
+        for (auto& pTask : m_TaskResults)
+        {
+            pTask->ProcessResult();
+        }
+        m_TaskResults.clear();
+    }
+
+    void CAsyncTaskScheduler::DoWork()
+    {
+        while (m_Running)
+        {
+            m_TasksMutex.lock();
+
+            // Sleep a bit if there are no tasks
+            if (m_Tasks.empty())
+            {
+                m_TasksMutex.unlock();
+                std::this_thread::sleep_for(std::chrono::milliseconds(4));
+                continue;
+            }
+                
+            // Get task and remove from front
+            std::unique_ptr<SBaseTask> pTask = std::move(m_Tasks.front());
+            m_Tasks.pop();
+
+            m_TasksMutex.unlock();
+
+            // Execute time-consuming task
+            pTask->Execute();
+
+            // Put into result queue
+            {
+                std::lock_guard<std::mutex>{ m_TaskResultsMutex };
+
+                m_TaskResults.push_back(std::move(pTask));
+            }
+        }
+    }
+}

Shared/sdk/SharedUtil.h

@@ -91,6 +91,7 @@ typedef float               FLOAT;      //  32
 #endif
 #include "SharedUtil.Profiling.h"
 #include "SharedUtil.Logging.h"
+#include "SharedUtil.AsyncTaskScheduler.h"
 #include "CFastList.h"
 #include "CDuplicateLineFilter.h"
 

Shared/sdk/SharedUtil.hpp

@@ -26,3 +26,4 @@
 #endif
 #include "SharedUtil.Profiling.hpp"
 #include "SharedUtil.Logging.hpp"
+#include "SharedUtil.AsyncTaskScheduler.hpp"