optimizations (#200)

* Use direct file read and direct buffer allocation for small IoBuffer materalization
* Reduce range of materialized data in CompositeBuffer reading
CompressedBuffer header reading often only need a small part and not the whole file
* reduce lock contention in IoBuffer::Materialize
* Reduce parsing of compressed headers
Validate header type at decompression
* faster CreateDirectories - start from leaf going up and recurse back
* optimized BufferHeader::IsValid
* Add ValidateCompressedHeader to use when we don't need the actual compressed data
Validate that we always get compressed data in CidStore::AddChunk
* changelog

1 files changed, 53 insertions, 33 deletions

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index bc06653b9..6e5422007 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -214,7 +214,9 @@ namespace detail {
 								Result.ElapsedSeconds += AttachmentResult.ElapsedSeconds;
 								Result.ErrorCode = AttachmentResult.ErrorCode;
 
-								if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response)))
+								IoHash	 RawHash;
+								uint64_t RawSize;
+								if (CompressedBuffer::ValidateCompressedHeader(AttachmentResult.Response, RawHash, RawSize))
 								{
 									Result.Response = AttachmentResult.Response;
 									++NumAttachments;
@@ -251,7 +253,10 @@ namespace detail {
 								Result.ElapsedSeconds += AttachmentResult.ElapsedSeconds;
 								Result.ErrorCode = AttachmentResult.ErrorCode;
 
-								if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response)))
+								IoHash	 RawHash;
+								uint64_t RawSize;
+								if (CompressedBuffer Chunk =
+										CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response), RawHash, RawSize))
 								{
 									Package.AddAttachment(CbAttachment(Chunk, AttachmentHash.AsHash()));
 								}
@@ -335,9 +340,15 @@ namespace detail {
 
 								if (BlobResult.ErrorCode == 0)
 								{
-									if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(BlobResult.Response)))
+									IoHash	 RawHash;
+									uint64_t RawSize;
+									if (CompressedBuffer Chunk =
+											CompressedBuffer::FromCompressed(SharedBuffer(BlobResult.Response), RawHash, RawSize))
 									{
-										Package.AddAttachment(CbAttachment(Chunk, AttachmentHash.AsHash()));
+										if (RawHash == AttachmentHash.AsHash())
+										{
+											Package.AddAttachment(CbAttachment(Chunk, RawHash));
+										}
 									}
 								}
 							});
@@ -398,9 +409,11 @@ namespace detail {
 			{
 				CacheChunkRequest& Request = *RequestPtr;
 				IoBuffer		   Payload;
+				IoHash			   RawHash = IoHash::Zero;
+				uint64_t		   RawSize = 0;
 
-				double			 ElapsedSeconds = 0.0;
-				CompressedBuffer Compressed;
+				double ElapsedSeconds = 0.0;
+				bool   IsCompressed	  = false;
 				if (!Result.Error)
 				{
 					std::string_view	   BlobStoreNamespace = GetActualBlobStoreNamespace(Session, Namespace);
@@ -416,15 +429,15 @@ namespace detail {
 					m_Status.SetFromErrorCode(BlobResult.ErrorCode, BlobResult.Reason);
 					if (Payload && IsCompressedBinary(Payload.GetContentType()))
 					{
-						Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Payload));
+						IsCompressed = CompressedBuffer::ValidateCompressedHeader(Payload, RawHash, RawSize);
 					}
 				}
 
-				if (Compressed)
+				if (IsCompressed)
 				{
 					OnComplete({.Request		= Request,
-								.RawHash		= IoHash::FromBLAKE3(Compressed.GetRawHash()),
-								.RawSize		= Compressed.GetRawSize(),
+								.RawHash		= RawHash,
+								.RawSize		= RawSize,
 								.Value			= Payload,
 								.ElapsedSeconds = ElapsedSeconds,
 								.Source			= &m_Info});
@@ -451,9 +464,11 @@ namespace detail {
 			{
 				CacheValueRequest& Request = *RequestPtr;
 				IoBuffer		   Payload;
+				IoHash			   RawHash = IoHash::Zero;
+				uint64_t		   RawSize = 0;
 
-				double			 ElapsedSeconds = 0.0;
-				CompressedBuffer Compressed;
+				double ElapsedSeconds = 0.0;
+				bool   IsCompressed	  = false;
 				if (!Result.Error)
 				{
 					std::string_view	   BlobStoreNamespace = GetActualBlobStoreNamespace(Session, Namespace);
@@ -470,13 +485,17 @@ namespace detail {
 					{
 						if (IsCompressedBinary(Payload.GetContentType()))
 						{
-							Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Payload));
+							IsCompressed = CompressedBuffer::ValidateCompressedHeader(Payload, RawHash, RawSize) && RawHash != PayloadHash;
 						}
 						else
 						{
-							Compressed	   = CompressedBuffer::Compress(SharedBuffer(Payload));
-							IoHash RawHash = IoHash::FromBLAKE3(Compressed.GetRawHash());
-							if (RawHash != PayloadHash)
+							CompressedBuffer Compressed = CompressedBuffer::Compress(SharedBuffer(Payload));
+							RawHash						= IoHash::FromBLAKE3(Compressed.DecodeRawHash());
+							if (RawHash == PayloadHash)
+							{
+								IsCompressed = true;
+							}
+							else
 							{
 								ZEN_WARN("Horde request for inline payload of {}/{}/{} has hash {}, expected hash {} from header",
 										 Namespace,
@@ -484,17 +503,16 @@ namespace detail {
 										 Request.Key.Hash.ToHexString(),
 										 RawHash.ToHexString(),
 										 PayloadHash.ToHexString());
-								Compressed.Reset();
 							}
 						}
 					}
 				}
 
-				if (Compressed)
+				if (IsCompressed)
 				{
 					OnComplete({.Request		= Request,
-								.RawHash		= IoHash::FromBLAKE3(Compressed.GetRawHash()),
-								.RawSize		= Compressed.GetRawSize(),
+								.RawHash		= RawHash,
+								.RawSize		= RawSize,
 								.Value			= Payload,
 								.ElapsedSeconds = ElapsedSeconds,
 								.Source			= &m_Info});
@@ -543,17 +561,16 @@ namespace detail {
 				}
 				else if (CacheRecord.Type == ZenContentType::kCompressedBinary)
 				{
-					CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(RecordValue));
-					if (!Compressed)
+					IoHash	 RawHash;
+					uint64_t RawSize;
+					if (!CompressedBuffer::ValidateCompressedHeader(RecordValue, RawHash, RawSize))
 					{
 						return {.Reason = std::string("Invalid compressed value buffer"), .Success = false};
 					}
 
-					IoHash RawHash = IoHash::FromBLAKE3(Compressed.GetRawHash());
-
 					CbObjectWriter ReferencingObject;
 					ReferencingObject.AddBinaryAttachment("RawHash", RawHash);
-					ReferencingObject.AddInteger("RawSize", Compressed.GetRawSize());
+					ReferencingObject.AddInteger("RawSize", RawSize);
 
 					return PerformStructuredPut(
 						Session,
@@ -1053,7 +1070,7 @@ namespace detail {
 									{
 										Payload = Compressed.GetCompressed().Flatten().AsIoBuffer();
 										Payload.SetContentType(ZenContentType::kCompressedBinary);
-										RawSize = Compressed.GetRawSize();
+										RawSize = Compressed.DecodeRawSize();
 										Success = true;
 									}
 								}
@@ -1189,7 +1206,7 @@ namespace detail {
 									{
 										Payload = Compressed.GetCompressed().Flatten().AsIoBuffer();
 										Payload.SetContentType(ZenContentType::kCompressedBinary);
-										RawSize = Compressed.GetRawSize();
+										RawSize = Compressed.DecodeRawSize();
 										Success = true;
 									}
 								}
@@ -1252,9 +1269,11 @@ namespace detail {
 
 					for (const IoBuffer& Value : Values)
 					{
-						if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Value)))
+						IoHash	 RawHash;
+						uint64_t RawSize;
+						if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Value), RawHash, RawSize))
 						{
-							Package.AddAttachment(CbAttachment(AttachmentBuffer, IoHash::FromBLAKE3(AttachmentBuffer.GetRawHash())));
+							Package.AddAttachment(CbAttachment(AttachmentBuffer, RawHash));
 						}
 						else
 						{
@@ -1282,7 +1301,9 @@ namespace detail {
 				}
 				else if (CacheRecord.Type == ZenContentType::kCompressedBinary)
 				{
-					CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(RecordValue));
+					IoHash			 RawHash;
+					uint64_t		 RawSize;
+					CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(RecordValue), RawHash, RawSize);
 					if (!Compressed)
 					{
 						return {.Reason = std::string("Invalid value compressed buffer"), .Success = false};
@@ -1312,9 +1333,8 @@ namespace detail {
 								}
 								BatchWriter.EndObject();
 								// Policy unspecified and expected to be Default
-								IoHash Hash = IoHash::FromBLAKE3(Compressed.GetRawHash());
-								BatchWriter.AddBinaryAttachment("RawHash"sv, Hash);
-								BatchPackage.AddAttachment(CbAttachment(Compressed, Hash));
+								BatchWriter.AddBinaryAttachment("RawHash"sv, RawHash);
+								BatchPackage.AddAttachment(CbAttachment(Compressed, RawHash));
 							}
 							BatchWriter.EndObject();
 						}