use partial blocks for oplog import (#780)

Feature: Add --allow-partial-block-requests to zen oplog-import
Improvement: zen oplog-import now uses partial block requests to reduce download size
Improvement: Use latency to Cloud Storage host and Zen Cache host when calculating partial block requests

1 files changed, 60 insertions, 19 deletions

diff --git a/src/zenremotestore/chunking/chunkblock.cpp b/src/zenremotestore/chunking/chunkblock.cpp
index 06cedae3f..d203e0292 100644
--- a/src/zenremotestore/chunking/chunkblock.cpp
+++ b/src/zenremotestore/chunking/chunkblock.cpp
@@ -597,7 +597,7 @@ ChunkBlockAnalyser::CalculatePartialBlockDownloads(std::span<const NeededBlock>
 
 				if (MaybeBlockRanges.has_value())
 				{
-					const std::vector<BlockRangeDescriptor>& BlockRanges = MaybeBlockRanges.value();
+					std::vector<BlockRangeDescriptor> BlockRanges = MaybeBlockRanges.value();
 					ZEN_ASSERT(!BlockRanges.empty());
 
 					uint64_t RequestedSize =
@@ -606,12 +606,54 @@ ChunkBlockAnalyser::CalculatePartialBlockDownloads(std::span<const NeededBlock>
 										uint64_t(0),
 										[](uint64_t Current, const BlockRangeDescriptor& Range) { return Current + Range.RangeLength; });
 
-					if ((PartialBlockDownloadMode != EPartialBlockDownloadMode::Exact) && ((RequestedSize * 100) / TotalBlockSize) >= 200)
+					if (PartialBlockDownloadMode != EPartialBlockDownloadMode::Exact && BlockRanges.size() > 1)
+					{
+						// TODO: Once we have support in our http client to request multiple ranges in one request this
+						// logic would need to change as the per-request overhead would go away
+
+						const double LatencySec = PartialBlockDownloadMode == EPartialBlockDownloadMode::MultiRangeHighSpeed
+													  ? m_Options.HostHighSpeedLatencySec
+													  : m_Options.HostLatencySec;
+						if (LatencySec > 0)
+						{
+							const uint64_t BytesPerSec = PartialBlockDownloadMode == EPartialBlockDownloadMode::MultiRangeHighSpeed
+															 ? m_Options.HostHighSpeedBytesPerSec
+															 : m_Options.HostSpeedBytesPerSec;
+
+							const double   ExtraRequestTimeSec	 = (BlockRanges.size() - 1) * LatencySec;
+							const uint64_t ExtraRequestTimeBytes = uint64_t(ExtraRequestTimeSec * BytesPerSec);
+
+							const uint64_t FullRangeSize =
+								BlockRanges.back().RangeStart + BlockRanges.back().RangeLength - BlockRanges.front().RangeStart;
+
+							if (ExtraRequestTimeBytes + RequestedSize >= FullRangeSize)
+							{
+								BlockRanges = std::vector<BlockRangeDescriptor>{MergeBlockRanges(BlockRanges)};
+
+								if (m_Options.IsVerbose)
+								{
+									ZEN_OPERATION_LOG_INFO(m_LogOutput,
+														   "Merging {} chunks ({}) from block {} ({}) to single request (extra bytes {})",
+														   NeededBlock.ChunkIndexes.size(),
+														   NiceBytes(RequestedSize),
+														   BlockDescription.BlockHash,
+														   NiceBytes(TotalBlockSize),
+														   NiceBytes(BlockRanges.front().RangeLength - RequestedSize));
+								}
+
+								RequestedSize = BlockRanges.front().RangeLength;
+							}
+						}
+					}
+
+					if ((PartialBlockDownloadMode != EPartialBlockDownloadMode::Exact) &&
+						((TotalBlockSize - RequestedSize) < (512u * 1024u)))
 					{
 						if (m_Options.IsVerbose)
 						{
 							ZEN_OPERATION_LOG_INFO(m_LogOutput,
-												   "Requesting {} chunks ({}) from block {} ({}) using full block request (extra bytes {})",
+												   "Requesting {} chunks ({}) from block {} ({}) using full block request due to small "
+												   "total slack (extra bytes {})",
 												   NeededBlock.ChunkIndexes.size(),
 												   NiceBytes(RequestedSize),
 												   BlockDescription.BlockHash,
@@ -624,19 +666,16 @@ ChunkBlockAnalyser::CalculatePartialBlockDownloads(std::span<const NeededBlock>
 					{
 						Result.BlockRanges.insert(Result.BlockRanges.end(), BlockRanges.begin(), BlockRanges.end());
 
-						if (RequestedSize > TotalWantedChunksSize)
+						if (m_Options.IsVerbose)
 						{
-							if (m_Options.IsVerbose)
-							{
-								ZEN_OPERATION_LOG_INFO(m_LogOutput,
-													   "Requesting {} chunks ({}) from block {} ({}) using {} requests (extra bytes {})",
-													   NeededBlock.ChunkIndexes.size(),
-													   NiceBytes(RequestedSize),
-													   BlockDescription.BlockHash,
-													   NiceBytes(TotalBlockSize),
-													   BlockRanges.size(),
-													   NiceBytes(RequestedSize - TotalWantedChunksSize));
-							}
+							ZEN_OPERATION_LOG_INFO(m_LogOutput,
+												   "Requesting {} chunks ({}) from block {} ({}) using {} requests (extra bytes {})",
+												   NeededBlock.ChunkIndexes.size(),
+												   NiceBytes(RequestedSize),
+												   BlockDescription.BlockHash,
+												   NiceBytes(TotalBlockSize),
+												   BlockRanges.size(),
+												   NiceBytes(RequestedSize - TotalWantedChunksSize));
 						}
 					}
 				}
@@ -786,7 +825,7 @@ ChunkBlockAnalyser::CollapseBlockRanges(const uint64_t AlwaysAcceptableGap, std:
 };
 
 uint64_t
-ChunkBlockAnalyser::CalculateNextGap(std::span<const BlockRangeDescriptor> BlockRanges)
+ChunkBlockAnalyser::CalculateNextGap(const uint64_t AlwaysAcceptableGap, std::span<const BlockRangeDescriptor> BlockRanges)
 {
 	ZEN_ASSERT(BlockRanges.size() > 1);
 	uint64_t AcceptableGap = (uint64_t)-1;
@@ -798,7 +837,7 @@ ChunkBlockAnalyser::CalculateNextGap(std::span<const BlockRangeDescriptor> Block
 		const uint64_t Gap = NextRange.RangeStart - (Range.RangeStart + Range.RangeLength);
 		AcceptableGap	   = Min(Gap, AcceptableGap);
 	}
-	AcceptableGap = RoundUp(AcceptableGap, 16u * 1024u);
+	AcceptableGap = RoundUp(AcceptableGap, AlwaysAcceptableGap);
 	return AcceptableGap;
 };
 
@@ -949,10 +988,12 @@ ChunkBlockAnalyser::CalculateBlockRanges(uint32_t					  BlockIndex,
 		return MakeOptionalBlockRangeVector(TotalBlockSize, MergedRange);
 	}
 
-	std::vector<BlockRangeDescriptor> CollapsedBlockRanges = CollapseBlockRanges(16u * 1024u, BlockRanges);
+	const uint64_t AlwaysAcceptableGap = 4u * 1024u;
+
+	std::vector<BlockRangeDescriptor> CollapsedBlockRanges = CollapseBlockRanges(AlwaysAcceptableGap, BlockRanges);
 	while (GetBlockRangeLimitForRange(ForceMergeLimits, TotalBlockSize, CollapsedBlockRanges))
 	{
-		CollapsedBlockRanges = CollapseBlockRanges(CalculateNextGap(CollapsedBlockRanges), CollapsedBlockRanges);
+		CollapsedBlockRanges = CollapseBlockRanges(CalculateNextGap(AlwaysAcceptableGap, CollapsedBlockRanges), CollapsedBlockRanges);
 	}
 
 	const std::uint64_t WantedCollapsedSize =