Merge branch 'main' into 273-integrated-memory-tracking273-integrated-memory-tracking

6 files changed, 64 insertions, 22 deletions

diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index 71e306eca..73a8ad538 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -51,6 +51,7 @@ BlockStoreFile::GetPath() const
 void
 BlockStoreFile::Open()
 {
+	ZEN_TRACE_CPU("BlockStoreFile::Open");
 	uint32_t RetriesLeft = 3;
 	m_File.Open(m_Path, BasicFile::Mode::kDelete, [&](std::error_code& Ec) {
 		if (RetriesLeft == 0)
@@ -149,6 +150,11 @@ BlockStoreFile::StreamByteRange(uint64_t FileOffset, uint64_t Size, std::functio
 {
 	m_File.StreamByteRange(FileOffset, Size, std::move(ChunkFun));
 }
+bool
+BlockStoreFile::IsOpen() const
+{
+	return !!m_IoBuffer;
+}
 
 constexpr uint64_t ScrubSmallChunkWindowSize = 4 * 1024 * 1024;
 
@@ -285,6 +291,7 @@ BlockStore::SyncExistingBlocksOnDisk(const BlockIndexSet& KnownLocations)
 BlockStore::BlockEntryCountMap
 BlockStore::GetBlocksToCompact(const BlockUsageMap& BlockUsage, uint32_t BlockUsageThresholdPercent)
 {
+	ZEN_TRACE_CPU("BlockStoreFile::GetBlocksToCompact");
 	BlockEntryCountMap Result;
 	{
 		RwLock::SharedLockScope InsertLock(m_InsertLock);
@@ -345,6 +352,7 @@ BlockStore::GetBlocksToCompact(const BlockUsageMap& BlockUsage, uint32_t BlockUs
 void
 BlockStore::Close()
 {
+	ZEN_TRACE_CPU("BlockStore::Close");
 	RwLock::ExclusiveLockScope InsertLock(m_InsertLock);
 	m_WriteBlock		  = nullptr;
 	m_CurrentInsertOffset = 0;
@@ -666,7 +674,7 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 		ZEN_TRACE_CPU("BlockStore::ReclaimSpace::Compact");
 		Ref<BlockStoreFile> NewBlockFile;
 		auto				NewBlockFileGuard = MakeGuard([&]() {
-			   if (NewBlockFile)
+			   if (NewBlockFile && NewBlockFile->IsOpen())
 			   {
 				   ZEN_DEBUG("dropping incomplete cas block store file '{}'", NewBlockFile->GetPath());
 				   m_TotalSize.fetch_sub(NewBlockFile->FileSize(), std::memory_order::relaxed);
@@ -754,10 +762,26 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 			if (OldBlockFile)
 			{
 				ZEN_TRACE_CPU("BlockStore::ReclaimSpace::MoveBlock");
+
+				ZEN_INFO("Moving {} chunks from '{}' to new block", KeepMap.size(), GetBlockPath(m_BlocksBasePath, BlockIndex));
+
+				uint64_t			 OldBlockSize = OldBlockFile->FileSize();
 				std::vector<uint8_t> Chunk;
 				for (const size_t& ChunkIndex : KeepMap)
 				{
 					const BlockStoreLocation ChunkLocation = ChunkLocations[ChunkIndex];
+					if (ChunkLocation.Offset + ChunkLocation.Size > OldBlockSize)
+					{
+						ZEN_WARN(
+							"ReclaimSpace skipping chunk outside of block range in '{}', Chunk start {}, Chunk size {} in Block {}, Block "
+							"size {}",
+							m_BlocksBasePath,
+							ChunkLocation.Offset,
+							ChunkLocation.Size,
+							OldBlockFile->GetPath(),
+							OldBlockSize);
+						continue;
+					}
 					Chunk.resize(ChunkLocation.Size);
 					OldBlockFile->Read(Chunk.data(), ChunkLocation.Size, ChunkLocation.Offset);
 
@@ -767,6 +791,7 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 
 						if (NewBlockFile)
 						{
+							ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
 							NewBlockFile->Flush();
 							NewBlockFile = nullptr;
 						}
@@ -820,8 +845,8 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 									ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
 								});
 								ZEN_ASSERT(m_ChunkBlocks[NextBlockIndex] == NewBlockFile);
+								ZEN_ASSERT_SLOW(!NewBlockFile->IsOpen());
 								m_ChunkBlocks.erase(NextBlockIndex);
-								NewBlockFile->MarkAsDeleteOnClose();
 								return;
 							}
 
@@ -846,6 +871,7 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 				Chunk.clear();
 				if (NewBlockFile)
 				{
+					ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
 					NewBlockFile->Flush();
 				}
 			}
@@ -880,6 +906,7 @@ BlockStore::ReclaimSpace(const ReclaimSnapshotState&			Snapshot,
 		}
 		if (NewBlockFile)
 		{
+			ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
 			NewBlockFile->Flush();
 			NewBlockFile = nullptr;
 		}
@@ -1044,6 +1071,8 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 						  const CompactCallback&		  ChangeCallback,
 						  const ClaimDiskReserveCallback& DiskReserveCallback)
 {
+	ZEN_TRACE_CPU("BlockStore::CompactBlocks");
+
 	uint64_t DeletedSize = 0;
 	uint64_t MovedCount	 = 0;
 	uint64_t MovedSize	 = 0;
@@ -1069,7 +1098,6 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 	auto				NewBlockFileGuard = MakeGuard([&]() {
 		   if (NewBlockFile)
 		   {
-			   ZEN_DEBUG("Dropping incomplete cas block store file '{}'", NewBlockFile->GetPath());
 			   {
 				   RwLock::ExclusiveLockScope _l(m_InsertLock);
 				   if (m_ChunkBlocks[NewBlockIndex] == NewBlockFile)
@@ -1077,7 +1105,11 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 					   m_ChunkBlocks.erase(NewBlockIndex);
 				   }
 			   }
-			   NewBlockFile->MarkAsDeleteOnClose();
+			   if (NewBlockFile->IsOpen())
+			   {
+				   ZEN_DEBUG("Dropping incomplete cas block store file '{}'", NewBlockFile->GetPath());
+				   NewBlockFile->MarkAsDeleteOnClose();
+			   }
 		   }
 	   });
 
@@ -1100,6 +1132,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 	CompactState.IterateBlocks([&](uint32_t								  BlockIndex,
 								   const std::vector<size_t>&			  KeepChunkIndexes,
 								   const std::vector<BlockStoreLocation>& ChunkLocations) -> bool {
+		ZEN_TRACE_CPU("BlockStore::CompactBlock");
 		Ref<BlockStoreFile> OldBlockFile;
 		{
 			RwLock::SharedLockScope _(m_InsertLock);
@@ -1125,6 +1158,8 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 		}
 		ZEN_ASSERT(OldBlockFile);
 
+		ZEN_INFO("Moving {} chunks from '{}' to new block", KeepChunkIndexes.size(), GetBlockPath(m_BlocksBasePath, BlockIndex));
+
 		uint64_t OldBlockSize = OldBlockFile->FileSize();
 
 		std::vector<uint8_t> Chunk;
@@ -1151,6 +1186,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 			{
 				if (NewBlockFile)
 				{
+					ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
 					NewBlockFile->Flush();
 					MovedSize += NewBlockFile->FileSize();
 					NewBlockFile = nullptr;
@@ -1179,7 +1215,6 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 					NewBlockFile				  = new BlockStoreFile(NewBlockPath);
 					m_ChunkBlocks[NextBlockIndex] = NewBlockFile;
 				}
-				ZEN_ASSERT(NewBlockFile);
 
 				std::error_code Error;
 				DiskSpace		Space = DiskSpaceInfo(m_BlocksBasePath, Error);
@@ -1191,7 +1226,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 						ZEN_ASSERT(m_ChunkBlocks[NextBlockIndex] == NewBlockFile);
 						m_ChunkBlocks.erase(NextBlockIndex);
 					}
-					NewBlockFile->MarkAsDeleteOnClose();
+					ZEN_ASSERT_SLOW(!NewBlockFile->IsOpen());
 					NewBlockFile = nullptr;
 					return false;
 				}
@@ -1210,7 +1245,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 							ZEN_ASSERT(m_ChunkBlocks[NextBlockIndex] == NewBlockFile);
 							m_ChunkBlocks.erase(NextBlockIndex);
 						}
-						NewBlockFile->MarkAsDeleteOnClose();
+						ZEN_ASSERT_SLOW(!NewBlockFile->IsOpen());
 						NewBlockFile = nullptr;
 						return false;
 					}
@@ -1251,6 +1286,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 
 	if (NewBlockFile)
 	{
+		ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
 		NewBlockFile->Flush();
 		MovedSize += NewBlockFile->FileSize();
 		NewBlockFile = nullptr;
diff --git a/src/zenstore/cas.cpp b/src/zenstore/cas.cpp
index d38099117..b20f2049a 100644
--- a/src/zenstore/cas.cpp
+++ b/src/zenstore/cas.cpp
@@ -91,7 +91,7 @@ CasImpl::~CasImpl()
 void
 CasImpl::Initialize(const CidStoreConfiguration& InConfig)
 {
-	ZEN_TRACE_CPU("Cas::Initialize");
+	ZEN_TRACE_CPU("CAS::Initialize");
 
 	m_Config = InConfig;
 
@@ -127,6 +127,7 @@ CasImpl::Initialize(const CidStoreConfiguration& InConfig)
 bool
 CasImpl::OpenOrCreateManifest()
 {
+	ZEN_TRACE_CPU("CAS::OpenOrCreateManifest");
 	bool IsNewStore = false;
 
 	std::filesystem::path ManifestPath = m_Config.RootDirectory;
@@ -189,6 +190,7 @@ CasImpl::OpenOrCreateManifest()
 void
 CasImpl::UpdateManifest()
 {
+	ZEN_TRACE_CPU("CAS::UpdateManifest");
 	if (!m_ManifestObject)
 	{
 		CbObjectWriter Cbo;
@@ -266,6 +268,7 @@ CasImpl::ContainsChunk(const IoHash& ChunkHash)
 void
 CasImpl::FilterChunks(HashKeySet& InOutChunks)
 {
+	ZEN_TRACE_CPU("CAS::FilterChunks");
 	m_SmallStrategy.FilterChunks(InOutChunks);
 	m_TinyStrategy.FilterChunks(InOutChunks);
 	m_LargeStrategy.FilterChunks(InOutChunks);
@@ -274,6 +277,7 @@ CasImpl::FilterChunks(HashKeySet& InOutChunks)
 void
 CasImpl::Flush()
 {
+	ZEN_TRACE_CPU("CAS::Flush");
 	ZEN_INFO("flushing CAS pool at '{}'", m_Config.RootDirectory);
 	m_SmallStrategy.Flush();
 	m_TinyStrategy.Flush();
diff --git a/src/zenstore/caslog.cpp b/src/zenstore/caslog.cpp
index cf3bd76da..2c26e522f 100644
--- a/src/zenstore/caslog.cpp
+++ b/src/zenstore/caslog.cpp
@@ -229,7 +229,8 @@ CasLogFile::Append(const void* DataPointer, uint64_t DataSize)
 
 	if (Ec)
 	{
-		throw std::system_error(Ec, fmt::format("Failed to write to log file '{}'", PathFromHandle(m_File.Handle())));
+		std::error_code Dummy;
+		throw std::system_error(Ec, fmt::format("Failed to write to log file '{}'", PathFromHandle(m_File.Handle(), Dummy)));
 	}
 }
 
diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index b21f9f8d8..64c1dadf8 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -244,6 +244,7 @@ CasContainerStrategy::FilterChunks(HashKeySet& InOutChunks)
 void
 CasContainerStrategy::Flush()
 {
+	ZEN_TRACE_CPU("CasContainer::Flush");
 	m_BlockStore.Flush(/*ForceNewBlock*/ false);
 	m_CasLog.Flush();
 	MakeIndexSnapshot();
@@ -470,7 +471,7 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 			size_t						  ChunkIndex   = ChunkLocations.size();
 
 			ChunkLocations.push_back(Location);
-			ChunkIndexToChunkHash[ChunkIndex] = ChunkHash;
+			ChunkIndexToChunkHash.push_back(ChunkHash);
 			if (Keep)
 			{
 				KeepChunkIndexes.push_back(ChunkIndex);
diff --git a/src/zenstore/gc.cpp b/src/zenstore/gc.cpp
index de653b0e3..4cc2c3ed1 100644
--- a/src/zenstore/gc.cpp
+++ b/src/zenstore/gc.cpp
@@ -1876,18 +1876,17 @@ GcScheduler::SchedulerThread()
 					NextTriggerStatus = Sb;
 				}
 
-				ZEN_INFO(
-					"{} used{}. '{}': {} in use, {} free. Disk writes last {} per {} [{}], peak {}/s.{}",
-					NiceBytes(TotalSize.DiskSize),
-					DiskSizeSoftLimit == 0 ? "" : fmt::format(", {} soft limit", NiceBytes(DiskSizeSoftLimit)),
-					m_Config.RootDirectory,
-					NiceBytes(Space.Total - Space.Free),
-					NiceBytes(Space.Free),
-					NiceTimeSpanMs(uint64_t(std::chrono::milliseconds(LoadGraphTime).count())),
-					NiceTimeSpanMs(uint64_t(std::chrono::milliseconds(LoadGraphTime).count() / PressureGraphLength)),
-					LoadGraph,
-					NiceBytes(MaxLoad * uint64_t(std::chrono::seconds(1).count()) / uint64_t(std::chrono::seconds(LoadGraphTime).count())),
-					NextTriggerStatus);
+				ZEN_INFO("{} used{}. '{}': {} in use, {} free. Disk writes last {} per {} [{}], peak {}/s.{}",
+						 NiceBytes(TotalSize.DiskSize),
+						 DiskSizeSoftLimit == 0 ? "" : fmt::format(", {} soft limit", NiceBytes(DiskSizeSoftLimit)),
+						 m_Config.RootDirectory,
+						 NiceBytes(Space.Total - Space.Free),
+						 NiceBytes(Space.Free),
+						 NiceTimeSpanMs(uint64_t(std::chrono::milliseconds(LoadGraphTime).count())),
+						 NiceTimeSpanMs(uint64_t(std::chrono::milliseconds(LoadGraphTime).count() / PressureGraphLength)),
+						 LoadGraph,
+						 NiceBytes(MaxLoad / uint64_t(std::chrono::seconds(m_Config.MonitorInterval).count())),
+						 NextTriggerStatus);
 
 				if (!DiskSpaceGCTriggered && !TimeBasedGCTriggered)
 				{
diff --git a/src/zenstore/include/zenstore/blockstore.h b/src/zenstore/include/zenstore/blockstore.h
index 786780b5e..bb36cb3cd 100644
--- a/src/zenstore/include/zenstore/blockstore.h
+++ b/src/zenstore/include/zenstore/blockstore.h
@@ -101,6 +101,7 @@ struct BlockStoreFile : public RefCounted
 	void						 Flush();
 	BasicFile&					 GetBasicFile();
 	void StreamByteRange(uint64_t FileOffset, uint64_t Size, std::function<void(const void* Data, uint64_t Size)>&& ChunkFun);
+	bool IsOpen() const;
 
 private:
 	const std::filesystem::path m_Path;