remote project store stats (#44)

* add remote oplog store statistics
* block chunking when uploading oplog to zenserver (mirroring)
* make sure we can move temporary dechunked file into cas store

1 files changed, 56 insertions, 19 deletions

diff --git a/src/zenserver/projectstore/remoteprojectstore.cpp b/src/zenserver/projectstore/remoteprojectstore.cpp
index 8efb92e6b..ba3ff6499 100644
--- a/src/zenserver/projectstore/remoteprojectstore.cpp
+++ b/src/zenserver/projectstore/remoteprojectstore.cpp
@@ -113,6 +113,34 @@ IsCancelled(JobContext* OptionalContext)
 	return OptionalContext->IsCancelled();
 }
 
+void
+ReportRemoteStoreStats(JobContext*								  OptionalContext,
+					   const RemoteProjectStore::RemoteStoreInfo& RemoteStoreInfo,
+					   const RemoteProjectStore::Stats&			  Stats)
+{
+	ReportMessage(
+		OptionalContext,
+		fmt::format(
+			"Remote store stats for '{}'\n"
+			"Requests: {} (avg {})\n"
+			"Sent: {} (avg {}, {}/s)\n"
+			"Received: {} (avg {}, {}/s)\n"
+			"Peak: Send {}, Receive {} ({}/s)",
+			RemoteStoreInfo.Description,
+			Stats.m_RequestCount,
+			NiceLatencyNs(Stats.m_RequestCount > 0 ? (Stats.m_RequestTimeNS / Stats.m_RequestCount) : 0u),
+			NiceBytes(Stats.m_SentBytes),
+			NiceBytes(Stats.m_RequestCount > 0u ? static_cast<uint64_t>((Stats.m_SentBytes) / Stats.m_RequestCount) : 0u),
+			NiceBytes(Stats.m_RequestTimeNS > 0u ? static_cast<uint64_t>((Stats.m_SentBytes * 1000000000) / Stats.m_RequestTimeNS) : 0u),
+			NiceBytes(Stats.m_ReceivedBytes),
+			NiceBytes(Stats.m_RequestCount > 0u ? static_cast<uint64_t>((Stats.m_ReceivedBytes) / Stats.m_RequestCount) : 0u),
+			NiceBytes(Stats.m_RequestTimeNS > 0u ? static_cast<uint64_t>((Stats.m_ReceivedBytes * 1000000000) / Stats.m_RequestTimeNS)
+												 : 0u),
+			NiceBytes(Stats.m_PeakSentBytes),
+			NiceBytes(Stats.m_PeakReceivedBytes),
+			NiceBytes(Stats.m_PeakBytesPerSec)));
+}
+
 bool
 IterateBlock(IoBuffer&& CompressedBlock, std::function<void(CompressedBuffer&& Chunk, const IoHash& AttachmentHash)> Visitor)
 {
@@ -292,6 +320,7 @@ BuildContainer(CidStore&																	ChunkStore,
 			   size_t																		ChunkFileSizeLimit,
 			   bool																			BuildBlocks,
 			   bool																			IgnoreMissingAttachments,
+			   bool																			AllowChunking,
 			   const std::vector<Block>&													KnownBlocks,
 			   WorkerThreadPool&															WorkerPool,
 			   const std::function<void(CompressedBuffer&&, const IoHash&)>&				AsyncOnBlock,
@@ -619,7 +648,8 @@ BuildContainer(CidStore&																	ChunkStore,
 			auto __ = MakeGuard([&SourceFile]() { SourceFile.Detach(); });
 
 			Chunked.Chunked = ChunkData(SourceFile, Offset, Size, UShaderByteCodeParams);
-			Chunked.Source	= RawData;
+			ZEN_ASSERT(Chunked.Chunked.Info.RawHash == RawHash);
+			Chunked.Source = RawData;
 
 			ZEN_INFO("Chunked large attachment '{}' {} into {} chunks in {}",
 					 RawHash,
@@ -667,6 +697,7 @@ BuildContainer(CidStore&																	ChunkStore,
 									 &ChunkedFiles,
 									 MaxChunkEmbedSize,
 									 ChunkFileSizeLimit,
+									 AllowChunking,
 									 &RemoteResult,
 									 OptionalContext]() {
 				auto _ = MakeGuard([&ResolveAttachmentsLatch] { ResolveAttachmentsLatch.CountDown(); });
@@ -804,7 +835,7 @@ BuildContainer(CidStore&																	ChunkStore,
 							};
 
 							IoBufferFileReference FileRef;
-							if (GetForChunking(ChunkFileSizeLimit, Data, FileRef))
+							if (AllowChunking && GetForChunking(ChunkFileSizeLimit, Data, FileRef))
 							{
 								ChunkedFile Chunked = ChunkFile(RawHash, Data, FileRef, OptionalContext);
 								ResolveLock.WithExclusiveLock(
@@ -1357,6 +1388,7 @@ BuildContainer(CidStore&																	ChunkStore,
 			   size_t																		ChunkFileSizeLimit,
 			   bool																			BuildBlocks,
 			   bool																			IgnoreMissingAttachments,
+			   bool																			AllowChunking,
 			   const std::function<void(CompressedBuffer&&, const IoHash&)>&				AsyncOnBlock,
 			   const std::function<void(const IoHash&, TGetAttachmentBufferFunc&&)>&		OnLargeAttachment,
 			   const std::function<void(std::vector<std::pair<IoHash, FetchChunkFunc>>&&)>& OnBlockChunks,
@@ -1373,6 +1405,7 @@ BuildContainer(CidStore&																	ChunkStore,
 											   ChunkFileSizeLimit,
 											   BuildBlocks,
 											   IgnoreMissingAttachments,
+											   AllowChunking,
 											   {},
 											   WorkerPool,
 											   AsyncOnBlock,
@@ -1880,6 +1913,8 @@ SaveOplog(CidStore&				 ChunkStore,
 												   ChunkFileSizeLimit,
 												   RemoteStoreInfo.CreateBlocks,
 												   IgnoreMissingAttachments,
+												   RemoteStoreInfo.AllowChunking,
+
 												   KnownBlocks,
 												   WorkerPool,
 												   OnBlock,
@@ -2039,6 +2074,8 @@ SaveOplog(CidStore&				 ChunkStore,
 							  Info.AttachmentsUploaded.load(),
 							  NiceBytes(Info.AttachmentBytesUploaded.load())));
 
+	ReportRemoteStoreStats(OptionalContext, RemoteStoreInfo, RemoteStore.GetStats());
+
 	return Result;
 };
 
@@ -2548,7 +2585,18 @@ LoadOplog(CidStore&			   ChunkStore,
 										 IgnoreMissingAttachments,
 										 &Info,
 										 OptionalContext]() {
-					auto _ = MakeGuard([&DechunkLatch] { DechunkLatch.CountDown(); });
+					auto _ = MakeGuard([&DechunkLatch, &TempFileName] {
+						std::error_code Ec;
+						if (std::filesystem::exists(TempFileName, Ec))
+						{
+							std::filesystem::remove(TempFileName, Ec);
+							if (Ec)
+							{
+								ZEN_INFO("Failed to remove temporary file '{}'. Reason: {}", TempFileName, Ec.message());
+							}
+						}
+						DechunkLatch.CountDown();
+					});
 					if (RemoteResult.IsError())
 					{
 						return;
@@ -2574,14 +2622,6 @@ LoadOplog(CidStore&			   ChunkStore,
 
 									// We only add 1 as the resulting missing count will be 1 for the dechunked file
 									Info.MissingAttachmentCount.fetch_add(1);
-									TmpFile.Close();
-									std::error_code Ec;
-									std::filesystem::remove(TempFileName, Ec);
-									if (Ec)
-									{
-										ZEN_INFO("Failed to remove temporary file '{}'. Reason: {}", TempFileName, Ec.message());
-									}
-
 									if (!IgnoreMissingAttachments)
 									{
 										RemoteResult.SetError(
@@ -2606,14 +2646,8 @@ LoadOplog(CidStore&			   ChunkStore,
 							UniqueBuffer Header = CompressedBuffer::CreateHeaderForNoneEncoder(Chunked.RawSize, RawHash);
 							TmpWriter.Write(Header.GetData(), Header.GetSize(), 0);
 						}
-						TmpFile.Flush();
-						uint64_t TmpFileSize = TmpFile.FileSize();
-						TmpBuffer			 = IoBuffer(IoBuffer::File, TmpFile.Detach(), 0, TmpFileSize, /*IsWholeFile*/ true);
-						IoHash	 ValidateRawHash;
-						uint64_t ValidateRawSize = 0;
-						ZEN_ASSERT(CompressedBuffer::ValidateCompressedHeader(TmpBuffer, ValidateRawHash, ValidateRawSize));
-						ZEN_ASSERT(ValidateRawHash == Chunked.RawHash);
-						ZEN_ASSERT(ValidateRawSize == Chunked.RawSize);
+						TmpFile.Close();
+						TmpBuffer = IoBufferBuilder::MakeFromTemporaryFile(TempFileName);
 					}
 					CidStore::InsertResult InsertResult =
 						ChunkStore.AddChunk(TmpBuffer, Chunked.RawHash, CidStore::InsertMode::kMayBeMovedInPlace);
@@ -2668,6 +2702,9 @@ LoadOplog(CidStore&			   ChunkStore,
 							  Info.AttachmentsStored.load(),
 							  NiceBytes(Info.AttachmentBytesStored.load()),
 							  Info.MissingAttachmentCount.load()));
+
+	ReportRemoteStoreStats(OptionalContext, RemoteStoreInfo, RemoteStore.GetStats());
+
 	return Result;
 }