faster oplog iteration (#471)

* use a CbObjectView instead of CbObject to avoid creating IOBufferCore instances
* use BasicFileBuffer directly where possible
* changelog

7 files changed, 72 insertions, 44 deletions

diff --git a/src/zenserver/projectstore/httpprojectstore.cpp b/src/zenserver/projectstore/httpprojectstore.cpp
index 3edcf5903..261485834 100644
--- a/src/zenserver/projectstore/httpprojectstore.cpp
+++ b/src/zenserver/projectstore/httpprojectstore.cpp
@@ -57,14 +57,13 @@ CSVWriteOp(CidStore&		  CidStore,
 		   bool				  AttachmentDetails,
 		   int				  LSN,
 		   const Oid&		  Key,
-		   CbObject			  Op,
+		   CbObjectView		  Op,
 		   StringBuilderBase& CSVWriter)
 {
 	StringBuilder<32> KeyStringBuilder;
 	Key.ToString(KeyStringBuilder);
 	const std::string_view KeyString = KeyStringBuilder.ToView();
 
-	SharedBuffer Buffer = Op.GetBuffer();
 	if (AttachmentDetails)
 	{
 		Op.IterateAttachments([&CidStore, &CSVWriter, &ProjectId, &OplogId, LSN, &KeyString](CbFieldView FieldView) {
@@ -86,8 +85,8 @@ CSVWriteOp(CidStore&		  CidStore,
 			AttachmentsSize += Attachment.GetSize();
 		});
 		CSVWriter << "\r\n"
-				  << ProjectId << ", " << OplogId << ", " << LSN << ", " << KeyString << ", " << gsl::narrow<uint64_t>(Buffer.GetSize())
-				  << ", " << AttachmentCount << ", " << gsl::narrow<uint64_t>(AttachmentsSize);
+				  << ProjectId << ", " << OplogId << ", " << LSN << ", " << KeyString << ", " << gsl::narrow<uint64_t>(Op.GetSize()) << ", "
+				  << AttachmentCount << ", " << gsl::narrow<uint64_t>(AttachmentsSize);
 	}
 	else
 	{
@@ -105,17 +104,16 @@ namespace {
 				   bool			   AttachmentDetails,
 				   int			   LSN,
 				   const Oid&	   Key,
-				   CbObject		   Op,
+				   CbObjectView	   Op,
 				   CbObjectWriter& CbWriter)
 	{
 		CbWriter.BeginObject();
 		{
-			SharedBuffer Buffer = Op.GetBuffer();
 			CbWriter.AddObjectId("key", Key);
 			if (Details)
 			{
 				CbWriter.AddInteger("lsn", LSN);
-				CbWriter.AddInteger("size", gsl::narrow<uint64_t>(Buffer.GetSize()));
+				CbWriter.AddInteger("size", gsl::narrow<uint64_t>(Op.GetSize()));
 			}
 			if (AttachmentDetails)
 			{
@@ -176,7 +174,7 @@ namespace {
 	{
 		Cbo.BeginArray("ops");
 		{
-			Oplog.IterateOplogWithKey([&Cbo, &CidStore, Details, OpDetails, AttachmentDetails](int LSN, const Oid& Key, CbObject Op) {
+			Oplog.IterateOplogWithKey([&Cbo, &CidStore, Details, OpDetails, AttachmentDetails](int LSN, const Oid& Key, CbObjectView Op) {
 				CbWriteOp(CidStore, Details, OpDetails, AttachmentDetails, LSN, Key, Op, Cbo);
 			});
 		}
@@ -1350,7 +1348,7 @@ HttpProjectService::HandleOpLogEntriesRequest(HttpRouterRequest& Req)
 		{
 			Response.BeginArray("entries"sv);
 
-			FoundLog->IterateOplog([&Response](CbObject Op) { Response << Op; });
+			FoundLog->IterateOplog([&Response](CbObjectView Op) { Response << Op; });
 
 			Response.EndArray();
 		}
@@ -1638,7 +1636,7 @@ HttpProjectService::HandleDetailsRequest(HttpRouterRequest& Req)
 		m_ProjectStore->IterateProjects([&](ProjectStore::Project& Project) {
 			Project.IterateOplogs([&](const RwLock::SharedLockScope&, ProjectStore::Oplog& Oplog) {
 				Oplog.IterateOplogWithKey(
-					[this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObject Op) {
+					[this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObjectView Op) {
 						CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
 					});
 			});
@@ -1693,7 +1691,7 @@ HttpProjectService::HandleProjectDetailsRequest(HttpRouterRequest& Req)
 
 		FoundProject->IterateOplogs([&](const RwLock::SharedLockScope&, ProjectStore::Oplog& Oplog) {
 			Oplog.IterateOplogWithKey(
-				[this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObject Op) {
+				[this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObjectView Op) {
 					CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
 				});
 		});
@@ -1748,9 +1746,10 @@ HttpProjectService::HandleOplogDetailsRequest(HttpRouterRequest& Req)
 		ExtendableStringBuilder<4096> CSVWriter;
 		CSVHeader(Details, AttachmentDetails, CSVWriter);
 
-		Oplog.IterateOplogWithKey([this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObject Op) {
-			CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
-		});
+		Oplog.IterateOplogWithKey(
+			[this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](int LSN, const Oid& Key, CbObjectView Op) {
+				CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
+			});
 		HttpReq.WriteResponse(HttpResponseCode::OK, HttpContentType::kText, CSVWriter.ToView());
 	}
 	else
diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index 02760f6dd..8faad58b6 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -288,7 +288,7 @@ struct ProjectStore::OplogStorage : public RefCounted
 		ZEN_ASSERT(!(m_NextOpsOffset & (m_OpsAlign - 1)));
 	}
 
-	void ReplayLog(std::function<void(CbObject, const OplogEntry&)>&& Handler)
+	void ReplayLog(std::function<void(CbObjectView, const OplogEntry&)>&& Handler)
 	{
 		ZEN_TRACE_CPU("Store::OplogStorage::ReplayLog");
 
@@ -299,7 +299,6 @@ struct ProjectStore::OplogStorage : public RefCounted
 		Stopwatch Timer;
 
 		uint64_t InvalidEntries = 0;
-		size_t	 LargestOp		= 256;
 
 		std::vector<OplogEntry> OpLogEntries;
 		std::vector<size_t>		OplogOrder;
@@ -312,10 +311,6 @@ struct ProjectStore::OplogStorage : public RefCounted
 						++InvalidEntries;
 						return;
 					}
-					if (LogEntry.OpCoreSize > LargestOp)
-					{
-						LargestOp = LogEntry.OpCoreSize;
-					}
 					const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
 					m_NextOpsOffset =
 						Max(m_NextOpsOffset.load(std::memory_order_relaxed), RoundUp(OpFileOffset + LogEntry.OpCoreSize, m_OpsAlign));
@@ -342,14 +337,29 @@ struct ProjectStore::OplogStorage : public RefCounted
 		});
 
 		BasicFileBuffer OpBlobsBuffer(m_OpBlobs, 65536);
-		IoBuffer		OpBuffer(LargestOp);
 
 		for (size_t OplogOrderIndex : OplogOrder)
 		{
 			const OplogEntry& LogEntry = OpLogEntries[OplogOrderIndex];
-			ZEN_ASSERT_SLOW(OpBuffer.GetSize() >= LogEntry.OpCoreSize);
 
 			const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
+			MemoryView	   OpBufferView = OpBlobsBuffer.MakeView(LogEntry.OpCoreSize, OpFileOffset);
+			if (OpBufferView.GetSize() == LogEntry.OpCoreSize)
+			{
+				// Verify checksum, ignore op data if incorrect
+				const auto OpCoreHash = uint32_t(XXH3_64bits(OpBufferView.GetData(), LogEntry.OpCoreSize) & 0xffffFFFF);
+
+				if (OpCoreHash != LogEntry.OpCoreHash)
+				{
+					ZEN_WARN("skipping oplog entry with bad checksum!");
+					InvalidEntries++;
+					continue;
+				}
+				Handler(CbObjectView(OpBufferView.GetData()), LogEntry);
+				continue;
+			}
+
+			IoBuffer OpBuffer(LogEntry.OpCoreSize);
 			OpBlobsBuffer.Read((void*)OpBuffer.Data(), LogEntry.OpCoreSize, OpFileOffset);
 
 			// Verify checksum, ignore op data if incorrect
@@ -361,7 +371,7 @@ struct ProjectStore::OplogStorage : public RefCounted
 				InvalidEntries++;
 				continue;
 			}
-			Handler(CbObject(SharedBuffer::MakeView(OpBuffer.Data(), LogEntry.OpCoreSize)), LogEntry);
+			Handler(CbObjectView(OpBuffer.Data()), LogEntry);
 		}
 
 		if (InvalidEntries)
@@ -375,22 +385,24 @@ struct ProjectStore::OplogStorage : public RefCounted
 				 m_NextOpsOffset.load());
 	}
 
-	void ReplayLogEntries(const std::span<OplogEntryAddress> Entries, std::function<void(CbObject)>&& Handler)
+	void ReplayLogEntries(const std::span<OplogEntryAddress> Entries, std::function<void(CbObjectView)>&& Handler)
 	{
 		ZEN_TRACE_CPU("Store::OplogStorage::ReplayLogEntries");
 
 		BasicFileBuffer OpBlobsBuffer(m_OpBlobs, 65536);
-		IoBuffer		OpBuffer(512);
 
 		for (const OplogEntryAddress& Entry : Entries)
 		{
 			const uint64_t OpFileOffset = Entry.Offset * m_OpsAlign;
-			if (OpBuffer.Size() < Entry.Size)
+			MemoryView	   OpBufferView = OpBlobsBuffer.MakeView(Entry.Size, OpFileOffset);
+			if (OpBufferView.GetSize() == Entry.Size)
 			{
-				OpBuffer = IoBuffer(Entry.Size);
+				Handler(CbObjectView(OpBufferView.GetData()));
+				continue;
 			}
+			IoBuffer OpBuffer(Entry.Size);
 			OpBlobsBuffer.Read((void*)OpBuffer.Data(), Entry.Size, OpFileOffset);
-			Handler(CbObject(SharedBuffer::MakeView(OpBuffer.Data(), Entry.Size)));
+			Handler(CbObjectView(OpBuffer.Data()));
 		}
 	}
 
@@ -509,7 +521,7 @@ ProjectStore::Oplog::GatherReferences(GcContext& GcCtx)
 
 	std::vector<IoHash> Cids;
 	Cids.reserve(1024);
-	IterateOplog([&](CbObject Op) {
+	IterateOplog([&](CbObjectView Op) {
 		Op.IterateAttachments([&](CbFieldView Visitor) { Cids.emplace_back(Visitor.AsAttachment()); });
 		if (Cids.size() >= 1024)
 		{
@@ -649,7 +661,7 @@ ProjectStore::Oplog::ReplayLog()
 	{
 		return;
 	}
-	m_Storage->ReplayLog([&](CbObject Op, const OplogEntry& OpEntry) { RegisterOplogEntry(OplogLock, GetMapping(Op), OpEntry); });
+	m_Storage->ReplayLog([&](CbObjectView Op, const OplogEntry& OpEntry) { RegisterOplogEntry(OplogLock, GetMapping(Op), OpEntry); });
 }
 
 IoBuffer
@@ -754,7 +766,7 @@ ProjectStore::Oplog::IterateFileMap(
 }
 
 void
-ProjectStore::Oplog::IterateOplog(std::function<void(CbObject)>&& Handler)
+ProjectStore::Oplog::IterateOplog(std::function<void(CbObjectView)>&& Handler)
 {
 	RwLock::SharedLockScope _(m_OplogLock);
 	if (!m_Storage)
@@ -777,11 +789,11 @@ ProjectStore::Oplog::IterateOplog(std::function<void(CbObject)>&& Handler)
 		return Lhs.Offset < Rhs.Offset;
 	});
 
-	m_Storage->ReplayLogEntries(Entries, [&](CbObject Op) { Handler(Op); });
+	m_Storage->ReplayLogEntries(Entries, [&](CbObjectView Op) { Handler(Op); });
 }
 
 void
-ProjectStore::Oplog::IterateOplogWithKey(std::function<void(int, const Oid&, CbObject)>&& Handler)
+ProjectStore::Oplog::IterateOplogWithKey(std::function<void(int, const Oid&, CbObjectView)>&& Handler)
 {
 	RwLock::SharedLockScope _(m_OplogLock);
 	if (!m_Storage)
@@ -822,7 +834,7 @@ ProjectStore::Oplog::IterateOplogWithKey(std::function<void(int, const Oid&, CbO
 	}
 
 	size_t EntryIndex = 0;
-	m_Storage->ReplayLogEntries(SortedEntries, [&](CbObject Op) {
+	m_Storage->ReplayLogEntries(SortedEntries, [&](CbObjectView Op) {
 		Handler(LSNs[EntryIndex], Keys[EntryIndex], Op);
 		EntryIndex++;
 	});
@@ -926,7 +938,7 @@ ProjectStore::Oplog::AddMetaMapping(const RwLock::ExclusiveLockScope&, Oid Chunk
 }
 
 ProjectStore::Oplog::OplogEntryMapping
-ProjectStore::Oplog::GetMapping(CbObject Core)
+ProjectStore::Oplog::GetMapping(CbObjectView Core)
 {
 	using namespace std::literals;
 
@@ -2748,7 +2760,7 @@ ProjectStore::Rpc(HttpServerRequest&	 HttpReq,
 		std::vector<CbObject>						 NewOps;
 		std::unordered_map<Oid, IoHash, Oid::Hasher> NewChunkMappings;
 
-		Oplog->IterateOplog([&](CbObject Op) {
+		Oplog->IterateOplog([&](CbObjectView Op) {
 			bool OpRewritten = false;
 			bool AllOk		 = true;
 
diff --git a/src/zenserver/projectstore/projectstore.h b/src/zenserver/projectstore/projectstore.h
index 0d55e37e9..0ed9ef532 100644
--- a/src/zenserver/projectstore/projectstore.h
+++ b/src/zenserver/projectstore/projectstore.h
@@ -93,8 +93,8 @@ public:
 
 		std::vector<ChunkInfo> GetAllChunksInfo();
 		void IterateFileMap(std::function<void(const Oid&, const std::string_view& ServerPath, const std::string_view& ClientPath)>&& Fn);
-		void IterateOplog(std::function<void(CbObject)>&& Fn);
-		void IterateOplogWithKey(std::function<void(int, const Oid&, CbObject)>&& Fn);
+		void IterateOplog(std::function<void(CbObjectView)>&& Fn);
+		void IterateOplogWithKey(std::function<void(int, const Oid&, CbObjectView)>&& Fn);
 		std::optional<CbObject> GetOpByKey(const Oid& Key);
 		std::optional<CbObject> GetOpByIndex(int Index);
 		int						GetOpIndexByKey(const Oid& Key);
@@ -191,7 +191,7 @@ public:
 			std::vector<FileMapping> Files;
 		};
 
-		OplogEntryMapping GetMapping(CbObject Core);
+		OplogEntryMapping GetMapping(CbObjectView Core);
 
 		/** Update tracking metadata for a new oplog entry
 		 *
diff --git a/src/zenserver/projectstore/remoteprojectstore.cpp b/src/zenserver/projectstore/remoteprojectstore.cpp
index ee3456ecc..02557dfa5 100644
--- a/src/zenserver/projectstore/remoteprojectstore.cpp
+++ b/src/zenserver/projectstore/remoteprojectstore.cpp
@@ -257,7 +257,7 @@ BuildContainer(CidStore&																	ChunkStore,
 		std::vector<SharedBuffer>						ChunksInBlock;
 		std::unordered_map<IoHash, int, IoHash::Hasher> Attachments;
 
-		auto RewriteOp = [&](int LSN, CbObject Op, const std::function<void(CbObject)>& CB) {
+		auto RewriteOp = [&](int LSN, CbObjectView Op, const std::function<void(CbObjectView)>& CB) {
 			bool		OpRewritten = false;
 			CbArrayView Files		= Op["files"sv].AsArrayView();
 			if (Files.Num() == 0)
@@ -401,7 +401,7 @@ BuildContainer(CidStore&																	ChunkStore,
 
 		tsl::robin_map<int, std::string> OpLSNToKey;
 
-		Oplog.IterateOplogWithKey([&](int LSN, const Oid&, CbObject Op) {
+		Oplog.IterateOplogWithKey([&](int LSN, const Oid&, CbObjectView Op) {
 			if (RemoteResult.IsError())
 			{
 				return;
@@ -411,7 +411,7 @@ BuildContainer(CidStore&																	ChunkStore,
 			Op.IterateAttachments([&](CbFieldView FieldView) { Attachments.insert({FieldView.AsAttachment(), LSN}); });
 			if (OutLooseAttachments != nullptr)
 			{
-				RewriteOp(LSN, Op, [&SectionOpsWriter](CbObject Op) { SectionOpsWriter << Op; });
+				RewriteOp(LSN, Op, [&SectionOpsWriter](CbObjectView Op) { SectionOpsWriter << Op; });
 			}
 			else
 			{
diff --git a/src/zenserver/vfs/vfsimpl.cpp b/src/zenserver/vfs/vfsimpl.cpp
index f74000900..8acbfba87 100644
--- a/src/zenserver/vfs/vfsimpl.cpp
+++ b/src/zenserver/vfs/vfsimpl.cpp
@@ -393,7 +393,7 @@ VfsServiceDataSource::PopulateDirectory(std::string NodePath, VfsTreeNode& DirNo
 					}
 				};
 
-				Oplog->IterateOplog([&](CbObject Op) {
+				Oplog->IterateOplog([&](CbObjectView Op) {
 					EmitFilesForDataArray(Op["packagedata"sv].AsArrayView());
 					EmitFilesForDataArray(Op["bulkdata"sv].AsArrayView());
 				});
diff --git a/src/zenutil/basicfile.cpp b/src/zenutil/basicfile.cpp
index 99aa6bf39..259d622c4 100644
--- a/src/zenutil/basicfile.cpp
+++ b/src/zenutil/basicfile.cpp
@@ -614,6 +614,22 @@ BasicFileBuffer::Read(void* Data, uint64_t Size, uint64_t FileOffset)
 	ZEN_ASSERT(Begin + Count == End);
 }
 
+MemoryView
+BasicFileBuffer::MakeView(uint64_t Size, uint64_t FileOffset)
+{
+	if (FileOffset < m_BufferStart || (FileOffset + Size) > m_BufferEnd)
+	{
+		if (m_Buffer == nullptr || (Size > m_BufferSize) || (FileOffset + Size > m_Size))
+		{
+			return {};
+		}
+		m_BufferStart = FileOffset;
+		m_BufferEnd	  = Min(m_BufferStart + m_BufferSize, m_Size);
+		m_Base.Read(m_Buffer, m_BufferEnd - m_BufferStart, m_BufferStart);
+	}
+	return MemoryView(m_Buffer + (FileOffset - m_BufferStart), Size);
+}
+
 /*
 		___________              __
 		\__    ___/___   _______/  |_  ______
diff --git a/src/zenutil/include/zenutil/basicfile.h b/src/zenutil/include/zenutil/basicfile.h
index 7cf8b05db..eb97c885c 100644
--- a/src/zenutil/include/zenutil/basicfile.h
+++ b/src/zenutil/include/zenutil/basicfile.h
@@ -133,7 +133,8 @@ public:
 	BasicFileBuffer(BasicFile& Base, uint64_t BufferSize);
 	~BasicFileBuffer();
 
-	void Read(void* Data, uint64_t Size, uint64_t FileOffset);
+	void	   Read(void* Data, uint64_t Size, uint64_t FileOffset);
+	MemoryView MakeView(uint64_t Size, uint64_t FileOffset);
 
 private:
 	BasicFile&	   m_Base;