reduce lock contention when checking for disk cache put reject (#465)

keep rawsize and rawhash if available when using batch for inline puts
keep rawsize and rawhash of input value if we have calculated it for validation already

1 files changed, 81 insertions, 91 deletions

diff --git a/src/zenstore/cache/cachedisklayer.cpp b/src/zenstore/cache/cachedisklayer.cpp
index e0030230f..219caca01 100644
--- a/src/zenstore/cache/cachedisklayer.cpp
+++ b/src/zenstore/cache/cachedisklayer.cpp
@@ -234,58 +234,25 @@ using namespace std::literals;
 namespace zen::cache::impl {
 
 static bool
-GetValueRawSizeAndHash(const ZenCacheValue& Value, uint64_t& OutValueRawSize, IoHash& OutValueRawHash)
+UpdateValueWithRawSizeAndHash(ZenCacheValue& Value)
 {
-	if ((Value.RawSize != 0) || (Value.RawHash != IoHash::Zero))
+	if ((Value.RawSize == 0) && (Value.RawHash == IoHash::Zero))
 	{
-		OutValueRawSize = Value.RawSize;
-		OutValueRawHash = Value.RawHash;
-		return true;
-	}
-	else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
-	{
-		return CompressedBuffer::ValidateCompressedHeader(Value.Value, OutValueRawHash, OutValueRawSize);
-	}
-
-	OutValueRawSize = Value.Value.GetSize();
-	OutValueRawHash = IoHash::HashBuffer(Value.Value);
-	return true;
-}
-
-static bool
-ValueMatchesRawSizeAndHash(const ZenCacheValue& Value, uint64_t RawSize, const std::function<IoHash()>& RawHashProvider)
-{
-	if ((Value.RawSize != 0) || (Value.RawHash != IoHash::Zero))
-	{
-		return ((RawSize == Value.RawSize) && (RawHashProvider() == Value.RawHash));
+		if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
+		{
+			return CompressedBuffer::ValidateCompressedHeader(Value.Value, Value.RawHash, Value.RawSize);
+		}
+		else
+		{
+			Value.RawSize = Value.Value.GetSize();
+			Value.RawHash = IoHash::HashBuffer(Value.Value);
+			return true;
+		}
 	}
-	else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
+	else
 	{
-		uint64_t ValueRawSize = 0;
-		IoHash	 ValueRawHash = IoHash::Zero;
-		return CompressedBuffer::ValidateCompressedHeader(Value.Value, ValueRawHash, ValueRawSize) && (RawSize == ValueRawSize) &&
-			   (RawHashProvider() == ValueRawHash);
+		return true;
 	}
-
-	return (RawSize == Value.Value.GetSize()) && (RawHashProvider() == IoHash::HashBuffer(Value.Value));
-}
-
-static bool
-ValueMatchesValue(const ZenCacheValue& Value1, const ZenCacheValue& Value2)
-{
-	if ((Value1.RawSize != 0) || (Value1.RawHash != IoHash::Zero))
-	{
-		return ValueMatchesRawSizeAndHash(Value2, Value1.RawSize, [&Value1]() { return Value1.RawHash; });
-	}
-	else if (Value1.Value.GetContentType() == ZenContentType::kCompressedBinary)
-	{
-		uint64_t Value1RawSize = 0;
-		IoHash	 Value1RawHash = IoHash::Zero;
-		return CompressedBuffer::ValidateCompressedHeader(Value1.Value, Value1RawHash, Value1RawSize) &&
-			   ValueMatchesRawSizeAndHash(Value2, Value1RawSize, [Value1RawHash]() { return Value1RawHash; });
-	}
-
-	return ValueMatchesRawSizeAndHash(Value2, Value1.Value.GetSize(), [&Value1]() { return IoHash::HashBuffer(Value1.Value); });
 }
 
 class BucketManifestSerializer
@@ -1376,9 +1343,9 @@ struct ZenCacheDiskLayer::CacheBucket::PutBatchHandle
 		std::vector<IoHash> HashKeyAndReferences;
 		bool				Overwrite;
 	};
-	std::vector<IoBuffer> Buffers;
-	std::vector<Entry>	  Entries;
-	std::vector<size_t>	  EntryResultIndexes;
+	std::vector<ZenCacheValue> Buffers;
+	std::vector<Entry>		   Entries;
+	std::vector<size_t>		   EntryResultIndexes;
 
 	std::vector<ZenCacheDiskLayer::PutResult>& OutResults;
 };
@@ -1410,22 +1377,21 @@ ZenCacheDiskLayer::CacheBucket::EndPutBatch(PutBatchHandle* Batch) noexcept
 				const std::vector<IoHash>& HashKeyAndReferences = Batch->Entries[Index].HashKeyAndReferences;
 				ZEN_ASSERT(HashKeyAndReferences.size() >= 1);
 
-				ZenCacheValue TemporaryValue;
-				TemporaryValue.Value = Batch->Buffers[Index];
+				ZenCacheValue&			Value = Batch->Buffers[Index];
 				std::span<const IoHash> ReferenceSpan(HashKeyAndReferences.begin() + 1, HashKeyAndReferences.end());
 				PutResult&				OutResult = Batch->OutResults[Batch->EntryResultIndexes[Index]];
 				OutResult						  = PutResult{zen::PutStatus::Success};
-				if (!ShouldRejectPut(HashKeyAndReferences[0], TemporaryValue, ReferenceSpan, Batch->Entries[Index].Overwrite, OutResult))
+				if (!ShouldRejectPut(HashKeyAndReferences[0], Value, Batch->Entries[Index].Overwrite, OutResult))
 				{
 					BufferToEntryIndexes.push_back(Index);
-					BuffersToCommit.push_back(TemporaryValue.Value);
+					BuffersToCommit.push_back(Value.Value);
 
 					uint8_t Flags = 0;
-					if (TemporaryValue.Value.GetContentType() == ZenContentType::kCbObject)
+					if (Value.Value.GetContentType() == ZenContentType::kCbObject)
 					{
 						Flags |= DiskLocation::kStructured;
 					}
-					else if (TemporaryValue.Value.GetContentType() == ZenContentType::kCompressedBinary)
+					else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
 					{
 						Flags |= DiskLocation::kCompressed;
 					}
@@ -1976,13 +1942,10 @@ ZenCacheDiskLayer::CacheBucket::Get(const IoHash& HashKey, ZenCacheValue& OutVal
 
 bool
 ZenCacheDiskLayer::CacheBucket::ShouldRejectPut(const IoHash&				  HashKey,
-												const ZenCacheValue&		  Value,
-												std::span<const IoHash>		  References,
+												ZenCacheValue&				  InOutValue,
 												bool						  Overwrite,
 												ZenCacheDiskLayer::PutResult& OutPutResult)
 {
-	ZEN_UNUSED(References);
-
 	const bool CheckExisting = m_Configuration.LimitOverwrites && !Overwrite;
 	if (CheckExisting)
 	{
@@ -1990,46 +1953,71 @@ ZenCacheDiskLayer::CacheBucket::ShouldRejectPut(const IoHash&				  HashKey,
 		auto					It = m_Index.find(HashKey);
 		if (It != m_Index.end())
 		{
-			PayloadIndex EntryIndex	  = It.value();
-			m_AccessTimes[EntryIndex] = GcClock::TickCount();
-			DiskLocation Location	  = m_Payloads[EntryIndex].Location;
+			const PayloadIndex EntryIndex = It.value();
+			m_AccessTimes[EntryIndex]	  = GcClock::TickCount();
+			const DiskLocation Location	  = m_Payloads[EntryIndex].Location;
 
-			bool				 ComparisonComplete = false;
-			const BucketPayload* Payload			= &m_Payloads[EntryIndex];
+			const BucketPayload* Payload = &m_Payloads[EntryIndex];
 			if (Payload->MetaData)
 			{
-				const BucketMetaData& MetaData = m_MetaDatas[Payload->MetaData];
+				const BucketMetaData MetaData = m_MetaDatas[Payload->MetaData];
 				if (MetaData)
 				{
-					if (!cache::impl::ValueMatchesRawSizeAndHash(Value, MetaData.RawSize, [&MetaData]() { return MetaData.RawHash; }))
+					IndexLock.ReleaseNow();
+					if (!cache::impl::UpdateValueWithRawSizeAndHash(InOutValue))
+					{
+						OutPutResult = PutResult{zen::PutStatus::Fail, "Value provided is of bad format"};
+						return true;
+					}
+					else if (MetaData.RawSize != InOutValue.RawSize || MetaData.RawHash != InOutValue.RawHash)
 					{
 						OutPutResult = PutResult{
 							zen::PutStatus::Conflict,
 							fmt::format("Value exists with different size '{}' or hash '{}'", MetaData.RawSize, MetaData.RawHash)};
 						return true;
 					}
-					ComparisonComplete = true;
+					return false;
 				}
 			}
 
-			if (!ComparisonComplete)
+			ZenCacheValue ExistingValue;
+			if (Payload->MemCached)
 			{
-				// We must release the index lock before calling Get as that will acquire it.
-				// Having the Get method not acquire the index lock is not viable because Get has potentially multiple acquisitions
-				// of the index lock, first as a shared lock, then as an exclusive lock.  If the caller has an exclusive lock, we
-				// could just accept that, and not have Get do any lock acquisitions, but it creates situations where Get is doing
-				// slow operations (eg: reading a file from disk) while under an exclusive index lock, and this would lead to
-				// responsiveness issues where zenserver could fail to respond to requests because of a long-held exclusive lock.
+				ExistingValue.Value = m_MemCachedPayloads[Payload->MemCached].Payload;
 				IndexLock.ReleaseNow();
-				ZenCacheValue ExistingValue;
-				if (Get(HashKey, ExistingValue) && !cache::impl::ValueMatchesValue(Value, ExistingValue))
+			}
+			else
+			{
+				IndexLock.ReleaseNow();
+
+				if (Location.IsFlagSet(DiskLocation::kStandaloneFile))
+				{
+					ExistingValue.Value = GetStandaloneCacheValue(Location, HashKey);
+				}
+				else
 				{
-					uint64_t RawSize = 0;
-					IoHash	 RawHash = IoHash::Zero;
-					cache::impl::GetValueRawSizeAndHash(ExistingValue, RawSize, RawHash);
-					OutPutResult = PutResult{zen::PutStatus::Conflict,
-											 fmt::format("Value exists with different size '{}' or hash '{}'", RawSize, RawHash)};
-					return true;
+					ExistingValue.Value = GetInlineCacheValue(Location);
+				}
+			}
+
+			if (ExistingValue.Value)
+			{
+				if (cache::impl::UpdateValueWithRawSizeAndHash(ExistingValue))
+				{
+					if (!cache::impl::UpdateValueWithRawSizeAndHash(InOutValue))
+					{
+						OutPutResult = PutResult{zen::PutStatus::Fail, "Value provided is of bad format"};
+						return true;
+					}
+
+					if (ExistingValue.RawSize != InOutValue.RawSize || ExistingValue.RawHash != InOutValue.RawHash)
+					{
+						OutPutResult = PutResult{zen::PutStatus::Conflict,
+												 fmt::format("Value exists with different size '{}' or hash '{}'",
+															 ExistingValue.RawSize,
+															 ExistingValue.RawHash)};
+						return true;
+					}
 				}
 			}
 		}
@@ -2052,7 +2040,8 @@ ZenCacheDiskLayer::CacheBucket::Put(const IoHash&		 HashKey,
 
 	if (Value.Value.Size() >= m_Configuration.LargeObjectThreshold)
 	{
-		if (ShouldRejectPut(HashKey, Value, References, Overwrite, Result))
+		ZenCacheValue AcceptedValue = Value;
+		if (ShouldRejectPut(HashKey, AcceptedValue, Overwrite, Result))
 		{
 			if (OptionalBatchHandle)
 			{
@@ -2060,7 +2049,7 @@ ZenCacheDiskLayer::CacheBucket::Put(const IoHash&		 HashKey,
 			}
 			return Result;
 		}
-		PutStandaloneCacheValue(HashKey, Value, References);
+		PutStandaloneCacheValue(HashKey, AcceptedValue, References);
 		if (OptionalBatchHandle)
 		{
 			OptionalBatchHandle->OutResults.push_back({zen::PutStatus::Success});
@@ -2932,7 +2921,7 @@ ZenCacheDiskLayer::CacheBucket::PutInlineCacheValue(const IoHash&		 HashKey,
 	PutResult Result{zen::PutStatus::Success};
 	if (OptionalBatchHandle != nullptr)
 	{
-		OptionalBatchHandle->Buffers.push_back(Value.Value);
+		OptionalBatchHandle->Buffers.push_back(Value);
 		OptionalBatchHandle->Entries.push_back({});
 		OptionalBatchHandle->EntryResultIndexes.push_back(OptionalBatchHandle->OutResults.size());
 		OptionalBatchHandle->OutResults.push_back(PutResult{zen::PutStatus::Fail});
@@ -2945,24 +2934,25 @@ ZenCacheDiskLayer::CacheBucket::PutInlineCacheValue(const IoHash&		 HashKey,
 		return Result;
 	}
 
-	if (ShouldRejectPut(HashKey, Value, References, Overwrite, Result))
+	ZenCacheValue AcceptedValue = Value;
+	if (ShouldRejectPut(HashKey, AcceptedValue, Overwrite, Result))
 	{
 		return Result;
 	}
 
 	uint8_t EntryFlags = 0;
 
-	if (Value.Value.GetContentType() == ZenContentType::kCbObject)
+	if (AcceptedValue.Value.GetContentType() == ZenContentType::kCbObject)
 	{
 		EntryFlags |= DiskLocation::kStructured;
 	}
-	else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
+	else if (AcceptedValue.Value.GetContentType() == ZenContentType::kCompressedBinary)
 	{
 		EntryFlags |= DiskLocation::kCompressed;
 	}
 
-	m_BlockStore.WriteChunk(Value.Value.Data(),
-							Value.Value.Size(),
+	m_BlockStore.WriteChunk(AcceptedValue.Value.Data(),
+							AcceptedValue.Value.Size(),
 							m_Configuration.PayloadAlignment,
 							[&](const BlockStoreLocation& BlockStoreLocation) {
 								ZEN_MEMSCOPE(GetCacheDiskTag());