Merge pull request #58 from EpicGames/de/cas-store-with-block-store

de/cas store with block store

5 files changed, 80 insertions, 48 deletions

diff --git a/zenserver/cache/structuredcachestore.cpp b/zenserver/cache/structuredcachestore.cpp
index 769167433..738e4c1fd 100644
--- a/zenserver/cache/structuredcachestore.cpp
+++ b/zenserver/cache/structuredcachestore.cpp
@@ -59,7 +59,11 @@ SaveCompactBinaryObject(const fs::path& Path, const CbObject& Object)
 	WriteFile(Path, Object.GetBuffer().AsIoBuffer());
 }
 
-ZenCacheStore::ZenCacheStore(CasGc& Gc, const std::filesystem::path& RootDir) : GcStorage(Gc), GcContributor(Gc), m_DiskLayer(RootDir)
+ZenCacheStore::ZenCacheStore(CasGc& Gc, const std::filesystem::path& RootDir)
+: GcStorage(Gc)
+, GcContributor(Gc)
+, m_RootDir(RootDir)
+, m_DiskLayer(RootDir)
 {
 	ZEN_INFO("initializing structured cache at '{}'", RootDir);
 	CreateDirectories(RootDir);
@@ -188,6 +192,10 @@ ZenCacheStore::Scrub(ScrubContext& Ctx)
 void
 ZenCacheStore::GatherReferences(GcContext& GcCtx)
 {
+	Stopwatch  Timer;
+	const auto Guard = MakeGuard(
+		[this, &Timer] { ZEN_INFO("cache gathered all references from '{}' in {}", m_RootDir, NiceTimeSpanMs(Timer.GetElapsedTimeMs())); });
+
 	access_tracking::AccessTimes AccessTimes;
 	m_MemLayer.GatherAccessTimes(AccessTimes);
 
@@ -476,25 +484,27 @@ ZenCacheDiskLayer::CacheBucket::OpenLog(const fs::path& BucketDir, const bool Is
 	std::filesystem::path SobsPath{BucketDir / "zen.sobs"};
 	std::filesystem::path SlogPath{BucketDir / "zen.slog"};
 
-	m_SobsFile.Open(SobsPath, IsNew);
-	m_SlogFile.Open(SlogPath, IsNew);
+	m_SobsFile.Open(SobsPath, IsNew ? BasicFile::Mode::kTruncate : BasicFile::Mode::kWrite);
+	m_SlogFile.Open(SlogPath, IsNew ? CasLogFile::Mode::kTruncate : CasLogFile::Mode::kWrite);
 
-	m_SlogFile.Replay([&](const DiskIndexEntry& Entry) {
-		if (Entry.Key == IoHash::Zero)
-		{
-			++InvalidEntryCount;
-		}
-		else if (Entry.Location.IsFlagSet(DiskLocation::kTombStone))
-		{
-			m_TotalSize.fetch_sub(Entry.Location.Size(), std::memory_order::relaxed);
-		}
-		else
-		{
-			m_Index.insert_or_assign(Entry.Key, IndexEntry(Entry.Location, GcClock::TickCount()));
-			m_TotalSize.fetch_add(Entry.Location.Size(), std::memory_order::relaxed);
-		}
-		MaxFileOffset = std::max<uint64_t>(MaxFileOffset, Entry.Location.Offset() + Entry.Location.Size());
-	});
+	m_SlogFile.Replay(
+		[&](const DiskIndexEntry& Entry) {
+			if (Entry.Key == IoHash::Zero)
+			{
+				++InvalidEntryCount;
+			}
+			else if (Entry.Location.IsFlagSet(DiskLocation::kTombStone))
+			{
+				m_TotalSize.fetch_sub(Entry.Location.Size(), std::memory_order::relaxed);
+			}
+			else
+			{
+				m_Index.insert_or_assign(Entry.Key, IndexEntry(Entry.Location, GcClock::TickCount()));
+				m_TotalSize.fetch_add(Entry.Location.Size(), std::memory_order::relaxed);
+			}
+			MaxFileOffset = std::max<uint64_t>(MaxFileOffset, Entry.Location.Offset() + Entry.Location.Size());
+		},
+		0);
 
 	if (InvalidEntryCount)
 	{
@@ -757,6 +767,10 @@ ZenCacheDiskLayer::CacheBucket::GatherReferences(GcContext& GcCtx)
 {
 	ZEN_TRACE_CPU("Z$::DiskLayer::CacheBucket::GatherReferences");
 
+	Stopwatch  Timer;
+	const auto Guard = MakeGuard(
+		[this, &Timer] { ZEN_INFO("gathered references from '{}' in {}", m_BucketDir, NiceTimeSpanMs(Timer.GetElapsedTimeMs())); });
+
 	const GcClock::TimePoint ExpireTime =
 		GcCtx.MaxCacheDuration() == GcClock::Duration::max() ? GcClock::TimePoint::min() : GcCtx.Time() - GcCtx.MaxCacheDuration();
 
@@ -905,8 +919,8 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 			m_SlogFile.Close();
 
 			const bool IsNew = true;
-			m_SobsFile.Open(m_BucketDir / "zen.sobs", IsNew);
-			m_SlogFile.Open(m_BucketDir / "zen.slog", IsNew);
+			m_SobsFile.Open(m_BucketDir / "zen.sobs", IsNew ? BasicFile::Mode::kTruncate : BasicFile::Mode::kWrite);
+			m_SlogFile.Open(m_BucketDir / "zen.slog", IsNew ? CasLogFile::Mode::kTruncate : CasLogFile::Mode::kWrite);
 
 			m_SobsCursor = 0;
 			m_TotalSize	 = 0;
@@ -967,8 +981,8 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 			uint64_t					TmpCursor{};
 			std::vector<uint8_t>		Chunk;
 
-			TmpSobs.Open(TmpSobsPath, true);
-			TmpLog.Open(TmpSlogPath, true);
+			TmpSobs.Open(TmpSobsPath, BasicFile::Mode::kTruncate);
+			TmpLog.Open(TmpSlogPath, CasLogFile::Mode::kTruncate);
 
 			for (const auto& Entry : ValidEntries)
 			{
diff --git a/zenserver/config.cpp b/zenserver/config.cpp
index b7fc18b4e..ac0f863cc 100644
--- a/zenserver/config.cpp
+++ b/zenserver/config.cpp
@@ -428,6 +428,13 @@ ParseCliOptions(int argc, char* argv[], ZenServerOptions& ServerOptions)
 					   "Max duration in seconds before Z$ entries get evicted.",
 					   cxxopts::value<int32_t>(ServerOptions.GcConfig.Cache.MaxDurationSeconds)->default_value("86400"),
 					   "");
+
+	options.add_option("gc",
+					   "",
+					   "disk-reserve-size",
+					   "Size of gc disk reserve in bytes.",
+					   cxxopts::value<uint64_t>(ServerOptions.GcConfig.DiskReserveSize)->default_value("268435456"),
+					   "");
 	try
 	{
 		auto result			  = options.parse(argc, argv);
@@ -699,6 +706,7 @@ ParseConfigFile(const std::filesystem::path& Path, ZenServerOptions& ServerOptio
 		if (sol::optional<sol::table> GcConfig = lua["gc"])
 		{
 			ServerOptions.GcConfig.IntervalSeconds = GcConfig.value().get_or("intervalseconds", 0);
+			ServerOptions.GcConfig.DiskReserveSize = GcConfig.value().get_or("diskreservesize", uint64_t(1u << 28));
 
 			if (sol::optional<sol::table> CacheGcConfig = GcConfig.value()["cache"])
 			{
diff --git a/zenserver/config.h b/zenserver/config.h
index a61a7f89f..9f1b3645c 100644
--- a/zenserver/config.h
+++ b/zenserver/config.h
@@ -91,6 +91,7 @@ struct ZenGcConfig
 	int32_t				   IntervalSeconds	   = 0;
 	bool				   CollectSmallObjects = true;
 	bool				   Enabled			   = true;
+	uint64_t			   DiskReserveSize	   = 1ul << 28;
 };
 
 struct ZenServerOptions
diff --git a/zenserver/projectstore.cpp b/zenserver/projectstore.cpp
index 58b806989..617f50660 100644
--- a/zenserver/projectstore.cpp
+++ b/zenserver/projectstore.cpp
@@ -8,6 +8,7 @@
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
 #include <zencore/logging.h>
+#include <zencore/scopeguard.h>
 #include <zencore/stream.h>
 #include <zencore/string.h>
 #include <zencore/testing.h>
@@ -114,10 +115,10 @@ struct ProjectStore::OplogStorage : public RefCounted
 			CreateDirectories(m_OplogStoragePath);
 		}
 
-		m_Oplog.Open(m_OplogStoragePath / "ops.zlog", IsCreate);
+		m_Oplog.Open(m_OplogStoragePath / "ops.zlog", IsCreate ? CasLogFile::Mode::kTruncate : CasLogFile::Mode::kWrite);
 		m_Oplog.Initialize();
 
-		m_OpBlobs.Open(m_OplogStoragePath / "ops.zops", IsCreate);
+		m_OpBlobs.Open(m_OplogStoragePath / "ops.zops", IsCreate ? BasicFile::Mode::kTruncate : BasicFile::Mode::kWrite);
 
 		ZEN_ASSERT(IsPow2(m_OpsAlign));
 		ZEN_ASSERT(!(m_NextOpsOffset & (m_OpsAlign - 1)));
@@ -180,36 +181,39 @@ struct ProjectStore::OplogStorage : public RefCounted
 
 		uint64_t InvalidEntries = 0;
 
-		m_Oplog.Replay([&](const zen::OplogEntry& LogEntry) {
-			if (LogEntry.OpCoreSize == 0)
-			{
-				++InvalidEntries;
+		m_Oplog.Replay(
+			[&](const zen::OplogEntry& LogEntry) {
+				if (LogEntry.OpCoreSize == 0)
+				{
+					++InvalidEntries;
 
-				return;
-			}
+					return;
+				}
 
-			IoBuffer OpBuffer(LogEntry.OpCoreSize);
+				IoBuffer OpBuffer(LogEntry.OpCoreSize);
 
-			const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
+				const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
 
-			m_OpBlobs.Read((void*)OpBuffer.Data(), LogEntry.OpCoreSize, OpFileOffset);
+				m_OpBlobs.Read((void*)OpBuffer.Data(), LogEntry.OpCoreSize, OpFileOffset);
 
-			// Verify checksum, ignore op data if incorrect
-			const auto OpCoreHash = uint32_t(XXH3_64bits(OpBuffer.Data(), OpBuffer.Size()) & 0xffffFFFF);
+				// Verify checksum, ignore op data if incorrect
+				const auto OpCoreHash = uint32_t(XXH3_64bits(OpBuffer.Data(), OpBuffer.Size()) & 0xffffFFFF);
 
-			if (OpCoreHash != LogEntry.OpCoreHash)
-			{
-				ZEN_WARN("skipping oplog entry with bad checksum!");
-				return;
-			}
+				if (OpCoreHash != LogEntry.OpCoreHash)
+				{
+					ZEN_WARN("skipping oplog entry with bad checksum!");
+					return;
+				}
 
-			CbObject Op(SharedBuffer::MakeView(OpBuffer.Data(), OpBuffer.Size()));
+				CbObject Op(SharedBuffer::MakeView(OpBuffer.Data(), OpBuffer.Size()));
 
-			m_NextOpsOffset = Max(m_NextOpsOffset.load(std::memory_order_relaxed), RoundUp(OpFileOffset + LogEntry.OpCoreSize, m_OpsAlign));
-			m_MaxLsn		= Max(m_MaxLsn.load(std::memory_order_relaxed), LogEntry.OpLsn);
+				m_NextOpsOffset =
+					Max(m_NextOpsOffset.load(std::memory_order_relaxed), RoundUp(OpFileOffset + LogEntry.OpCoreSize, m_OpsAlign));
+				m_MaxLsn = Max(m_MaxLsn.load(std::memory_order_relaxed), LogEntry.OpLsn);
 
-			Handler(Op, LogEntry);
-		});
+				Handler(Op, LogEntry);
+			},
+			0);
 
 		if (InvalidEntries)
 		{
@@ -653,7 +657,7 @@ ProjectStore::Project::Read()
 	ZEN_INFO("reading config for project '{}' from {}", Identifier, ProjectStateFilePath);
 
 	BasicFile Blob;
-	Blob.Open(ProjectStateFilePath, false);
+	Blob.Open(ProjectStateFilePath, BasicFile::Mode::kRead);
 
 	IoBuffer		Obj				= Blob.ReadAll();
 	CbValidateError ValidationError = ValidateCompactBinary(MemoryView(Obj.Data(), Obj.Size()), CbValidateMode::All);
@@ -693,7 +697,7 @@ ProjectStore::Project::Write()
 	ZEN_INFO("persisting config for project '{}' to {}", Identifier, ProjectStateFilePath);
 
 	BasicFile Blob;
-	Blob.Open(ProjectStateFilePath, true);
+	Blob.Open(ProjectStateFilePath, BasicFile::Mode::kTruncate);
 	Blob.Write(Mem.Data(), Mem.Size(), 0);
 	Blob.Flush();
 }
@@ -970,6 +974,10 @@ ProjectStore::Scrub(ScrubContext& Ctx)
 void
 ProjectStore::GatherReferences(GcContext& GcCtx)
 {
+	Stopwatch  Timer;
+	const auto Guard =
+		MakeGuard([this, &Timer] { ZEN_INFO("project store gathered all references in {}", NiceTimeSpanMs(Timer.GetElapsedTimeMs())); });
+
 	DiscoverProjects();
 
 	RwLock::SharedLockScope _(m_ProjectsLock);
diff --git a/zenserver/zenserver.cpp b/zenserver/zenserver.cpp
index 667bcd317..f81deb167 100644
--- a/zenserver/zenserver.cpp
+++ b/zenserver/zenserver.cpp
@@ -369,6 +369,7 @@ public:
 			.MaxCacheDuration	 = std::chrono::seconds(ServerOptions.GcConfig.Cache.MaxDurationSeconds),
 			.CollectSmallObjects = ServerOptions.GcConfig.CollectSmallObjects,
 			.Enabled			 = ServerOptions.GcConfig.Enabled,
+			.DiskReserveSize	 = ServerOptions.GcConfig.DiskReserveSize,
 		};
 		m_GcScheduler.Initialize(GcConfig);