Use multithreading to fetch size/rawsize of entries in `/prj/{project}/oplog/{log}/chunkinfos` and `/prj/{project}/oplog/{log}/files` (#30)

- Improvement: Use multithreading to fetch size/rawsize of entries in `/prj/{project}/oplog/{log}/chunkinfos` and `/prj/{project}/oplog/{log}/files`
- Improvement: Add `GetMediumWorkerPool()` in addition to `LargeWorkerPool()` and `SmallWorkerPool()`

1 files changed, 161 insertions, 58 deletions

diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index f01134fa5..dd390d08c 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -20,6 +20,7 @@
 #include <zenstore/scrubcontext.h>
 #include <zenutil/cache/rpcrecording.h>
 #include <zenutil/packageformat.h>
+#include <zenutil/workerpools.h>
 
 #include "fileremoteprojectstore.h"
 #include "jupiterremoteprojectstore.h"
@@ -844,6 +845,15 @@ ProjectStore::Oplog::ReplayLog()
 }
 
 IoBuffer
+ProjectStore::Oplog::GetChunkByRawHash(const IoHash& RawHash)
+{
+	IoBuffer Chunk = m_CidStore.FindChunkByCid(RawHash);
+	Chunk.SetContentType(ZenContentType::kCompressedBinary);
+
+	return Chunk;
+}
+
+IoBuffer
 ProjectStore::Oplog::FindChunk(const Oid& ChunkId)
 {
 	RwLock::SharedLockScope OplogLock(m_OplogLock);
@@ -2532,49 +2542,101 @@ ProjectStore::GetProjectFiles(const std::string_view				 ProjectId,
 	const bool WantsRawSizeField	= WantsAllFields || WantedFieldNames.contains("rawsize");
 	const bool WantsSizeField		= WantsAllFields || WantedFieldNames.contains("size");
 
-	CbObjectWriter Response;
-	Response.BeginArray("files"sv);
+	std::vector<Oid>		 Ids;
+	std::vector<std::string> ServerPaths;
+	std::vector<std::string> ClientPaths;
+	std::vector<uint64_t>	 Sizes;
+	std::vector<uint64_t>	 RawSizes;
 
+	size_t Count = 0;
 	FoundLog->IterateFileMap([&](const Oid& Id, const std::string_view& ServerPath, const std::string_view& ClientPath) {
+		if (WantsIdField || WantsRawSizeField || WantsSizeField)
+		{
+			Ids.push_back(Id);
+		}
+		if (WantsServerPathField)
+		{
+			ServerPaths.push_back(std::string(ServerPath));
+		}
+		if (WantsClientPathField)
+		{
+			ClientPaths.push_back(std::string(ClientPath));
+		}
+		Count++;
+	});
+	if (WantsRawSizeField || WantsSizeField)
+	{
+		if (WantsSizeField)
+		{
+			Sizes.resize(Ids.size(), 0u);
+		}
+		if (WantsRawSizeField)
+		{
+			RawSizes.resize(Ids.size(), 0u);
+		}
+
+		WorkerThreadPool& WorkerPool = GetSmallWorkerPool();  // GetSyncWorkerPool();
+		Latch			  WorkLatch(1);
+
+		for (size_t Index = 0; Index < Ids.size(); Index++)
+		{
+			WorkLatch.AddCount(1);
+			WorkerPool.ScheduleWork(
+				[&WorkLatch, FoundLog, WantsSizeField, WantsRawSizeField, &Sizes, &RawSizes, Index, ChunkId = Ids[Index]]() {
+					auto _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
+					if (IoBuffer Chunk = FoundLog->FindChunk(ChunkId))
+					{
+						uint64_t Size = Chunk.GetSize();
+						if (WantsRawSizeField)
+						{
+							uint64_t RawSize = Size;
+							if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
+							{
+								IoHash __;
+								(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), __, RawSize);
+							}
+							RawSizes[Index] = RawSize;
+						}
+						if (WantsSizeField)
+						{
+							Sizes[Index] = Size;
+						}
+					}
+				});
+		}
+		WorkLatch.CountDown();
+		WorkLatch.Wait();
+	}
+
+	CbObjectWriter Response;
+	Response.BeginArray("files"sv);
+	for (size_t Index = 0; Index < Count; Index++)
+	{
 		Response.BeginObject();
 		if (WantsIdField)
 		{
-			Response << "id"sv << Id;
+			Response << "id"sv << Ids[Index];
+		}
+		if (WantsServerPathField)
+		{
+			Response << "serverpath"sv << ServerPaths[Index];
 		}
 		if (WantsClientPathField)
 		{
-			Response << "clientpath"sv << ClientPath;
+			Response << "clientpath"sv << ClientPaths[Index];
 		}
-		if (WantsServerPathField && !ServerPath.empty())
+		if (WantsSizeField)
 		{
-			Response << "serverpath"sv << ServerPath;
+			Response << "size"sv << Sizes[Index];
 		}
-		if (WantsRawSizeField || WantsSizeField)
+		if (WantsRawSizeField)
 		{
-			IoBuffer Chunk = FoundLog->FindChunk(Id);
-			if (WantsSizeField)
-			{
-				Response << "size"sv << Chunk.GetSize();
-			}
-			if (WantsRawSizeField)
-			{
-				if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
-				{
-					IoHash	 _;
-					uint64_t RawSize = 0;
-					(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), _, RawSize);
-					Response << "rawsize"sv << RawSize;
-				}
-				else
-				{
-					Response << "rawsize"sv << Chunk.GetSize();
-				}
-			}
+			Response << "rawsize"sv << RawSizes[Index];
 		}
 		Response.EndObject();
-	});
-
+	}
 	Response.EndArray();
+
 	OutPayload = Response.Save();
 	return {HttpResponseCode::OK, {}};
 }
@@ -2603,9 +2665,6 @@ ProjectStore::GetProjectChunkInfos(const std::string_view				  ProjectId,
 	}
 	Project->TouchOplog(OplogId);
 
-	std::vector<std::pair<Oid, IoHash>> ChunkInfos;
-	FoundLog->IterateChunkMap([&ChunkInfos](const Oid& Id, const IoHash& Hash) { ChunkInfos.push_back({Id, Hash}); });
-
 	const bool WantsAllFields = WantedFieldNames.empty();
 
 	const bool WantsIdField		 = WantsAllFields || WantedFieldNames.contains("id");
@@ -2613,45 +2672,89 @@ ProjectStore::GetProjectChunkInfos(const std::string_view				  ProjectId,
 	const bool WantsRawSizeField = WantsAllFields || WantedFieldNames.contains("rawsize");
 	const bool WantsSizeField	 = WantsAllFields || WantedFieldNames.contains("size");
 
+	std::vector<Oid>	  Ids;
+	std::vector<IoHash>	  Hashes;
+	std::vector<uint64_t> RawSizes;
+	std::vector<uint64_t> Sizes;
+
+	size_t Count = 0;
+	FoundLog->IterateChunkMap([&](const Oid& Id, const IoHash& Hash) {
+		if (WantsIdField)
+		{
+			Ids.push_back(Id);
+		}
+		if (WantsRawHashField || WantsRawSizeField || WantsSizeField)
+		{
+			Hashes.push_back(Hash);
+		}
+		Count++;
+	});
+
+	if (WantsRawSizeField || WantsSizeField)
+	{
+		if (WantsRawSizeField)
+		{
+			RawSizes.resize(Hashes.size(), 0u);
+		}
+		if (WantsSizeField)
+		{
+			Sizes.resize(Hashes.size(), 0u);
+		}
+
+		WorkerThreadPool& WorkerPool = GetSmallWorkerPool();  // GetSyncWorkerPool();
+		Latch			  WorkLatch(1);
+
+		for (size_t Index = 0; Index < Hashes.size(); Index++)
+		{
+			WorkLatch.AddCount(1);
+			WorkerPool.ScheduleWork(
+				[&WorkLatch, FoundLog, WantsSizeField, WantsRawSizeField, &Sizes, &RawSizes, Index, RawHash = Hashes[Index]]() {
+					auto _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
+					if (IoBuffer Chunk = FoundLog->GetChunkByRawHash(RawHash))
+					{
+						uint64_t Size = Chunk.GetSize();
+						if (WantsRawSizeField)
+						{
+							uint64_t RawSize = Size;
+							if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
+							{
+								IoHash __;
+								(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), __, RawSize);
+							}
+							RawSizes[Index] = RawSize;
+						}
+						if (WantsSizeField)
+						{
+							Sizes[Index] = Size;
+						}
+					}
+				});
+		}
+		WorkLatch.CountDown();
+		WorkLatch.Wait();
+	}
+
 	CbObjectWriter Response;
 	Response.BeginArray("chunkinfos"sv);
 
-	for (const auto& ChunkInfo : ChunkInfos)
+	for (size_t Index = 0; Index < Count; Index++)
 	{
 		Response.BeginObject();
 		if (WantsIdField)
 		{
-			Response << "id"sv << ChunkInfo.first;
+			Response << "id"sv << Ids[Index];
 		}
 		if (WantsRawHashField)
 		{
-			Response << "rawhash"sv << ChunkInfo.second;
+			Response << "rawhash"sv << Hashes[Index];
 		}
-
-		if (WantsRawSizeField || WantsSizeField)
+		if (WantsSizeField)
 		{
-			if (IoBuffer Chunk = FoundLog->FindChunk(ChunkInfo.first))
-			{
-				uint64_t Size = Chunk.GetSize();
-				if (WantsSizeField)
-				{
-					Response << "size"sv << Size;
-				}
-				if (WantsRawSizeField)
-				{
-					if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
-					{
-						IoHash	 _;
-						uint64_t RawSize = 0;
-						(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), _, RawSize);
-						Response << "rawsize"sv << RawSize;
-					}
-					else
-					{
-						Response << "rawsize"sv << Size;
-					}
-				}
-			}
+			Response << "size"sv << Sizes[Index];
+		}
+		if (WantsRawSizeField)
+		{
+			Response << "rawsize"sv << RawSizes[Index];
 		}
 		Response.EndObject();
 	}