reduce blocking in scrub (#743)

* reduce held locks while performing scrub operation

1 files changed, 86 insertions, 56 deletions

diff --git a/src/zenstore/cache/cachedisklayer.cpp b/src/zenstore/cache/cachedisklayer.cpp
index b2e045632..ead7e4f3a 100644
--- a/src/zenstore/cache/cachedisklayer.cpp
+++ b/src/zenstore/cache/cachedisklayer.cpp
@@ -2410,74 +2410,95 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 
 	try
 	{
-		std::vector<BlockStoreLocation> ChunkLocations;
-		std::vector<IoHash>				ChunkIndexToChunkHash;
+		std::vector<DiskLocation> ChunkLocations;
+		std::vector<IoHash>		  ChunkIndexToChunkHash;
+		std::vector<DiskLocation> StandaloneLocations;
+		std::vector<IoHash>		  StandaloneIndexToKeysHash;
 
-		RwLock::SharedLockScope _(m_IndexLock);
+		{
+			RwLock::SharedLockScope _(m_IndexLock);
 
-		const size_t BlockChunkInitialCount = m_Index.size() / 4;
-		ChunkLocations.reserve(BlockChunkInitialCount);
-		ChunkIndexToChunkHash.reserve(BlockChunkInitialCount);
+			const size_t InitialCount = m_Index.size() / 4;
+			ChunkLocations.reserve(InitialCount);
+			ChunkIndexToChunkHash.reserve(InitialCount);
+			StandaloneLocations.reserve(InitialCount);
+			StandaloneIndexToKeysHash.reserve(InitialCount);
 
-		// Do a pass over the index and verify any standalone file values straight away
-		// all other storage classes are gathered and verified in bulk in order to enable
-		// more efficient I/O scheduling
+			for (auto& Kv : m_Index)
+			{
+				const IoHash&		 HashKey = Kv.first;
+				const BucketPayload& Payload = m_Payloads[Kv.second];
+				const DiskLocation&	 Loc	 = Payload.Location;
 
-		for (auto& Kv : m_Index)
-		{
-			const IoHash&		 HashKey = Kv.first;
-			const BucketPayload& Payload = m_Payloads[Kv.second];
-			const DiskLocation&	 Loc	 = Payload.Location;
+				Ctx.ThrowIfDeadlineExpired();
 
+				if (Loc.IsFlagSet(DiskLocation::kStandaloneFile))
+				{
+					StandaloneLocations.push_back(Loc);
+					StandaloneIndexToKeysHash.push_back(HashKey);
+				}
+				else
+				{
+					ChunkLocations.push_back(Loc);
+					ChunkIndexToChunkHash.push_back(HashKey);
+				}
+			}
+		}
+
+		for (size_t StandaloneKeyIndex = 0; StandaloneKeyIndex < StandaloneIndexToKeysHash.size(); StandaloneKeyIndex++)
+		{
 			Ctx.ThrowIfDeadlineExpired();
 
-			if (Loc.IsFlagSet(DiskLocation::kStandaloneFile))
-			{
-				ChunkCount.fetch_add(1);
-				VerifiedChunkBytes.fetch_add(Loc.Size());
+			const IoHash&		HashKey = StandaloneIndexToKeysHash[StandaloneKeyIndex];
+			const DiskLocation& Loc		= StandaloneLocations[StandaloneKeyIndex];
 
-				if (Loc.GetContentType() == ZenContentType::kBinary)
-				{
-					// Blob cache value, not much we can do about data integrity checking
-					// here since there's no hash available
-					ExtendablePathBuilder<256> DataFilePath;
-					BuildPath(DataFilePath, HashKey);
+			ChunkCount.fetch_add(1);
+			VerifiedChunkBytes.fetch_add(Loc.Size());
 
-					RwLock::SharedLockScope ValueLock(LockForHash(HashKey));
+			if (Loc.GetContentType() == ZenContentType::kBinary)
+			{
+				// Blob cache value, not much we can do about data integrity checking
+				// here since there's no hash available
+				ExtendablePathBuilder<256> DataFilePath;
+				BuildPath(DataFilePath, HashKey);
 
-					std::error_code Ec;
-					uintmax_t		size = FileSizeFromPath(DataFilePath.ToPath(), Ec);
-					if (Ec)
-					{
-						ReportBadKey(HashKey);
-					}
-					if (size != Loc.Size())
-					{
-						ReportBadKey(HashKey);
-					}
-					continue;
+				RwLock::SharedLockScope ValueLock(LockForHash(HashKey));
+
+				std::error_code Ec;
+				uintmax_t		Size = FileSizeFromPath(DataFilePath.ToPath(), Ec);
+				if (Ec)
+				{
+					ReportBadKey(HashKey);
 				}
-				else
+				ValueLock.ReleaseNow();
+
+				if (Size != Loc.Size())
 				{
-					// Structured cache value
-					IoBuffer Buffer = GetStandaloneCacheValue(Loc, HashKey);
-					if (!Buffer)
+					// Make sure we verify that size hasn't changed behind our back...
+					RwLock::SharedLockScope _(m_IndexLock);
+					if (auto It = m_Index.find(HashKey); It != m_Index.end())
 					{
-						ReportBadKey(HashKey);
-						continue;
-					}
-					if (!ValidateIoBuffer(Loc.GetContentType(), std::move(Buffer)))
-					{
-						ReportBadKey(HashKey);
-						continue;
+						const BucketPayload& Payload	= m_Payloads[It->second];
+						const DiskLocation&	 CurrentLoc = Payload.Location;
+						if (Size != CurrentLoc.Size())
+						{
+							ReportBadKey(HashKey);
+						}
 					}
 				}
 			}
 			else
 			{
-				ChunkLocations.emplace_back(Loc.GetBlockLocation(m_Configuration.PayloadAlignment));
-				ChunkIndexToChunkHash.push_back(HashKey);
-				continue;
+				// Structured cache value
+				IoBuffer Buffer = GetStandaloneCacheValue(Loc, HashKey);
+				if (!Buffer)
+				{
+					ReportBadKey(HashKey);
+				}
+				else if (!ValidateIoBuffer(Loc.GetContentType(), std::move(Buffer)))
+				{
+					ReportBadKey(HashKey);
+				}
 			}
 		}
 
@@ -2502,8 +2523,9 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 				ReportBadKey(Hash);
 				return true;
 			}
-			const BucketPayload& Payload	 = m_Payloads[m_Index.at(Hash)];
-			ZenContentType		 ContentType = Payload.Location.GetContentType();
+
+			const DiskLocation& Loc			= ChunkLocations[ChunkIndex];
+			ZenContentType		ContentType = Loc.GetContentType();
 			Buffer.SetContentType(ContentType);
 			if (!ValidateIoBuffer(ContentType, std::move(Buffer)))
 			{
@@ -2525,8 +2547,8 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 				ReportBadKey(Hash);
 				return true;
 			}
-			const BucketPayload& Payload	 = m_Payloads[m_Index.at(Hash)];
-			ZenContentType		 ContentType = Payload.Location.GetContentType();
+			const DiskLocation& Loc			= ChunkLocations[ChunkIndex];
+			ZenContentType		ContentType = Loc.GetContentType();
 			Buffer.SetContentType(ContentType);
 			if (!ValidateIoBuffer(ContentType, std::move(Buffer)))
 			{
@@ -2536,8 +2558,16 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 			return true;
 		};
 
-		m_BlockStore.IterateChunks(ChunkLocations, [&](uint32_t, std::span<const size_t> ChunkIndexes) {
-			return m_BlockStore.IterateBlock(ChunkLocations, ChunkIndexes, ValidateSmallChunk, ValidateLargeChunk, 0);
+		std::vector<BlockStoreLocation> ChunkBlockLocations;
+		ChunkBlockLocations.reserve(ChunkLocations.size());
+
+		for (const DiskLocation& Loc : ChunkLocations)
+		{
+			ChunkBlockLocations.push_back(Loc.GetBlockLocation(m_Configuration.PayloadAlignment));
+		}
+
+		m_BlockStore.IterateChunks(ChunkBlockLocations, [&](uint32_t, std::span<const size_t> ChunkIndexes) {
+			return m_BlockStore.IterateBlock(ChunkBlockLocations, ChunkIndexes, ValidateSmallChunk, ValidateLargeChunk, 0);
 		});
 	}
 	catch (ScrubDeadlineExpiredException&)