Support for asynchronous HTTP response processing (#19)

This change introduces WriteResponseAsync which can be used to move potentially slow request handler code (like upstream lookups) off the I/O service thread to ensure we are always able to serve as many HTTP requests as possible. The current implementation defaults to 16 async worker threads and there is currently no back-pressure.

- Added RequestStats - Metrics for network requests. Aggregates tracking of duration, payload sizes into a single class for ease of use
- Added some metrics on upstream communication

Co-authored-by: Per Larsson <[email protected]>

1 files changed, 114 insertions, 114 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 27bb1c5cd..4a2a3748a 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -286,28 +286,104 @@ HttpStructuredCacheService::HandleCacheRecordRequest(HttpServerRequest& Request,
 void
 HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
 {
-	const ZenContentType AcceptType = Request.AcceptContentType();
+	const ZenContentType AcceptType		 = Request.AcceptContentType();
+	const bool			 SkipData		 = (Policy & CachePolicy::SkipData) == CachePolicy::SkipData;
+	const bool			 SkipAttachments = (Policy & CachePolicy::SkipAttachments) == CachePolicy::SkipAttachments;
+	const bool			 QueryUpstream	 = m_UpstreamCache && ((Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote);
 
-	ZenCacheValue Value;
-	bool		  Success		  = m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, /* out */ Value);
-	bool		  InUpstreamCache = false;
+	bool		  Success = false;
+	ZenCacheValue LocalCacheValue;
 
-	const bool QueryUpstream = !Success && m_UpstreamCache && (CachePolicy::QueryRemote == (Policy & CachePolicy::QueryRemote));
+	if (m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, LocalCacheValue))
+	{
+		Success = true;
 
-	if (QueryUpstream)
+		if (!SkipData && AcceptType == ZenContentType::kCbPackage)
+		{
+			CbPackage	 Package;
+			CbObjectView CacheRecord(LocalCacheValue.Value.Data());
+			uint32_t	 AttachmentCount = 0;
+			uint32_t	 ValidCount		 = 0;
+
+			if (!SkipAttachments)
+			{
+				CacheRecord.IterateAttachments([this, &Ref, &Package, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
+					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+					{
+						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
+						ValidCount++;
+					}
+					AttachmentCount++;
+				});
+
+				if (ValidCount != AttachmentCount)
+				{
+					Success = false;
+					ZEN_WARN("GET - '{}/{}' '{}' FAILED, found '{}' of '{}' attachments",
+							 Ref.BucketSegment,
+							 Ref.HashKey,
+							 ToString(AcceptType),
+							 ValidCount,
+							 AttachmentCount);
+				}
+			}
+
+			Package.SetObject(LoadCompactBinaryObject(LocalCacheValue.Value));
+
+			BinaryWriter MemStream;
+			Package.Save(MemStream);
+
+			LocalCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+			LocalCacheValue.Value.SetContentType(HttpContentType::kCbPackage);
+		}
+	}
+
+	if (Success)
 	{
-		const ZenContentType CacheRecordType = AcceptType;
+		ZEN_DEBUG("HIT - '{}/{}' {} '{}' (LOCAL)",
+				  Ref.BucketSegment,
+				  Ref.HashKey,
+				  NiceBytes(LocalCacheValue.Value.Size()),
+				  ToString(LocalCacheValue.Value.GetContentType()));
 
-		if (auto UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, CacheRecordType);
+		m_CacheStats.HitCount++;
+
+		if (SkipData)
+		{
+			return Request.WriteResponse(HttpResponseCode::OK);
+		}
+		else
+		{
+			return Request.WriteResponse(HttpResponseCode::OK, LocalCacheValue.Value.GetContentType(), LocalCacheValue.Value);
+		}
+	}
+	else if (!QueryUpstream)
+	{
+		ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
+		m_CacheStats.MissCount++;
+		return Request.WriteResponse(HttpResponseCode::NotFound);
+	}
+
+	// Issue upstream query asynchronously in order to keep requests flowing without
+	// hogging I/O servicing threads with blocking work
+
+	Request.WriteResponseAsync([this, AcceptType, SkipData, SkipAttachments, Ref](HttpServerRequest& AsyncRequest) {
+		bool		  Success = false;
+		ZenCacheValue UpstreamCacheValue;
+
+		metrics::OperationTiming::Scope $(m_UpstreamGetRequestTiming);
+
+		if (GetUpstreamCacheResult UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, AcceptType);
 			UpstreamResult.Success)
 		{
-			Value.Value		= UpstreamResult.Value;
-			Success			= true;
-			InUpstreamCache = true;
+			Success					 = true;
+			UpstreamCacheValue.Value = UpstreamResult.Value;
+
+			UpstreamCacheValue.Value.SetContentType(AcceptType);
 
-			if (CacheRecordType == ZenContentType::kBinary || CacheRecordType == ZenContentType::kCbObject)
+			if (AcceptType == ZenContentType::kBinary || AcceptType == ZenContentType::kCbObject)
 			{
-				if (CacheRecordType == ZenContentType::kCbObject)
+				if (AcceptType == ZenContentType::kCbObject)
 				{
 					const CbValidateError ValidationResult = ValidateCompactBinary(UpstreamResult.Value, CbValidateMode::All);
 
@@ -320,7 +396,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 						CacheRecord.IterateAttachments([&](CbFieldView Attachment) { IndexData.AddHash(Attachment.AsHash()); });
 						IndexData.EndArray();
 
-						Value.IndexData = IndexData.Save();
+						UpstreamCacheValue.IndexData = IndexData.Save();
 					}
 					else
 					{
@@ -334,19 +410,17 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 
 				if (Success)
 				{
-					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, Value);
+					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, UpstreamCacheValue);
 				}
 			}
-			else
+			else if (AcceptType == ZenContentType::kCbPackage)
 			{
-				ZEN_ASSERT(CacheRecordType == ZenContentType::kCbPackage);
-
 				CbPackage Package;
-				if (Package.TryLoad(UpstreamResult.Value))
+				if (Package.TryLoad(UpstreamCacheValue.Value))
 				{
+					CbObject CacheRecord	 = Package.GetObject();
 					uint32_t AttachmentCount = 0;
 					uint32_t ValidCount		 = 0;
-					CbObject CacheRecord	 = Package.GetObject();
 
 					CacheRecord.IterateAttachments([this, &Package, &Ref, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
 						if (const CbAttachment* Attachment = Package.FindAttachment(AttachmentHash.AsHash()))
@@ -371,7 +445,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 					{
 						m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = CacheRecord.GetBuffer().AsIoBuffer()});
 
-						if (zen::CachePolicy::SkipAttachments == (Policy & zen::CachePolicy::SkipAttachments))
+						if (SkipAttachments)
 						{
 							CbPackage PackageWithoutAttachments;
 							PackageWithoutAttachments.SetObject(CacheRecord);
@@ -379,7 +453,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 							BinaryWriter MemStream;
 							PackageWithoutAttachments.Save(MemStream);
 
-							Value.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+							UpstreamCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
 						}
 					}
 					else
@@ -398,109 +472,34 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 				}
 			}
 		}
-	}
-
-	if (!Success)
-	{
-		ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
-		m_CacheStats.MissCount++;
-		return Request.WriteResponse(HttpResponseCode::NotFound);
-	}
-
-	if (AcceptType == ZenContentType::kCbPackage && !InUpstreamCache)
-	{
-		CbObjectView CacheRecord(Value.Value.Data());
 
-		const CbValidateError ValidationResult = ValidateCompactBinary(Value.Value, CbValidateMode::All);
-
-		if (ValidationResult != CbValidateError::None)
+		if (Success)
 		{
-			ZEN_WARN("GET - '{}/{}' '{}' FAILED, invalid compact binary object", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
-			m_CacheStats.MissCount++;
-			return Request.WriteResponse(HttpResponseCode::NotFound, HttpContentType::kText, "Invalid cache record"sv);
-		}
+			ZEN_DEBUG("HIT - '{}/{}' {} '{}' (UPSTREAM)",
+					  Ref.BucketSegment,
+					  Ref.HashKey,
+					  NiceBytes(UpstreamCacheValue.Value.Size()),
+					  ToString(UpstreamCacheValue.Value.GetContentType()));
 
-		if ((Policy & CachePolicy::SkipData) == CachePolicy::SkipData)
-		{
 			m_CacheStats.HitCount++;
-			return Request.WriteResponse(HttpResponseCode::OK);
-		}
-
-		const bool SkipAttachments = (Policy & CachePolicy::SkipAttachments) == CachePolicy::SkipAttachments;
-		uint32_t   AttachmentCount = 0;
-		uint32_t   ValidCount	   = 0;
-		uint64_t   AttachmentBytes = 0ull;
-
-		CbPackage Package;
-
-		if (!SkipAttachments)
-		{
-			CacheRecord.IterateAttachments(
-				[this, &Ref, &Package, &AttachmentCount, &ValidCount, &AttachmentBytes](CbFieldView AttachmentHash) {
-					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
-					{
-						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
-						AttachmentBytes += Chunk.Size();
-						ValidCount++;
-					}
-					AttachmentCount++;
-				});
+			m_CacheStats.UpstreamHitCount++;
 
-			if (ValidCount != AttachmentCount)
+			if (SkipData)
 			{
-				ZEN_WARN("GET - '{}/{}' '{}' FAILED, found '{}' of '{}' attachments",
-						 Ref.BucketSegment,
-						 Ref.HashKey,
-						 ToString(AcceptType),
-						 ValidCount,
-						 AttachmentCount);
-
-				m_CacheStats.MissCount++;
-				return Request.WriteResponse(HttpResponseCode::NotFound, HttpContentType::kText, "Missing attachments"sv);
+				AsyncRequest.WriteResponse(HttpResponseCode::OK);
+			}
+			else
+			{
+				AsyncRequest.WriteResponse(HttpResponseCode::OK, UpstreamCacheValue.Value.GetContentType(), UpstreamCacheValue.Value);
 			}
-		}
-
-		Package.SetObject(LoadCompactBinaryObject(Value.Value));
-
-		ZEN_DEBUG("HIT - '{}/{}' {} '{}', {} attachments (LOCAL)",
-				  Ref.BucketSegment,
-				  Ref.HashKey,
-				  NiceBytes(AttachmentBytes + Value.Value.Size()),
-				  ToString(HttpContentType::kCbPackage),
-				  AttachmentCount);
-
-		BinaryWriter MemStream;
-		Package.Save(MemStream);
-
-		IoBuffer Response(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
-
-		m_CacheStats.HitCount++;
-		Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, Response);
-	}
-	else
-	{
-		ZEN_DEBUG("HIT - '{}/{}' {} '{}' ({})",
-				  Ref.BucketSegment,
-				  Ref.HashKey,
-				  NiceBytes(Value.Value.Size()),
-				  ToString(Value.Value.GetContentType()),
-				  InUpstreamCache ? "UPSTREAM" : "LOCAL");
-
-		m_CacheStats.HitCount++;
-		if (InUpstreamCache)
-		{
-			m_CacheStats.UpstreamHitCount++;
-		}
-
-		if ((Policy & CachePolicy::SkipData) == CachePolicy::SkipData)
-		{
-			Request.WriteResponse(HttpResponseCode::OK);
 		}
 		else
 		{
-			Request.WriteResponse(HttpResponseCode::OK, Value.Value.GetContentType(), Value.Value);
+			ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
+			m_CacheStats.MissCount++;
+			AsyncRequest.WriteResponse(HttpResponseCode::NotFound);
 		}
-	}
+	});
 }
 
 void
@@ -875,6 +874,7 @@ HttpStructuredCacheService::HandleStatsRequest(zen::HttpServerRequest& Request)
 	CbObjectWriter Cbo;
 
 	EmitSnapshot("requests", m_HttpRequests, Cbo);
+	EmitSnapshot("upstream_gets", m_UpstreamGetRequestTiming, Cbo);
 
 	const uint64_t HitCount			= m_CacheStats.HitCount;
 	const uint64_t UpstreamHitCount = m_CacheStats.UpstreamHitCount;