Replaced ACE_Task_Base based DatabaseWorker with PCQ

Note: Not exactly sure how shutdown should be handled, currently the queue clears itself out before shutting down
This might need to be changed if the queue should be fully processed before being deleted

src/server/shared/Database/DatabaseWorker.cpp

@@ -20,32 +20,42 @@
 #include "SQLOperation.h"
 #include "MySQLConnection.h"
 #include "MySQLThreading.h"
+#include "ProducerConsumerQueue.h"
 
-DatabaseWorker::DatabaseWorker(ACE_Activation_Queue* new_queue, MySQLConnection* con) :
-m_queue(new_queue),
-m_conn(con)
+DatabaseWorker::DatabaseWorker(ProducerConsumerQueue<SQLOperation*>* newQueue, MySQLConnection* connection)
 {
-    /// Assign thread to task
-    activate();
+    _connection = connection;
+    _queue = newQueue;
+    _cancelationToken = false;
+    _workerThread = std::thread(&DatabaseWorker::WorkerThread, this);
 }
 
-int DatabaseWorker::svc()
+DatabaseWorker::~DatabaseWorker()
 {
-    if (!m_queue)
-        return -1;
+    _cancelationToken = true;
+
+    _queue->Cancel();
+
+    _workerThread.join();
+}
+
+void DatabaseWorker::WorkerThread()
+{
+    if (!_queue)
+        return;
 
-    SQLOperation *request = NULL;
     while (1)
     {
-        request = (SQLOperation*)(m_queue->dequeue());
-        if (!request)
-            break;
+        SQLOperation* operation = nullptr;
 
-        request->SetConnection(m_conn);
-        request->call();
+        _queue->WaitAndPop(operation);
 
-        delete request;
+        if (_cancelationToken)
+            return;
+
+        operation->SetConnection(_connection);
+        operation->call();
+
+        delete operation;
     }
-
-    return 0;
 }

src/server/shared/Database/DatabaseWorker.h

@@ -19,23 +19,24 @@
 #define _WORKERTHREAD_H
 
 #include "Define.h"
-#include <ace/Task.h>
-#include <ace/Activation_Queue.h>
+#include "ProducerConsumerQueue.h"
 
 class MySQLConnection;
 
-class DatabaseWorker : protected ACE_Task_Base
+class DatabaseWorker
 {
     public:
-        DatabaseWorker(ACE_Activation_Queue* new_queue, MySQLConnection* con);
-
-        ///- Inherited from ACE_Task_Base
-        int svc();
-        int wait() { return ACE_Task_Base::wait(); }
+        DatabaseWorker(ProducerConsumerQueue<SQLOperation*>* newQueue, MySQLConnection* connection);
+        ~DatabaseWorker();
 
     private:
-        ACE_Activation_Queue* m_queue;
-        MySQLConnection* m_conn;
+        ProducerConsumerQueue<SQLOperation*>* _queue;
+        MySQLConnection* _connection;
+
+        void WorkerThread();
+        std::thread _workerThread;
+
+        std::atomic_bool _cancelationToken;
 
         DatabaseWorker(DatabaseWorker const& right) = delete;
         DatabaseWorker& operator=(DatabaseWorker const& right) = delete;

src/server/shared/Database/DatabaseWorkerPool.h

@@ -52,7 +52,7 @@ class DatabaseWorkerPool
         DatabaseWorkerPool() : _connectionInfo(NULL)
         {
             _messageQueue = new ACE_Message_Queue<ACE_SYNCH>(8 * 1024 * 1024, 8 * 1024 * 1024);
-            _queue = new ACE_Activation_Queue(_messageQueue);
+            _queue = new ProducerConsumerQueue<SQLOperation*>();
             memset(_connectionCount, 0, sizeof(_connectionCount));
             _connections.resize(IDX_SIZE);
 
@@ -107,16 +107,10 @@ class DatabaseWorkerPool
         {
             TC_LOG_INFO("sql.driver", "Closing down DatabasePool '%s'.", GetDatabaseName());
 
-            //! Shuts down delaythreads for this connection pool by underlying deactivate().
-            //! The next dequeue attempt in the worker thread tasks will result in an error,
-            //! ultimately ending the worker thread task.
-            _queue->queue()->close();
-
             for (uint8 i = 0; i < _connectionCount[IDX_ASYNC]; ++i)
             {
                 T* t = _connections[IDX_ASYNC][i];
                 DatabaseWorker* worker = t->m_worker;
-                worker->wait();     //! Block until no more threads are running this task.
                 delete worker;
                 t->Close();         //! Closes the actualy MySQL connection.
             }
@@ -488,7 +482,7 @@ class DatabaseWorkerPool
 
         void Enqueue(SQLOperation* op)
         {
-            _queue->enqueue(op);
+            _queue->Push(op);
         }
 
         //! Gets a free connection in the synchronous connection pool.
@@ -523,11 +517,11 @@ class DatabaseWorkerPool
             IDX_SIZE
         };
 
-        ACE_Message_Queue<ACE_SYNCH>*   _messageQueue;      //! Message Queue used by ACE_Activation_Queue
-        ACE_Activation_Queue*           _queue;             //! Queue shared by async worker threads.
-        std::vector< std::vector<T*> >  _connections;
-        uint32                          _connectionCount[2];       //! Counter of MySQL connections;
-        MySQLConnectionInfo*            _connectionInfo;
+        ACE_Message_Queue<ACE_SYNCH>*         _messageQueue;      //! Message Queue used by ACE_Activation_Queue
+        ProducerConsumerQueue<SQLOperation*>* _queue;             //! Queue shared by async worker threads.
+        std::vector< std::vector<T*> >        _connections;
+        uint32                                _connectionCount[2];       //! Counter of MySQL connections;
+        MySQLConnectionInfo*                  _connectionInfo;
 };
 
 #endif

src/server/shared/Database/Implementation/CharacterDatabase.h

@@ -26,7 +26,7 @@ class CharacterDatabaseConnection : public MySQLConnection
     public:
         //- Constructors for sync and async connections
         CharacterDatabaseConnection(MySQLConnectionInfo& connInfo) : MySQLConnection(connInfo) { }
-        CharacterDatabaseConnection(ACE_Activation_Queue* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
+        CharacterDatabaseConnection(ProducerConsumerQueue<SQLOperation*>* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
 
         //- Loads database type specific prepared statements
         void DoPrepareStatements();

src/server/shared/Database/Implementation/LoginDatabase.h

@@ -26,7 +26,7 @@ class LoginDatabaseConnection : public MySQLConnection
     public:
         //- Constructors for sync and async connections
         LoginDatabaseConnection(MySQLConnectionInfo& connInfo) : MySQLConnection(connInfo) { }
-        LoginDatabaseConnection(ACE_Activation_Queue* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
+        LoginDatabaseConnection(ProducerConsumerQueue<SQLOperation*>* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
 
         //- Loads database type specific prepared statements
         void DoPrepareStatements();

src/server/shared/Database/Implementation/WorldDatabase.h

@@ -26,7 +26,7 @@ class WorldDatabaseConnection : public MySQLConnection
     public:
         //- Constructors for sync and async connections
         WorldDatabaseConnection(MySQLConnectionInfo& connInfo) : MySQLConnection(connInfo) { }
-        WorldDatabaseConnection(ACE_Activation_Queue* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
+        WorldDatabaseConnection(ProducerConsumerQueue<SQLOperation*>* q, MySQLConnectionInfo& connInfo) : MySQLConnection(q, connInfo) { }
 
         //- Loads database type specific prepared statements
         void DoPrepareStatements();

src/server/shared/Database/MySQLConnection.cpp

@@ -33,8 +33,7 @@
 #include "DatabaseWorker.h"
 #include "Timer.h"
 #include "Log.h"
-
-#include <ace/OS_NS_unistd.h>
+#include "ProducerConsumerQueue.h"
 
 MySQLConnection::MySQLConnection(MySQLConnectionInfo& connInfo) :
 m_reconnecting(false),
@@ -45,7 +44,7 @@ m_Mysql(NULL),
 m_connectionInfo(connInfo),
 m_connectionFlags(CONNECTION_SYNCH) { }
 
-MySQLConnection::MySQLConnection(ACE_Activation_Queue* queue, MySQLConnectionInfo& connInfo) :
+MySQLConnection::MySQLConnection(ProducerConsumerQueue<SQLOperation*>* queue, MySQLConnectionInfo& connInfo) :
 m_reconnecting(false),
 m_prepareError(false),
 m_queue(queue),
@@ -502,8 +501,8 @@ bool MySQLConnection::_HandleMySQLErrno(uint32 errNo)
             }
 
             uint32 lErrno = mysql_errno(GetHandle());   // It's possible this attempted reconnect throws 2006 at us. To prevent crazy recursive calls, sleep here.
-            ACE_OS::sleep(3);                           // Sleep 3 seconds
-            return _HandleMySQLErrno(lErrno);           // Call self (recursive)
+            std::this_thread::sleep_for(std::chrono::seconds(3)); // Sleep 3 seconds
+            return _HandleMySQLErrno(lErrno);                     // Call self (recursive)
         }
 
         case ER_LOCK_DEADLOCK:
@@ -517,12 +516,12 @@ bool MySQLConnection::_HandleMySQLErrno(uint32 errNo)
         case ER_BAD_FIELD_ERROR:
         case ER_NO_SUCH_TABLE:
             TC_LOG_ERROR("sql.sql", "Your database structure is not up to date. Please make sure you've executed all queries in the sql/updates folders.");
-            ACE_OS::sleep(10);
+            std::this_thread::sleep_for(std::chrono::seconds(10));
             std::abort();
             return false;
         case ER_PARSE_ERROR:
             TC_LOG_ERROR("sql.sql", "Error while parsing SQL. Core fix required.");
-            ACE_OS::sleep(10);
+            std::this_thread::sleep_for(std::chrono::seconds(10));
             std::abort();
             return false;
         default:

src/server/shared/Database/MySQLConnection.h

@@ -20,6 +20,7 @@
 #include "DatabaseWorkerPool.h"
 #include "Transaction.h"
 #include "Util.h"
+#include "ProducerConsumerQueue.h"
 
 #ifndef _MYSQLCONNECTION_H
 #define _MYSQLCONNECTION_H
@@ -70,7 +71,7 @@ class MySQLConnection
 
     public:
         MySQLConnection(MySQLConnectionInfo& connInfo);                               //! Constructor for synchronous connections.
-        MySQLConnection(ACE_Activation_Queue* queue, MySQLConnectionInfo& connInfo);  //! Constructor for asynchronous connections.
+        MySQLConnection(ProducerConsumerQueue<SQLOperation*>* queue, MySQLConnectionInfo& connInfo);  //! Constructor for asynchronous connections.
         virtual ~MySQLConnection();
 
         virtual bool Open();
@@ -125,7 +126,7 @@ class MySQLConnection
         bool _HandleMySQLErrno(uint32 errNo);
 
     private:
-        ACE_Activation_Queue* m_queue;                      //! Queue shared with other asynchronous connections.
+        ProducerConsumerQueue<SQLOperation*>* m_queue;      //! Queue shared with other asynchronous connections.
         DatabaseWorker*       m_worker;                     //! Core worker task.
         MYSQL *               m_Mysql;                      //! MySQL Handle.
         MySQLConnectionInfo&  m_connectionInfo;             //! Connection info (used for logging)

src/server/shared/Database/SQLOperation.h

@@ -18,9 +18,6 @@
 #ifndef _SQLOPERATION_H
 #define _SQLOPERATION_H
 
-#include <ace/Method_Request.h>
-#include <ace/Activation_Queue.h>
-
 #include "QueryResult.h"
 
 //- Forward declare (don't include header to prevent circular includes)
@@ -56,7 +53,7 @@ union SQLResultSetUnion
 
 class MySQLConnection;
 
-class SQLOperation : public ACE_Method_Request
+class SQLOperation
 {
     public:
         SQLOperation(): m_conn(NULL) { }

src/server/shared/Threading/ProducerConsumerQueue.h

@@ -21,6 +21,7 @@
 #include <condition_variable>
 #include <mutex>
 #include <queue>
+#include <atomic>
 
 template <typename T>
 class ProducerConsumerQueue
@@ -29,9 +30,12 @@ private:
     std::mutex _queueLock;
     std::queue<T> _queue;
     std::condition_variable _condition;
+    std::atomic<bool> _shutdown;
 
 public:
 
+    ProducerConsumerQueue<T>() : _shutdown(false) { }
+
     void Push(const T& value)
     {
         _queueLock.lock();
@@ -68,7 +72,10 @@ public:
     {
         std::unique_lock<std::mutex> lock(_queueLock);
 
-        _condition.wait(lock, [this](){ return !_queue.empty(); });
+        while (_queue.empty() && !_shutdown)
+        {
+            _condition.wait(lock);
+        }
         
         if (_queue.empty())
             return;
@@ -91,6 +98,8 @@ public:
             _queue.pop();
         }
 
+        _shutdown = true;
+
         _queueLock.unlock();
 
         _condition.notify_all();