Add ThreadSafeT helper

Wraps a templated value and a templated mutex (defaults to shared_mutex)
and provides safe access, locked with either a shared_lock (const value)
or lock_guard (non-const value).
Applies this to existingEntities.

3 files changed, 55 insertions, 5 deletions

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index 3e6e307..e8dceb8 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -325,7 +325,7 @@ namespace WebStat {
 		}
 		catch (const std::exception & excp) {
 			log(LOG_ERR, "Unhandled exception: %s, clearing known entity list", excp.what());
-			existingEntities.clear();
+			existingEntities()->clear();
 		}
 		auto count = std::distance(processingLines.begin(), storedEnd);
 		processingLines.erase(processingLines.begin(), storedEnd);
@@ -372,7 +372,7 @@ namespace WebStat {
 				try {
 					DB::TransactionScope lineTx {*dbconn};
 					storeNewEntities(dbconn, valuesEntities);
-					existingEntities.insert_range(valuesEntities | entityIds);
+					existingEntities()->insert_range(valuesEntities | entityIds);
 					storeLogLine(dbconn, values);
 				}
 				catch (const DB::Error & originalError) {
@@ -558,8 +558,9 @@ namespace WebStat {
 	void
 	Ingestor::fillKnownEntities(const std::span<Entity *> entities) const
 	{
+		auto lockedEntities = existingEntities.shared();
 		for (const auto entity : entities) {
-			if (auto existing = existingEntities.find(entity->hash); existing != existingEntities.end()) {
+			if (auto existing = lockedEntities->find(entity->hash); existing != lockedEntities->end()) {
 				entity->id = existing->second;
 			}
 		}
@@ -639,7 +640,7 @@ namespace WebStat {
 				"Statistics: linesQueued %zu, linesRead %zu, linesParsed %zu, linesParseFailed %zu, logsInserted %zu, "
 				"entitiesInserted %zu, entitiesKnown %zu",
 				queuedLines.size(), stats.linesRead, stats.linesParsed, stats.linesParseFailed, stats.logsInserted,
-				stats.entitiesInserted, existingEntities.size());
+				stats.entitiesInserted, existingEntities->size());
 	}
 
 	void
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index 738357b..c2a47a4 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -99,7 +99,7 @@ namespace WebStat {
 		DB::ConnectionPoolPtr dbpool;
 		mutable Stats stats {};
 
-		std::map<EntityHash, EntityId> existingEntities;
+		ThreadSafeT<std::map<EntityHash, EntityId>> existingEntities;
 		LineBatch queuedLines, processingLines;
 
 		bool terminated = false;
diff --git a/src/util.hpp b/src/util.hpp
index f7254e8..5cac5a3 100644
--- a/src/util.hpp
+++ b/src/util.hpp
@@ -3,6 +3,7 @@
 #include <chrono>
 #include <command.h>
 #include <scn/scan.h>
+#include <shared_mutex>
 #include <tuple>
 
 namespace WebStat {
@@ -95,4 +96,52 @@ namespace WebStat {
 		}
 		return false;
 	}
+
+	template<typename ValueType, typename MutexType = std::shared_mutex> class ThreadSafeT {
+	public:
+		template<typename... P> ThreadSafeT(P &&... params) : value {std::forward<P>(params)...} { }
+
+		template<typename LockedValueType, typename LockType> class Locked {
+		public:
+			Locked(LockedValueType & valueRef, MutexType & mutex) : value {valueRef}, lock {mutex} { }
+
+			LockedValueType *
+			operator->()
+			{
+				return &value;
+			}
+
+		private:
+			LockedValueType & value;
+			LockType lock;
+		};
+
+		Locked<const ValueType, std::shared_lock<MutexType>>
+		shared() const
+		{
+			return {value, mutex};
+		}
+
+		Locked<ValueType, std::lock_guard<MutexType>>
+		unique()
+		{
+			return {value, mutex};
+		}
+
+		Locked<const ValueType, std::shared_lock<MutexType>>
+		operator->() const
+		{
+			return shared();
+		}
+
+		Locked<ValueType, std::lock_guard<MutexType>>
+		operator()()
+		{
+			return unique();
+		}
+
+	private:
+		ValueType value;
+		mutable MutexType mutex;
+	};
 }