Move database queue onto Boost.Asio

Signed-off-by: AlterEgo <admin@projectskyfire.org>

CMakeLists.txt

@@ -56,6 +56,10 @@ include(CheckIncludeFiles)
 # set default buildoptions and print them
 include(cmake/options.cmake)
 
+if(SKYFIRE_BUILD_TESTS)
+  enable_testing()
+endif()
+
 # turn off PCH totally if enabled (hidden setting, mainly for devs)
 if( NOPCH )
   set(USE_COREPCH 0)

cmake/options.cmake

@@ -14,4 +14,4 @@ option(WITHOUT_GIT        "Disable the GIT testing routines"
 option(AUTH_SERVER        "Build authserver"                                            1)
 option(WITH_CXX_23_STD    "Use c++23 standard"                                          1)
 option(WITH_CXX_DRAFT_STD "Use c++ draft standard"                                      0)
-
+option(SKYFIRE_BUILD_TESTS "Build SkyFire developer test binaries"                      0)

src/CMakeLists.txt

@@ -10,3 +10,7 @@ if(TOOLS)
   add_subdirectory(tools)
 endif(TOOLS)
 
+if(SKYFIRE_BUILD_TESTS)
+  add_subdirectory(tests)
+endif()
+

src/server/shared/Database/DatabaseQueue.cpp

@@ -0,0 +1,85 @@
+/*
+* This file is part of Project SkyFire https://www.projectskyfire.org.
+* See LICENSE.md file for Copyright information
+*/
+
+#include "DatabaseQueue.h"
+#include "SQLOperation.h"
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/post.hpp>
+
+#include <mutex>
+
+namespace Skyfire
+{
+    namespace
+    {
+        thread_local MySQLConnection* CurrentDatabaseConnection = nullptr;
+    }
+
+    struct DatabaseQueue::Impl
+    {
+        typedef boost::asio::executor_work_guard<boost::asio::io_context::executor_type> WorkGuard;
+
+        Impl()
+            : ioContext(), workGuard(new WorkGuard(boost::asio::make_work_guard(ioContext))), closed(false)
+        {
+        }
+
+        boost::asio::io_context ioContext;
+        std::unique_ptr<WorkGuard> workGuard;
+        std::mutex stateLock;
+        bool closed;
+    };
+
+    DatabaseQueue::DatabaseQueue()
+        : _impl(new Impl)
+    {
+    }
+
+    DatabaseQueue::~DatabaseQueue()
+    {
+        close();
+    }
+
+    void DatabaseQueue::enqueue(SQLOperation* operation)
+    {
+        if (!operation)
+            return;
+
+        std::lock_guard<std::mutex> guard(_impl->stateLock);
+        if (_impl->closed)
+            return;
+
+        boost::asio::post(_impl->ioContext,
+            [operation]
+            {
+                operation->SetConnection(CurrentDatabaseConnection);
+                operation->call();
+                delete operation;
+            });
+    }
+
+    int DatabaseQueue::run(MySQLConnection* connection)
+    {
+        if (!connection)
+            return -1;
+
+        CurrentDatabaseConnection = connection;
+        _impl->ioContext.run();
+        CurrentDatabaseConnection = nullptr;
+        return 0;
+    }
+
+    void DatabaseQueue::close()
+    {
+        std::lock_guard<std::mutex> guard(_impl->stateLock);
+        if (_impl->closed)
+            return;
+
+        _impl->closed = true;
+        _impl->workGuard.reset();
+    }
+}

src/server/shared/Database/DatabaseQueue.h

@@ -6,10 +6,9 @@
 #ifndef _DATABASEQUEUE_H
 #define _DATABASEQUEUE_H
 
-#include <condition_variable>
-#include <mutex>
-#include <queue>
+#include <memory>
 
+class MySQLConnection;
 class SQLOperation;
 
 namespace Skyfire
@@ -17,47 +16,20 @@ namespace Skyfire
     class DatabaseQueue
     {
     public:
-        void enqueue(SQLOperation* operation)
-        {
-            {
-                std::lock_guard<std::mutex> lock(_mutex);
-                if (_closed)
-                    return;
+        DatabaseQueue();
+        ~DatabaseQueue();
 
-                _queue.push(operation);
-            }
-
-            _condition.notify_one();
-        }
-
-        SQLOperation* dequeue()
-        {
-            std::unique_lock<std::mutex> lock(_mutex);
-            _condition.wait(lock, [this] { return _closed || !_queue.empty(); });
-
-            if (_queue.empty())
-                return nullptr;
-
-            SQLOperation* operation = _queue.front();
-            _queue.pop();
-            return operation;
-        }
-
-        void close()
-        {
-            {
-                std::lock_guard<std::mutex> lock(_mutex);
-                _closed = true;
-            }
-
-            _condition.notify_all();
-        }
+        void enqueue(SQLOperation* operation);
+        int run(MySQLConnection* connection);
+        void close();
 
     private:
-        std::mutex _mutex;
-        std::condition_variable _condition;
-        std::queue<SQLOperation*> _queue;
-        bool _closed = false;
+        struct Impl;
+
+        std::unique_ptr<Impl> _impl;
+
+        DatabaseQueue(DatabaseQueue const& right) = delete;
+        DatabaseQueue& operator=(DatabaseQueue const& right) = delete;
     };
 }
 

src/server/shared/Database/DatabaseWorker.cpp

@@ -26,20 +26,7 @@ int DatabaseWorker::svc()
     if (!m_queue)
         return -1;
 
-    SQLOperation* request = NULL;
-    while (1)
-    {
-        request = m_queue->dequeue();
-        if (!request)
-            break;
-
-        request->SetConnection(m_conn);
-        request->call();
-
-        delete request;
-    }
-
-    return 0;
+    return m_queue->run(m_conn);
 }
 
 int DatabaseWorker::wait()

src/tests/CMakeLists.txt

@@ -0,0 +1,37 @@
+#
+# This file is part of Project SkyFire https://www.projectskyfire.org.
+# See COPYRIGHT file for Copyright information
+#
+
+add_executable(database_queue_tests
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Database/DatabaseQueue.cpp
+  DatabaseQueueTests.cpp
+)
+
+target_include_directories(database_queue_tests PRIVATE
+  ${CMAKE_BINARY_DIR}
+  ${CMAKE_SOURCE_DIR}/src/server/shared
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Configuration
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Database
+  ${CMAKE_SOURCE_DIR}/src/server/shared/DataStores
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Debugging
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Dynamic
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Logging
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Network
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Packets
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Platform
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Threading
+  ${CMAKE_SOURCE_DIR}/src/server/shared/Utilities
+  ${Boost_INCLUDE_DIRS}
+  ${MYSQL_INCLUDE_DIR}
+)
+
+if(TARGET Boost::headers)
+  target_link_libraries(database_queue_tests Boost::headers)
+elseif(TARGET Boost::boost)
+  target_link_libraries(database_queue_tests Boost::boost)
+endif()
+
+target_link_libraries(database_queue_tests Threads::Threads)
+
+add_test(NAME database_queue_tests COMMAND database_queue_tests)

src/tests/DatabaseQueueTests.cpp

@@ -0,0 +1,74 @@
+/*
+* This file is part of Project SkyFire https://www.projectskyfire.org.
+* See LICENSE.md file for Copyright information
+*/
+
+#include "DatabaseQueue.h"
+#include "SQLOperation.h"
+
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace
+{
+    class RecordingOperation : public SQLOperation
+    {
+    public:
+        RecordingOperation(std::vector<int>& completed, std::mutex& completedLock, int value, MySQLConnection* expectedConnection)
+            : _completed(completed), _completedLock(completedLock), _value(value), _expectedConnection(expectedConnection)
+        {
+        }
+
+        bool Execute() override
+        {
+            if (m_conn != _expectedConnection)
+                return false;
+
+            std::lock_guard<std::mutex> guard(_completedLock);
+            _completed.push_back(_value);
+            return true;
+        }
+
+    private:
+        std::vector<int>& _completed;
+        std::mutex& _completedLock;
+        int _value;
+        MySQLConnection* _expectedConnection;
+    };
+}
+
+int main()
+{
+    Skyfire::DatabaseQueue queue;
+
+    std::vector<int> completed;
+    std::mutex completedLock;
+    MySQLConnection* connection = reinterpret_cast<MySQLConnection*>(0x1);
+    int runResult = -1;
+
+    std::thread worker([&queue, connection, &runResult]
+    {
+        runResult = queue.run(connection);
+    });
+
+    queue.enqueue(new RecordingOperation(completed, completedLock, 1, connection));
+    queue.enqueue(new RecordingOperation(completed, completedLock, 2, connection));
+    queue.close();
+    worker.join();
+
+    if (runResult != 0)
+    {
+        std::cerr << "DatabaseQueue::run returned " << runResult << '\n';
+        return 1;
+    }
+
+    if (completed != std::vector<int>{ 1, 2 })
+    {
+        std::cerr << "DatabaseQueue did not drain queued operations before closing\n";
+        return 1;
+    }
+
+    return 0;
+}