Step 3

apps/LoopThroughWithOpenGLCompositing/runtime/persistence/PersistenceWriter.cpp

@@ -2,10 +2,96 @@
 
 #include <windows.h>
 
+#include <algorithm>
 #include <filesystem>
 #include <fstream>
+#include <utility>
+
+PersistenceWriter::PersistenceWriter(std::chrono::milliseconds debounceDelay, SnapshotSink sink) :
+	mDebounceDelay(debounceDelay),
+	mSink(std::move(sink))
+{
+}
+
+PersistenceWriter::~PersistenceWriter()
+{
+	StopAndFlush();
+}
 
 bool PersistenceWriter::WriteSnapshot(const PersistenceSnapshot& snapshot, std::string& error) const
+{
+	if (!ValidateSnapshot(snapshot, error))
+		return false;
+
+	return WriteSnapshotThroughSink(snapshot, error);
+}
+
+bool PersistenceWriter::EnqueueSnapshot(const PersistenceSnapshot& snapshot, std::string& error)
+{
+	if (!ValidateSnapshot(snapshot, error))
+		return false;
+
+	std::lock_guard<std::mutex> lock(mMutex);
+	if (!mAcceptingRequests)
+	{
+		error = "Persistence writer is stopping.";
+		return false;
+	}
+
+	StartWorkerLocked();
+
+	const auto now = std::chrono::steady_clock::now();
+	if (snapshot.debounceAllowed)
+	{
+		const std::string debounceKey = snapshot.debounceKey.empty() ? snapshot.targetPath.string() : snapshot.debounceKey;
+		PendingSnapshot& pending = mDebouncedSnapshots[debounceKey];
+		if (!pending.snapshot.targetPath.empty())
+			++mCoalescedCount;
+		else
+			++mEnqueuedCount;
+
+		pending.snapshot = snapshot;
+		pending.readyAt = snapshot.flushRequested ? now : now + mDebounceDelay;
+	}
+	else
+	{
+		mImmediateSnapshots.push_back(snapshot);
+		++mEnqueuedCount;
+	}
+
+	mCondition.notify_one();
+	return true;
+}
+
+void PersistenceWriter::StopAndFlush()
+{
+	{
+		std::lock_guard<std::mutex> lock(mMutex);
+		mAcceptingRequests = false;
+		mStopping = true;
+		const auto now = std::chrono::steady_clock::now();
+		for (auto& entry : mDebouncedSnapshots)
+			entry.second.readyAt = now;
+	}
+	mCondition.notify_all();
+
+	if (mWorker.joinable())
+		mWorker.join();
+}
+
+PersistenceWriterMetrics PersistenceWriter::GetMetrics() const
+{
+	std::lock_guard<std::mutex> lock(mMutex);
+	PersistenceWriterMetrics metrics;
+	metrics.pendingCount = PendingCountLocked();
+	metrics.enqueuedCount = mEnqueuedCount;
+	metrics.coalescedCount = mCoalescedCount;
+	metrics.writtenCount = mWrittenCount;
+	metrics.failedCount = mFailedCount;
+	return metrics;
+}
+
+bool PersistenceWriter::ValidateSnapshot(const PersistenceSnapshot& snapshot, std::string& error) const
 {
 	if (snapshot.targetPath.empty())
 	{
@@ -13,6 +99,14 @@ bool PersistenceWriter::WriteSnapshot(const PersistenceSnapshot& snapshot, std::
 		return false;
 	}
 
+	return true;
+}
+
+bool PersistenceWriter::WriteSnapshotThroughSink(const PersistenceSnapshot& snapshot, std::string& error) const
+{
+	if (mSink)
+		return mSink(snapshot, error);
+
 	std::error_code fsError;
 	std::filesystem::create_directories(snapshot.targetPath.parent_path(), fsError);
 
@@ -42,3 +136,82 @@ bool PersistenceWriter::WriteSnapshot(const PersistenceSnapshot& snapshot, std::
 
 	return true;
 }
+
+void PersistenceWriter::StartWorkerLocked()
+{
+	if (mWorkerRunning)
+		return;
+
+	mWorkerRunning = true;
+	mWorker = std::thread([this]() { WorkerMain(); });
+}
+
+void PersistenceWriter::WorkerMain()
+{
+	for (;;)
+	{
+		PersistenceSnapshot snapshot;
+		{
+			std::unique_lock<std::mutex> lock(mMutex);
+			for (;;)
+			{
+				if (!mImmediateSnapshots.empty())
+				{
+					snapshot = std::move(mImmediateSnapshots.front());
+					mImmediateSnapshots.pop_front();
+					break;
+				}
+
+				if (!mDebouncedSnapshots.empty())
+				{
+					const auto now = std::chrono::steady_clock::now();
+					auto readyIt = mDebouncedSnapshots.end();
+					auto nextReadyAt = (std::chrono::steady_clock::time_point::max)();
+					for (auto it = mDebouncedSnapshots.begin(); it != mDebouncedSnapshots.end(); ++it)
+					{
+						if (it->second.readyAt <= now)
+						{
+							readyIt = it;
+							break;
+						}
+						if (it->second.readyAt < nextReadyAt)
+							nextReadyAt = it->second.readyAt;
+					}
+
+					if (readyIt != mDebouncedSnapshots.end())
+					{
+						snapshot = std::move(readyIt->second.snapshot);
+						mDebouncedSnapshots.erase(readyIt);
+						break;
+					}
+
+					mCondition.wait_until(lock, nextReadyAt);
+					continue;
+				}
+
+				if (mStopping)
+				{
+					mWorkerRunning = false;
+					return;
+				}
+
+				mCondition.wait(lock);
+			}
+		}
+
+		std::string error;
+		const bool succeeded = WriteSnapshotThroughSink(snapshot, error);
+		{
+			std::lock_guard<std::mutex> lock(mMutex);
+			if (succeeded)
+				++mWrittenCount;
+			else
+				++mFailedCount;
+		}
+	}
+}
+
+std::size_t PersistenceWriter::PendingCountLocked() const
+{
+	return mImmediateSnapshots.size() + mDebouncedSnapshots.size();
+}