various TSAN/ASAN/LeakAnalyzer fixes (#622)

* fix JobQueue test threading issue. The inner job queued with `QueueJob` would reference `I` from inside the captured closure which would subsequently disappear
* made sure application exit is thread safe
* don't try to access string data out of bounds
* keep-alive flag is accessed from multiple threads
* fix memory leaks in Zen upstream client code
* TSAN fixes for Event

6 files changed, 32 insertions, 10 deletions

diff --git a/src/zencore/include/zencore/string.h b/src/zencore/include/zencore/string.h
index 3aec1647d..b0232d883 100644
--- a/src/zencore/include/zencore/string.h
+++ b/src/zencore/include/zencore/string.h
@@ -638,7 +638,12 @@ ToHexNumber(UnsignedIntegral auto Value, char* OutString)
 bool
 ParseHexNumber(const std::string_view HexString, UnsignedIntegral auto& OutValue)
 {
-	return ParseHexNumber(HexString.data(), sizeof(OutValue) * 2, (uint8_t*)&OutValue);
+	size_t ExpectedCharacterCount = sizeof(OutValue) * 2;
+	if (HexString.size() != ExpectedCharacterCount)
+	{
+		return false;
+	}
+	return ParseHexNumber(HexString.data(), ExpectedCharacterCount, (uint8_t*)&OutValue);
 }
 
 //////////////////////////////////////////////////////////////////////////
diff --git a/src/zencore/jobqueue.cpp b/src/zencore/jobqueue.cpp
index 1755b9fe9..4bcc5c885 100644
--- a/src/zencore/jobqueue.cpp
+++ b/src/zencore/jobqueue.cpp
@@ -422,8 +422,10 @@ TEST_CASE("JobQueue")
 	{
 		JobsLatch.AddCount(1);
 		Pool.ScheduleWork([&Queue, &JobsLatch, I]() {
-			auto _	= MakeGuard([&JobsLatch]() { JobsLatch.CountDown(); });
-			auto Id = Queue->QueueJob(fmt::format("busy {}", I), [&](JobContext& Context) {
+			auto _ = MakeGuard([&JobsLatch]() { JobsLatch.CountDown(); });
+			JobsLatch.AddCount(1);
+			auto Id = Queue->QueueJob(fmt::format("busy {}", I), [&JobsLatch, I](JobContext& Context) {
+				auto $ = MakeGuard([&JobsLatch]() { JobsLatch.CountDown(); });
 				if (Context.IsCancelled())
 				{
 					return;
@@ -523,7 +525,6 @@ TEST_CASE("JobQueue")
 	}
 	JobsLatch.Wait();
 }
-
 #endif
 
 }  // namespace zen
diff --git a/src/zencore/thread.cpp b/src/zencore/thread.cpp
index 149a0d781..cb3aced33 100644
--- a/src/zencore/thread.cpp
+++ b/src/zencore/thread.cpp
@@ -156,6 +156,7 @@ Event::Event()
 	auto* Inner	  = new EventInner();
 	Inner->bSet	  = bInitialState;
 	m_EventHandle = Inner;
+	std::atomic_thread_fence(std::memory_order_release);
 #endif
 }
 
@@ -170,12 +171,13 @@ Event::Set()
 #if ZEN_USE_WINDOWS_EVENTS
 	SetEvent(m_EventHandle);
 #else
-	auto* Inner	  = (EventInner*)m_EventHandle;
+	std::atomic_thread_fence(std::memory_order_acquire);
+	auto* Inner = (EventInner*)m_EventHandle;
 	{
 		std::unique_lock Lock(Inner->Mutex);
 		Inner->bSet.store(true);
+		Inner->CondVar.notify_all();
 	}
-	Inner->CondVar.notify_all();
 #endif
 }
 
@@ -185,6 +187,7 @@ Event::Reset()
 #if ZEN_USE_WINDOWS_EVENTS
 	ResetEvent(m_EventHandle);
 #else
+	std::atomic_thread_fence(std::memory_order_acquire);
 	auto* Inner = (EventInner*)m_EventHandle;
 	{
 		std::unique_lock Lock(Inner->Mutex);
@@ -198,15 +201,18 @@ Event::Close()
 {
 #if ZEN_USE_WINDOWS_EVENTS
 	CloseHandle(m_EventHandle);
+	m_EventHandle = nullptr;
 #else
+	std::atomic_thread_fence(std::memory_order_acquire);
 	auto* Inner = (EventInner*)m_EventHandle;
 	{
 		std::unique_lock Lock(Inner->Mutex);
 		Inner->bSet.store(true);
 	}
+	m_EventHandle = nullptr;
+	std::atomic_thread_fence(std::memory_order_release);
 	delete Inner;
 #endif
-	m_EventHandle = nullptr;
 }
 
 bool
@@ -226,6 +232,7 @@ Event::Wait(int TimeoutMs)
 
 	return (Result == WAIT_OBJECT_0);
 #else
+	std::atomic_thread_fence(std::memory_order_acquire);
 	auto* Inner = reinterpret_cast<EventInner*>(m_EventHandle);
 
 	if (Inner->bSet.load())
diff --git a/src/zencore/zencore.cpp b/src/zencore/zencore.cpp
index eed903f54..d0acac608 100644
--- a/src/zencore/zencore.cpp
+++ b/src/zencore/zencore.cpp
@@ -37,6 +37,8 @@
 
 #include <fmt/format.h>
 
+#include <atomic>
+
 namespace zen::assert {
 
 void
@@ -104,8 +106,8 @@ IsInteractiveSession()
 
 //////////////////////////////////////////////////////////////////////////
 
-static int	s_ApplicationExitCode = 0;
-static bool s_ApplicationExitRequested;
+static std::atomic_int	s_ApplicationExitCode{0};
+static std::atomic_bool s_ApplicationExitRequested{false};
 
 bool
 IsApplicationExitRequested()
diff --git a/src/zenhttp/servers/httpparser.h b/src/zenhttp/servers/httpparser.h
index 219ac351d..bdbcab4d9 100644
--- a/src/zenhttp/servers/httpparser.h
+++ b/src/zenhttp/servers/httpparser.h
@@ -9,6 +9,8 @@ ZEN_THIRD_PARTY_INCLUDES_START
 #include <http_parser.h>
 ZEN_THIRD_PARTY_INCLUDES_END
 
+#include <atomic>
+
 namespace zen {
 
 class HttpRequestParserCallbacks
@@ -85,7 +87,7 @@ private:
 	int8_t						m_ContentTypeHeaderIndex;
 	int8_t						m_RangeHeaderIndex;
 	HttpVerb					m_RequestVerb;
-	bool						m_KeepAlive			= false;
+	std::atomic_bool			m_KeepAlive{false};
 	bool						m_Expect100Continue = false;
 	int							m_RequestId			= -1;
 	Oid							m_SessionId{};
diff --git a/src/zenserver/upstream/zen.cpp b/src/zenserver/upstream/zen.cpp
index 8ae33597a..2d52236b3 100644
--- a/src/zenserver/upstream/zen.cpp
+++ b/src/zenserver/upstream/zen.cpp
@@ -59,6 +59,11 @@ ZenStructuredCacheClient::ZenStructuredCacheClient(const ZenStructuredCacheClien
 
 ZenStructuredCacheClient::~ZenStructuredCacheClient()
 {
+	RwLock::ExclusiveLockScope _(m_SessionStateLock);
+	for (auto& CacheEntry : m_SessionStateCache)
+	{
+		delete CacheEntry;
+	}
 }
 
 detail::ZenCacheSessionState*