diff options
| -rw-r--r-- | zenserver/upstream/upstreamapply.cpp | 12 |
1 files changed, 8 insertions, 4 deletions
diff --git a/zenserver/upstream/upstreamapply.cpp b/zenserver/upstream/upstreamapply.cpp index b96adef2a..6ff5c5da2 100644 --- a/zenserver/upstream/upstreamapply.cpp +++ b/zenserver/upstream/upstreamapply.cpp @@ -17,6 +17,7 @@ #include <zencore/session.h> #include <zencore/stats.h> #include <zencore/stream.h> +#include <zencore/thread.h> #include <zencore/timer.h> #include <zenstore/cas.h> @@ -1202,6 +1203,8 @@ public: if (m_RunState.IsRunning) { + m_ShutdownEvent.Reset(); + for (uint32_t Idx = 0; Idx < m_Options.ThreadCount; Idx++) { m_UpstreamThreads.emplace_back(&DefaultUpstreamApply::ProcessUpstreamQueue, this); @@ -1425,11 +1428,9 @@ private: void ProcessUpstreamUpdates() { - const auto& UpdateSleep = std::chrono::seconds(m_Options.UpdatesInterval); - for (;;) + const auto& UpdateSleep = std::chrono::milliseconds(m_Options.UpdatesInterval); + while (!m_ShutdownEvent.Wait(uint32_t(UpdateSleep.count()))) { - std::this_thread::sleep_for(UpdateSleep); - if (!m_RunState.IsRunning) { break; @@ -1492,6 +1493,8 @@ private: { if (m_RunState.Stop()) { + m_ShutdownEvent.Set(); + m_UpstreamQueue.CompleteAdding(); for (std::thread& Thread : m_UpstreamThreads) { @@ -1539,6 +1542,7 @@ private: UpstreamApplyTasks m_ApplyTasks; std::mutex m_ApplyTasksMutex; std::vector<std::unique_ptr<UpstreamApplyEndpoint>> m_Endpoints; + Event m_ShutdownEvent; std::vector<std::thread> m_UpstreamThreads; std::thread m_UpstreamUpdatesThread; std::thread m_EndpointMonitorThread; |