Added simple GC interval scheduling.

1 files changed, 136 insertions, 19 deletions

diff --git a/zenstore/gc.cpp b/zenstore/gc.cpp
index 1b987ca08..7ab2045b5 100644
--- a/zenstore/gc.cpp
+++ b/zenstore/gc.cpp
@@ -2,14 +2,47 @@
 
 #include <zenstore/gc.h>
 
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinaryvalidation.h>
+#include <zencore/filesystem.h>
 #include <zencore/logging.h>
 #include <zencore/string.h>
 #include <zencore/timer.h>
 #include <zenstore/CAS.h>
 #include <zenstore/cidstore.h>
 
+#include <filesystem>
+
 namespace zen {
 
+using namespace std::literals;
+
+//////////////////////////////////////////////////////////////////////////
+
+CbObject
+LoadCompactBinaryObject(const std::filesystem::path& Path)
+{
+	FileContents Result = ReadFile(Path);
+
+	if (!Result.ErrorCode)
+	{
+		IoBuffer Buffer = Result.Flatten();
+		if (CbValidateError Error = ValidateCompactBinary(Buffer, CbValidateMode::All); Error == CbValidateError::None)
+		{
+			return LoadCompactBinaryObject(Buffer);
+		}
+	}
+
+	return CbObject();
+}
+
+void
+SaveCompactBinaryObject(const std::filesystem::path& Path, const CbObject& Object)
+{
+	WriteFile(Path, Object.GetBuffer().AsIoBuffer());
+}
+
 //////////////////////////////////////////////////////////////////////////
 
 struct GcContext::GcState
@@ -145,16 +178,12 @@ CasGc::RemoveGcStorage(GcStorage* Storage)
 }
 
 void
-CasGc::CollectGarbage()
+CasGc::CollectGarbage(GcContext& GcCtx)
 {
 	RwLock::SharedLockScope _(m_Lock);
 
 	// First gather reference set
 
-	GcContext GcCtx;
-	GcCtx.SetDeletionMode(true);
-	GcCtx.SetContainerGcEnabled(false);
-
 	for (GcContributor* Contributor : m_GcContribs)
 	{
 		Contributor->GatherReferences(GcCtx);
@@ -219,28 +248,116 @@ CasGc::OnDroppedCidReferences(std::span<IoHash> Hashes)
 	ZEN_UNUSED(Hashes);
 }
 
-bool
-Gc::Trigger()
+//////////////////////////////////////////////////////////////////////////
+
+GcScheduler::GcScheduler(CasGc& CasGc) : m_Log(logging::Get("gc")), m_CasGc(CasGc)
+{
+}
+
+GcScheduler::~GcScheduler()
+{
+	Shutdown();
+}
+
+void
+GcScheduler::Initialize(const GcSchedulerConfig& Config)
 {
-	uint32_t Expected = uint32_t(GcStatus::kIdle);
-	if (!m_Status.compare_exchange_strong(Expected, static_cast<uint32_t>(GcStatus::kRunning)))
+	using namespace std::chrono;
+
+	m_Config = Config;
+
+	std::filesystem::create_directories(Config.RootDirectory);
+
+	Timepoint CurrentTime = Clock::now();
+
+	if (CbObject SchedulerState = LoadCompactBinaryObject(Config.RootDirectory / "gc_state"))
 	{
-		return false;
+		const int64_t LastGcTime = SchedulerState["LastGcTime"sv].AsInt64();
+		CurrentTime				 = Timepoint(Clock::duration(LastGcTime));
+
+		if (CurrentTime + m_Config.Interval < Clock::now())
+		{
+			CurrentTime = Clock::now();
+		}
 	}
 
-	ZEN_ASSERT(GcStatus::kRunning == Status());
+	m_NextGcTime = CurrentTime + m_Config.Interval;
+	m_GcThread	 = std::jthread(&GcScheduler::SchedulerThread, this);
+}
 
-	m_GcThread = std::jthread([this]() {
-		Stopwatch Timer;
-		ZEN_INFO("garbage collection STARTING");
+void
+GcScheduler::Shutdown()
+{
+	if (static_cast<uint32_t>(GcSchedulerStatus::kStopped) != m_Status)
+	{
+		m_Status = static_cast<uint32_t>(GcSchedulerStatus::kStopped);
+		m_GcSignal.notify_one();
+	}
+}
+
+void
+GcScheduler::SchedulerThread()
+{
+	for (;;)
+	{
+		{
+			std::unique_lock Lock(m_GcMutex);
+			if (m_GcSignal.wait_until(Lock, m_NextGcTime, [this]() { return Status() != GcSchedulerStatus::kIdle; }))
+			{
+				if (Status() == GcSchedulerStatus::kStopped)
+				{
+					break;
+				}
+			}
+			else
+			{
+				m_Status.store(static_cast<uint32_t>(GcSchedulerStatus::kRunning));
+			}
+		}
+
+		if (m_Config.Enabled)
+		{
+			Stopwatch Timer;
+			ZEN_INFO("garbage collection STARTING");
 
-		m_CasGc.CollectGarbage();
-		m_Status = static_cast<uint32_t>(GcStatus::kIdle);
+			GcContext GcCtx;
+			GcCtx.SetDeletionMode(true);
+			GcCtx.SetContainerGcEnabled(true);
 
-		ZEN_INFO("garbage collection DONE after {}", NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
-	});
+			m_CasGc.CollectGarbage(GcCtx);
 
-	return true;
+			m_Status	 = static_cast<uint32_t>(GcSchedulerStatus::kIdle);
+			m_NextGcTime = Clock::now() + m_Config.Interval;
+
+			{
+				CbObjectWriter SchedulderState;
+				SchedulderState << "LastGcTime"sv << static_cast<int64_t>(Clock::now().time_since_epoch().count());
+				SaveCompactBinaryObject(m_Config.RootDirectory / "gc_state", SchedulderState.Save());
+			}
+
+			ZEN_INFO("garbage collection DONE after {}", NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+		}
+	}
 }
 
+bool
+GcScheduler::ScheduleNow()
+{
+	if (m_Config.Enabled)
+	{
+		ZEN_DEBUG("schedule NOW");
+		std::unique_lock Lock(m_GcMutex);
+		if (static_cast<uint32_t>(GcSchedulerStatus::kIdle) == m_Status)
+		{
+			m_Status = static_cast<uint32_t>(GcSchedulerStatus::kRunning);
+			m_GcSignal.notify_one();
+			return true;
+		}
+	}
+
+	return false;
+}
+
+//////////////////////////////////////////////////////////////////////////
+
 }  // namespace zen