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]>

2 files changed, 120 insertions, 120 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;
diff --git a/zenserver/cache/structuredcache.h b/zenserver/cache/structuredcache.h
index 703e24ed3..ad7253f79 100644
--- a/zenserver/cache/structuredcache.h
+++ b/zenserver/cache/structuredcache.h
@@ -30,12 +30,12 @@ enum class CachePolicy : uint8_t;
  *
  *   {BucketId}/{KeyHash}
  *
- * Where BucketId is an alphanumeric string, and KeyHash is a 40-character hexadecimal
- * sequence. The hash value may be derived in any number of ways, it's up to the
- * application to pick an approach.
+ * Where BucketId is a lower-case alphanumeric string, and KeyHash is a 40-character
+ * hexadecimal sequence. The hash value may be derived in any number of ways, it's
+ * up to the application to pick an approach.
  *
  * Values may be structured or unstructured. Structured values are encoded using Unreal
- * Engine's compact binary encoding
+ * Engine's compact binary encoding (see CbObject)
  *
  * Additionally, attachments may be addressed as:
  *
@@ -62,8 +62,7 @@ public:
 	~HttpStructuredCacheService();
 
 	virtual const char* BaseUri() const override;
-
-	virtual void HandleRequest(zen::HttpServerRequest& Request) override;
+	virtual void		HandleRequest(zen::HttpServerRequest& Request) override;
 
 	void Flush();
 	void Scrub(ScrubContext& Ctx);
@@ -104,6 +103,7 @@ private:
 	std::unique_ptr<UpstreamCache> m_UpstreamCache;
 	uint64_t					   m_LastScrubTime = 0;
 	metrics::OperationTiming	   m_HttpRequests;
+	metrics::OperationTiming	   m_UpstreamGetRequestTiming;
 	CacheStats					   m_CacheStats;
 };