stream decompress (#293)

* clean up latency parameters and slow down rate updates
* add DecompressToStream

1 files changed, 186 insertions, 0 deletions

diff --git a/src/zencore/compress.cpp b/src/zencore/compress.cpp
index 0e2ce2b54..f13f8b9ca 100644
--- a/src/zencore/compress.cpp
+++ b/src/zencore/compress.cpp
@@ -185,6 +185,12 @@ public:
 												  const MemoryView	   HeaderView,
 												  uint64_t			   RawOffset,
 												  uint64_t			   RawSize) const = 0;
+
+	virtual bool DecompressToStream(const BufferHeader&													 Header,
+									const CompositeBuffer&												 CompressedData,
+									uint64_t															 RawOffset,
+									uint64_t															 RawSize,
+									std::function<void(uint64_t Offset, const CompositeBuffer& Range)>&& Callback) const = 0;
 };
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -263,6 +269,22 @@ public:
 	}
 
 	[[nodiscard]] uint64_t GetHeaderSize(const BufferHeader&) const final { return sizeof(BufferHeader); }
+
+	virtual bool DecompressToStream(const BufferHeader&													 Header,
+									const CompositeBuffer&												 CompressedData,
+									uint64_t															 RawOffset,
+									uint64_t															 RawSize,
+									std::function<void(uint64_t Offset, const CompositeBuffer& Range)>&& Callback) const final
+	{
+		if (Header.Method == CompressionMethod::None && Header.TotalCompressedSize == CompressedData.GetSize() &&
+			Header.TotalCompressedSize == Header.TotalRawSize + sizeof(BufferHeader) && RawOffset < Header.TotalRawSize &&
+			(RawOffset + RawSize) <= Header.TotalRawSize)
+		{
+			Callback(0, CompressedData.Mid(sizeof(BufferHeader) + RawOffset, RawSize));
+			return true;
+		}
+		return false;
+	}
 };
 
 //////////////////////////////////////////////////////////////////////////
@@ -447,6 +469,12 @@ public:
 									   MutableMemoryView	  RawView,
 									   uint64_t				  RawOffset) const final;
 
+	virtual bool DecompressToStream(const BufferHeader&													 Header,
+									const CompositeBuffer&												 CompressedData,
+									uint64_t															 RawOffset,
+									uint64_t															 RawSize,
+									std::function<void(uint64_t Offset, const CompositeBuffer& Range)>&& Callback) const final;
+
 protected:
 	virtual bool DecompressBlock(MutableMemoryView RawData, MemoryView CompressedData) const = 0;
 };
@@ -569,6 +597,143 @@ BlockDecoder::DecompressToComposite(const BufferHeader& Header, const CompositeB
 }
 
 bool
+BlockDecoder::DecompressToStream(const BufferHeader&												  Header,
+								 const CompositeBuffer&												  CompressedData,
+								 uint64_t															  RawOffset,
+								 uint64_t															  RawSize,
+								 std::function<void(uint64_t Offset, const CompositeBuffer& Range)>&& Callback) const
+{
+	if (Header.TotalCompressedSize != CompressedData.GetSize())
+	{
+		return false;
+	}
+
+	const uint64_t BlockSize = uint64_t(1) << Header.BlockSizeExponent;
+
+	UniqueBuffer BlockSizeBuffer;
+	MemoryView BlockSizeView = CompressedData.ViewOrCopyRange(sizeof(BufferHeader), Header.BlockCount * sizeof(uint32_t), BlockSizeBuffer);
+	std::span<uint32_t const> CompressedBlockSizes(reinterpret_cast<const uint32_t*>(BlockSizeView.GetData()), Header.BlockCount);
+
+	UniqueBuffer CompressedBlockCopy;
+
+	const size_t   FirstBlockIndex	  = uint64_t(RawOffset / BlockSize);
+	const size_t   LastBlockIndex	  = uint64_t((RawOffset + RawSize - 1) / BlockSize);
+	const uint64_t LastBlockSize	  = BlockSize - ((Header.BlockCount * BlockSize) - Header.TotalRawSize);
+	uint64_t	   OffsetInFirstBlock = RawOffset % BlockSize;
+	uint64_t	   CompressedOffset	  = sizeof(BufferHeader) + uint64_t(Header.BlockCount) * sizeof(uint32_t);
+	uint64_t	   RemainingRawSize	  = RawSize;
+
+	for (size_t BlockIndex = 0; BlockIndex < FirstBlockIndex; BlockIndex++)
+	{
+		const uint32_t CompressedBlockSize = ByteSwap(CompressedBlockSizes[BlockIndex]);
+		CompressedOffset += CompressedBlockSize;
+	}
+
+	UniqueBuffer RawDataBuffer;
+
+	IoBufferFileReference FileRef = {nullptr, 0, 0};
+	if ((CompressedData.GetSegments().size() == 1) && CompressedData.GetSegments()[0].AsIoBuffer().GetFileReference(FileRef))
+	{
+		ZEN_ASSERT(FileRef.FileHandle != nullptr);
+		BasicFile Source;
+		Source.Attach(FileRef.FileHandle);
+
+		for (size_t BlockIndex = FirstBlockIndex; BlockIndex <= LastBlockIndex; BlockIndex++)
+		{
+			const uint64_t UncompressedBlockSize = BlockIndex == Header.BlockCount - 1 ? LastBlockSize : BlockSize;
+			const uint32_t CompressedBlockSize	 = ByteSwap(CompressedBlockSizes[BlockIndex]);
+			const bool	   IsCompressed			 = CompressedBlockSize < UncompressedBlockSize;
+
+			const uint64_t BytesToUncompress = OffsetInFirstBlock > 0 ? zen::Min(RawSize, UncompressedBlockSize - OffsetInFirstBlock)
+																	  : zen::Min(RemainingRawSize, BlockSize);
+
+			if (CompressedBlockCopy.GetSize() < CompressedBlockSize)
+			{
+				CompressedBlockCopy = UniqueBuffer::Alloc(CompressedBlockSize);
+			}
+			Source.Read(CompressedBlockCopy.GetData(), CompressedBlockSize, FileRef.FileChunkOffset + CompressedOffset);
+
+			MemoryView CompressedBlock = CompressedBlockCopy.GetView().Left(CompressedBlockSize);
+
+			if (IsCompressed)
+			{
+				if (RawDataBuffer.IsNull())
+				{
+					RawDataBuffer = UniqueBuffer::Alloc(zen::Min(RawSize, UncompressedBlockSize));
+				}
+				else
+				{
+					ZEN_ASSERT(RawDataBuffer.GetSize() >= UncompressedBlockSize);
+				}
+				MutableMemoryView UncompressedBlock = RawDataBuffer.GetMutableView().Left(UncompressedBlockSize);
+				if (!DecompressBlock(UncompressedBlock, CompressedBlock))
+				{
+					Source.Detach();
+					return false;
+				}
+				Callback(BlockIndex * BlockSize + OffsetInFirstBlock,
+						 CompositeBuffer(IoBuffer(IoBuffer::Wrap, RawDataBuffer.GetData(), BytesToUncompress)));
+			}
+			else
+			{
+				Callback(BlockIndex * BlockSize + OffsetInFirstBlock,
+						 CompositeBuffer(
+							 IoBuffer(IoBuffer::Wrap, CompressedBlockCopy.GetView().Mid(OffsetInFirstBlock).GetData(), BytesToUncompress)));
+			}
+
+			OffsetInFirstBlock = 0;
+			RemainingRawSize -= BytesToUncompress;
+			CompressedOffset += CompressedBlockSize;
+		}
+		Source.Detach();
+	}
+	else
+	{
+		for (size_t BlockIndex = FirstBlockIndex; BlockIndex <= LastBlockIndex; BlockIndex++)
+		{
+			const uint64_t UncompressedBlockSize = BlockIndex == Header.BlockCount - 1 ? LastBlockSize : BlockSize;
+			const uint32_t CompressedBlockSize	 = ByteSwap(CompressedBlockSizes[BlockIndex]);
+			const bool	   IsCompressed			 = CompressedBlockSize < UncompressedBlockSize;
+
+			const uint64_t BytesToUncompress = OffsetInFirstBlock > 0 ? zen::Min(RawSize, UncompressedBlockSize - OffsetInFirstBlock)
+																	  : zen::Min(RemainingRawSize, BlockSize);
+
+			MemoryView CompressedBlock = CompressedData.ViewOrCopyRange(CompressedOffset, CompressedBlockSize, CompressedBlockCopy);
+
+			if (IsCompressed)
+			{
+				if (RawDataBuffer.IsNull())
+				{
+					RawDataBuffer = UniqueBuffer::Alloc(zen::Min(RawSize, UncompressedBlockSize));
+				}
+				else
+				{
+					ZEN_ASSERT(RawDataBuffer.GetSize() >= UncompressedBlockSize);
+				}
+				MutableMemoryView UncompressedBlock = RawDataBuffer.GetMutableView().Left(UncompressedBlockSize);
+				if (!DecompressBlock(UncompressedBlock, CompressedBlock))
+				{
+					return false;
+				}
+				Callback(BlockIndex * BlockSize + OffsetInFirstBlock,
+						 CompositeBuffer(IoBuffer(IoBuffer::Wrap, RawDataBuffer.GetData(), BytesToUncompress)));
+			}
+			else
+			{
+				Callback(BlockIndex * BlockSize + OffsetInFirstBlock,
+						 CompositeBuffer(
+							 IoBuffer(IoBuffer::Wrap, CompressedBlockCopy.GetView().Mid(OffsetInFirstBlock).GetData(), BytesToUncompress)));
+			}
+
+			OffsetInFirstBlock = 0;
+			RemainingRawSize -= BytesToUncompress;
+			CompressedOffset += CompressedBlockSize;
+		}
+	}
+	return true;
+}
+
+bool
 BlockDecoder::TryDecompressTo(const BufferHeader&	 Header,
 							  const CompositeBuffer& CompressedData,
 							  MutableMemoryView		 RawView,
@@ -1644,6 +1809,27 @@ CompressedBuffer::DecompressToComposite() const
 }
 
 bool
+CompressedBuffer::DecompressToStream(uint64_t															  RawOffset,
+									 uint64_t															  RawSize,
+									 std::function<void(uint64_t Offset, const CompositeBuffer& Range)>&& Callback) const
+{
+	using namespace detail;
+	if (CompressedData)
+	{
+		const BufferHeader Header = BufferHeader::Read(CompressedData);
+		if (Header.Magic == BufferHeader::ExpectedMagic)
+		{
+			if (const BaseDecoder* const Decoder = GetDecoder(Header.Method))
+			{
+				const uint64_t TotalRawSize = RawSize < ~uint64_t(0) ? RawSize : Header.TotalRawSize - RawOffset;
+				return Decoder->DecompressToStream(Header, CompressedData, RawOffset, TotalRawSize, std::move(Callback));
+			}
+		}
+	}
+	return false;
+}
+
+bool
 CompressedBuffer::TryGetCompressParameters(OodleCompressor&		  OutCompressor,
 										   OodleCompressionLevel& OutCompressionLevel,
 										   uint64_t&			  OutBlockSize) const