Wait on an event instead of using sleep()

The sleep() means it can take up to "update_interval * 2" (currently 10
seconds) to shutdown the thread. Not only is it desirable that binaries
exit cleanly as quickly as possible, units tests were taking a
considerable amount of time to run due to the sleep.

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;