fix(Core/Database): Gracefully close database workers (#20936)

* Gracefully close database workers

* Change init order. Such a silly compiler flag

* Fix hang if db connection failed to open

5 files changed, 21 insertions, 55 deletions

diff --git a/src/common/Threading/PCQueue.h b/src/common/Threading/PCQueue.h
index 57f3b08efb..1efb494ba0 100644
--- a/src/common/Threading/PCQueue.h
+++ b/src/common/Threading/PCQueue.h
@@ -28,10 +28,11 @@ private:
     std::mutex _queueLock;
     std::queue<T> _queue;
     std::condition_variable _condition;
+    std::atomic<bool> _cancel;
     std::atomic<bool> _shutdown;
 
 public:
-    ProducerConsumerQueue() : _shutdown(false) { }
+    ProducerConsumerQueue() : _cancel(false), _shutdown(false) { }
 
     void Push(const T& value)
     {
@@ -57,10 +58,8 @@ public:
     {
         std::lock_guard<std::mutex> lock(_queueLock);
 
-        if (_queue.empty() || _shutdown)
-        {
+        if (_queue.empty() || _cancel)
             return false;
-        }
 
         value = _queue.front();
 
@@ -75,21 +74,18 @@ public:
 
         // we could be using .wait(lock, predicate) overload here but it is broken
         // https://connect.microsoft.com/VisualStudio/feedback/details/1098841
-        while (_queue.empty() && !_shutdown)
-        {
+        while (_queue.empty() && !_cancel && !_shutdown)
             _condition.wait(lock);
-        }
 
-        if (_queue.empty() || _shutdown)
-        {
+        if (_queue.empty() || _cancel)
             return;
-        }
 
         value = _queue.front();
 
         _queue.pop();
     }
 
+    // Clears the queue and will immediately stop any consumers
     void Cancel()
     {
         std::unique_lock<std::mutex> lock(_queueLock);
@@ -103,8 +99,15 @@ public:
             _queue.pop();
         }
 
-        _shutdown = true;
+        _cancel = true;
+
+        _condition.notify_all();
+    }
 
+    // Graceful stop, will wait for queue to become empty before stopping consumers
+    void Shutdown()
+    {
+        _shutdown = true;
         _condition.notify_all();
     }
 
diff --git a/src/server/database/Database/DatabaseWorker.cpp b/src/server/database/Database/DatabaseWorker.cpp
index 527eb9b502..49965e1835 100644
--- a/src/server/database/Database/DatabaseWorker.cpp
+++ b/src/server/database/Database/DatabaseWorker.cpp
@@ -23,16 +23,11 @@ DatabaseWorker::DatabaseWorker(ProducerConsumerQueue<SQLOperation*>* newQueue, M
 {
     _connection = connection;
     _queue = newQueue;
-    _cancelationToken = false;
     _workerThread = std::thread(&DatabaseWorker::WorkerThread, this);
 }
 
 DatabaseWorker::~DatabaseWorker()
 {
-    _cancelationToken = true;
-
-    _queue->Cancel();
-
     _workerThread.join();
 }
 
@@ -47,7 +42,7 @@ void DatabaseWorker::WorkerThread()
 
         _queue->WaitAndPop(operation);
 
-        if (_cancelationToken || !operation)
+        if (!operation)
             return;
 
         operation->SetConnection(_connection);
diff --git a/src/server/database/Database/DatabaseWorker.h b/src/server/database/Database/DatabaseWorker.h
index bc6724216d..7ce560b3d1 100644
--- a/src/server/database/Database/DatabaseWorker.h
+++ b/src/server/database/Database/DatabaseWorker.h
@@ -41,8 +41,6 @@ private:
     void WorkerThread();
     std::thread _workerThread;
 
-    std::atomic<bool> _cancelationToken;
-
     DatabaseWorker(DatabaseWorker const& right) = delete;
     DatabaseWorker& operator=(DatabaseWorker const& right) = delete;
 };
diff --git a/src/server/database/Database/DatabaseWorkerPool.cpp b/src/server/database/Database/DatabaseWorkerPool.cpp
index 0fd20d4b47..968e4e4282 100644
--- a/src/server/database/Database/DatabaseWorkerPool.cpp
+++ b/src/server/database/Database/DatabaseWorkerPool.cpp
@@ -112,7 +112,11 @@ uint32 DatabaseWorkerPool<T>::Open()
 template <class T>
 void DatabaseWorkerPool<T>::Close()
 {
-    LOG_INFO("sql.driver", "Closing down DatabasePool '{}'.", GetDatabaseName());
+    LOG_INFO("sql.driver", "Closing down DatabasePool '{}'. Waiting for {} queries to finish...", GetDatabaseName(), _queue->Size());
+
+    // Gracefully close async query queue, worker threads will block when the destructor
+    // is called from the .clear() functions below until the queue is empty
+    _queue->Shutdown();
 
     //! Closes the actualy MySQL connection.
     _connections[IDX_ASYNC].clear();
@@ -432,6 +436,7 @@ uint32 DatabaseWorkerPool<T>::OpenConnections(InternalIndex type, uint8 numConne
         if (uint32 error = connection->Open())
         {
             // Failed to open a connection or invalid version, abort and cleanup
+            _queue->Cancel();
             _connections[type].clear();
             return error;
         }
diff --git a/src/server/game/World/World.cpp b/src/server/game/World/World.cpp
index 72ea18e999..01aebe6187 100644
--- a/src/server/game/World/World.cpp
+++ b/src/server/game/World/World.cpp
@@ -95,11 +95,6 @@
 #include <boost/asio/ip/address.hpp>
 #include <cmath>
 
-namespace
-{
-    TaskScheduler playersSaveScheduler;
-}
-
 std::atomic_long World::_stopEvent = false;
 uint8 World::_exitCode = SHUTDOWN_EXIT_CODE;
 uint32 World::m_worldLoopCounter = 0;
@@ -2498,11 +2493,6 @@ void World::Update(uint32 diff)
     }
 
     {
-        METRIC_TIMER("world_update_time", METRIC_TAG("type", "Update playersSaveScheduler"));
-        playersSaveScheduler.Update(diff);
-    }
-
-    {
         METRIC_TIMER("world_update_time", METRIC_TAG("type", "Update metrics"));
         // Stats logger update
         sMetric->Update();
@@ -2691,31 +2681,6 @@ void World::ShutdownServ(uint32 time, uint32 options, uint8 exitcode, const std:
     _shutdownMask = options;
     _exitCode = exitcode;
 
-    auto const& playersOnline = GetActiveSessionCount();
-
-    if (time < 5 && playersOnline)
-    {
-        // Set time to 5s for save all players
-        time = 5;
-    }
-
-    playersSaveScheduler.CancelAll();
-
-    if (time >= 5)
-    {
-        playersSaveScheduler.Schedule(Seconds(time - 5), [this](TaskContext /*context*/)
-        {
-            if (!GetActiveSessionCount())
-            {
-                LOG_INFO("server", "> No players online. Skip save before shutdown");
-                return;
-            }
-
-            LOG_INFO("server", "> Save players before shutdown server");
-            ObjectAccessor::SaveAllPlayers();
-        });
-    }
-
     LOG_WARN("server", "Time left until shutdown/restart: {}", time);
 
     ///- If the shutdown time is 0, set m_stopEvent (except if shutdown is 'idle' with remaining sessions)