aboutsummaryrefslogtreecommitdiff
path: root/src/zenutil
diff options
context:
space:
mode:
authorDan Engelbrecht <[email protected]>2025-02-26 15:10:14 +0100
committerGitHub Enterprise <[email protected]>2025-02-26 15:10:14 +0100
commit7d8fe45af3b49d800f84f0ddce051c0b3b2e837d (patch)
treec8dd564dcf247d7b2537bb5c2ebfbca57bafd205 /src/zenutil
parentimprovements and infrastructure for upcoming builds api command line (#284) (diff)
downloadzen-7d8fe45af3b49d800f84f0ddce051c0b3b2e837d.tar.xz
zen-7d8fe45af3b49d800f84f0ddce051c0b3b2e837d.zip
builds upload command (#278)
- Feature: **EXPERIMENTAL** New `zen builds` command to list, upload and download folders to Cloud Build API - `builds list` list available builds (**INCOMPLETE - FILTERING MISSING**) - `builds upload` upload a folder to Cloud Build API - `--local-path` source folder to upload - `--create-build` creates a new parent build object (using the object id), if omitted a parent build must exist and `--build-id` must be given - `--build-id` an Oid in hex form for the Build identifier to use - omit to have the id auto generated - `--build-part-id` and Oid in hex form for the Build Part identifier for the folder - omit to have the id auto generated - `--build-part-name` name of the build part - if omitted the name of the leaf folder name give in `--local-path` - `--metadata-path` path to a json formatted file with meta data information about the build. Meta-data must be provided if `--create-build` is set - `--metadata` key-value pairs separated by ';' with build meta data for the build. (key1=value1;key2=value2). Meta-data must be provided if `--create-build` is set - `--clean` ignore any existing blocks of chunk data and upload a fresh set of blocks - `--allow-multipart` enable usage of multi-part http upload requests - `--manifest-path` path to text file listing files to include in upload. Exclude to upload everything in `--local-path` - `builds download` download a folder from Cloud Build API (**INCOMPLETE - WILL WIPE UNTRACKED DATA FROM TARGET FOLDER**) - `--local-path` target folder to download to - `--build-id` an Oid in hex form for the Build identifier to use - `--build-part-id` a comma separated list of Oid in hex for the build part identifier(s) to download - mutually exclusive to `--build-part-name` - `--build-part-name` a comma separated list of names for the build part(s) to download - if omitted the name of the leaf folder name give in `--local-path` - `--clean` deletes all data in target folder before downloading (NON-CLEAN IS NOT IMPLEMENTED YET) - `--allow-multipart` enable usage of multi-part http download reqeusts - `builds diff` download a folder from Cloud Build API - `--local-path` target folder to download to - `--compare-path` folder to compare target with - `--only-chunked` compare only files that would be chunked - `builds fetch-blob` fetch and validate a blob from remote store - `--build-id` an Oid in hex form for the Build identifier to use - `--blob-hash` an IoHash in hex form identifying the blob to download - `builds validate part` fetch a build part and validate all referenced attachments - `--build-id` an Oid in hex form for the Build identifier to use - `--build-part-id` an Oid in hex for the build part identifier to validate - mutually exclusive to `--build-part-name` - `--build-part-name` a name for the build part to validate - mutually exclusive to `--build-part-id` - `builds test` a series of operation that uploads, downloads and test various aspects of incremental operations - `--local-path` source folder to upload - Options for Cloud Build API remote store (`list`, `upload`, `download`, `fetch-blob`, `validate-part`) - `--url` Cloud Builds URL - `--assume-http2` assume that the builds endpoint is a HTTP/2 endpoint skipping HTTP/1.1 upgrade handshake - `--namespace` Builds Storage namespace - `--bucket` Builds Storage bucket - Authentication options for Cloud Build API - Auth token - `--access-token` http auth Cloud Storage access token - `--access-token-env` name of environment variable that holds the Http auth Cloud Storage access token - `--access-token-path` path to json file that holds the Http auth Cloud Storage access token - OpenId authentication - `--openid-provider-name` Open ID provider name - `--openid-provider-url` Open ID provider url - `--openid-client-id`Open ID client id - `--openid-refresh-token` Open ID refresh token - `--encryption-aes-key` 256 bit AES encryption key for storing OpenID credentials - `--encryption-aes-iv` 128 bit AES encryption initialization vector for storing OpenID credentials - OAuth authentication - `--oauth-url` OAuth provier url - `--oauth-clientid` OAuth client id - `--oauth-clientsecret` OAuth client secret - Options for file based remote store used for for testing purposes (`list`, `upload`, `download`, `fetch-blob`, `validate-part`, `test`) - `--storage-path` path to folder to store builds data - `--json-metadata` enable json output in store for all compact binary objects (off by default) - Output options for all builds commands - `--plain-progress` use plain line-by-line progress output - `--verbose`
Diffstat (limited to 'src/zenutil')
-rw-r--r--src/zenutil/chunkblock.cpp94
-rw-r--r--src/zenutil/chunkedcontent.cpp865
-rw-r--r--src/zenutil/chunkingcontroller.cpp265
-rw-r--r--src/zenutil/filebuildstorage.cpp616
-rw-r--r--src/zenutil/include/zenutil/buildstorage.h55
-rw-r--r--src/zenutil/include/zenutil/chunkblock.h17
-rw-r--r--src/zenutil/include/zenutil/chunkedcontent.h256
-rw-r--r--src/zenutil/include/zenutil/chunkingcontroller.h55
-rw-r--r--src/zenutil/include/zenutil/filebuildstorage.h16
-rw-r--r--src/zenutil/include/zenutil/jupiter/jupiterbuildstorage.h17
-rw-r--r--src/zenutil/include/zenutil/parallellwork.h69
-rw-r--r--src/zenutil/jupiter/jupiterbuildstorage.cpp371
12 files changed, 2678 insertions, 18 deletions
diff --git a/src/zenutil/chunkblock.cpp b/src/zenutil/chunkblock.cpp
index 6dae5af11..a19cf5c1b 100644
--- a/src/zenutil/chunkblock.cpp
+++ b/src/zenutil/chunkblock.cpp
@@ -3,6 +3,7 @@
#include <zenutil/chunkblock.h>
#include <zencore/compactbinarybuilder.h>
+#include <zencore/fmtutils.h>
#include <zencore/logging.h>
#include <vector>
@@ -18,20 +19,27 @@ ParseChunkBlockDescription(const CbObjectView& BlockObject)
Result.BlockHash = BlockObject["rawHash"sv].AsHash();
if (Result.BlockHash != IoHash::Zero)
{
+ Result.HeaderSize = BlockObject["headerSize"sv].AsUInt64();
CbArrayView ChunksArray = BlockObject["rawHashes"sv].AsArrayView();
- Result.ChunkHashes.reserve(ChunksArray.Num());
+ Result.ChunkRawHashes.reserve(ChunksArray.Num());
for (CbFieldView ChunkView : ChunksArray)
{
- Result.ChunkHashes.push_back(ChunkView.AsHash());
+ Result.ChunkRawHashes.push_back(ChunkView.AsHash());
}
- CbArrayView ChunkRawLengthsArray = BlockObject["chunkRawLengths"sv].AsArrayView();
- std::vector<uint32_t> ChunkLengths;
+ CbArrayView ChunkRawLengthsArray = BlockObject["chunkRawLengths"sv].AsArrayView();
Result.ChunkRawLengths.reserve(ChunkRawLengthsArray.Num());
for (CbFieldView ChunkView : ChunkRawLengthsArray)
{
Result.ChunkRawLengths.push_back(ChunkView.AsUInt32());
}
+
+ CbArrayView ChunkCompressedLengthsArray = BlockObject["chunkCompressedLengths"sv].AsArrayView();
+ Result.ChunkCompressedLengths.reserve(ChunkCompressedLengthsArray.Num());
+ for (CbFieldView ChunkView : ChunkCompressedLengthsArray)
+ {
+ Result.ChunkCompressedLengths.push_back(ChunkView.AsUInt32());
+ }
}
return Result;
}
@@ -57,18 +65,23 @@ ParseChunkBlockDescriptionList(const CbObjectView& BlocksObject)
CbObject
BuildChunkBlockDescription(const ChunkBlockDescription& Block, CbObjectView MetaData)
{
- ZEN_ASSERT(Block.ChunkRawLengths.size() == Block.ChunkHashes.size());
+ ZEN_ASSERT(Block.BlockHash != IoHash::Zero);
+ ZEN_ASSERT(Block.HeaderSize > 0);
+ ZEN_ASSERT(Block.ChunkRawLengths.size() == Block.ChunkRawHashes.size());
+ ZEN_ASSERT(Block.ChunkCompressedLengths.size() == Block.ChunkRawHashes.size());
CbObjectWriter Writer;
Writer.AddHash("rawHash"sv, Block.BlockHash);
+ Writer.AddInteger("headerSize"sv, Block.HeaderSize);
Writer.BeginArray("rawHashes"sv);
{
- for (const IoHash& ChunkHash : Block.ChunkHashes)
+ for (const IoHash& ChunkHash : Block.ChunkRawHashes)
{
Writer.AddHash(ChunkHash);
}
}
Writer.EndArray();
+
Writer.BeginArray("chunkRawLengths");
{
for (uint32_t ChunkSize : Block.ChunkRawLengths)
@@ -78,11 +91,58 @@ BuildChunkBlockDescription(const ChunkBlockDescription& Block, CbObjectView Meta
}
Writer.EndArray();
+ Writer.BeginArray("chunkCompressedLengths");
+ {
+ for (uint32_t ChunkSize : Block.ChunkCompressedLengths)
+ {
+ Writer.AddInteger(ChunkSize);
+ }
+ }
+ Writer.EndArray();
+
Writer.AddObject("metadata", MetaData);
return Writer.Save();
}
+ChunkBlockDescription
+GetChunkBlockDescription(const SharedBuffer& BlockPayload, const IoHash& RawHash)
+{
+ ChunkBlockDescription BlockDescription = {{.BlockHash = IoHash::HashBuffer(BlockPayload)}};
+ if (BlockDescription.BlockHash != RawHash)
+ {
+ throw std::runtime_error(fmt::format("Block {} content hash {} does not match block hash", RawHash, BlockDescription.BlockHash));
+ }
+ if (IterateChunkBlock(
+ BlockPayload,
+ [&BlockDescription, RawHash](CompressedBuffer&& Chunk, const IoHash& AttachmentHash) {
+ if (CompositeBuffer Decompressed = Chunk.DecompressToComposite(); Decompressed)
+ {
+ IoHash ChunkHash = IoHash::HashBuffer(Decompressed.Flatten());
+ if (ChunkHash != AttachmentHash)
+ {
+ throw std::runtime_error(
+ fmt::format("Chunk {} in block {} content hash {} does not match chunk", AttachmentHash, RawHash, ChunkHash));
+ }
+ BlockDescription.ChunkRawHashes.push_back(AttachmentHash);
+ BlockDescription.ChunkRawLengths.push_back(gsl::narrow<uint32_t>(Decompressed.GetSize()));
+ BlockDescription.ChunkCompressedLengths.push_back(gsl::narrow<uint32_t>(Chunk.GetCompressedSize()));
+ }
+ else
+ {
+ throw std::runtime_error(fmt::format("Chunk {} in block {} is not a compressed buffer", AttachmentHash, RawHash));
+ }
+ },
+ BlockDescription.HeaderSize))
+ {
+ return BlockDescription;
+ }
+ else
+ {
+ throw std::runtime_error(fmt::format("Block {} is malformed", RawHash));
+ }
+}
+
CompressedBuffer
GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks, ChunkBlockDescription& OutBlock)
{
@@ -91,8 +151,9 @@ GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks,
std::vector<SharedBuffer> ChunkSegments;
ChunkSegments.resize(1);
ChunkSegments.reserve(1 + ChunkCount);
- OutBlock.ChunkHashes.reserve(ChunkCount);
+ OutBlock.ChunkRawHashes.reserve(ChunkCount);
OutBlock.ChunkRawLengths.reserve(ChunkCount);
+ OutBlock.ChunkCompressedLengths.reserve(ChunkCount);
{
IoBuffer TempBuffer(ChunkCount * 9);
MutableMemoryView View = TempBuffer.GetMutableView();
@@ -106,16 +167,19 @@ GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks,
std::span<const SharedBuffer> Segments = Chunk.second.GetCompressed().GetSegments();
for (const SharedBuffer& Segment : Segments)
{
+ ZEN_ASSERT(Segment.IsOwned());
ChunkSize += Segment.GetSize();
ChunkSegments.push_back(Segment);
}
BufferEndPtr += WriteVarUInt(ChunkSize, BufferEndPtr);
- OutBlock.ChunkHashes.push_back(It.first);
+ OutBlock.ChunkRawHashes.push_back(It.first);
OutBlock.ChunkRawLengths.push_back(gsl::narrow<uint32_t>(Chunk.first));
+ OutBlock.ChunkCompressedLengths.push_back(gsl::narrow<uint32_t>(ChunkSize));
}
ZEN_ASSERT(BufferEndPtr <= View.GetDataEnd());
ptrdiff_t TempBufferLength = std::distance(BufferStartPtr, BufferEndPtr);
ChunkSegments[0] = SharedBuffer(IoBuffer(TempBuffer, 0, gsl::narrow<size_t>(TempBufferLength)));
+ OutBlock.HeaderSize = TempBufferLength;
}
CompressedBuffer CompressedBlock =
CompressedBuffer::Compress(CompositeBuffer(std::move(ChunkSegments)), OodleCompressor::Mermaid, OodleCompressionLevel::None);
@@ -124,7 +188,9 @@ GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks,
}
bool
-IterateChunkBlock(const SharedBuffer& BlockPayload, std::function<void(CompressedBuffer&& Chunk, const IoHash& AttachmentHash)> Visitor)
+IterateChunkBlock(const SharedBuffer& BlockPayload,
+ std::function<void(CompressedBuffer&& Chunk, const IoHash& AttachmentHash)> Visitor,
+ uint64_t& OutHeaderSize)
{
ZEN_ASSERT(BlockPayload);
if (BlockPayload.GetSize() < 1)
@@ -144,21 +210,23 @@ IterateChunkBlock(const SharedBuffer& BlockPayload, std::function<void(Compresse
ChunkSizes.push_back(ReadVarUInt(ReadPtr, NumberSize));
ReadPtr += NumberSize;
}
+ uint64_t Offset = std::distance((const uint8_t*)BlockView.GetData(), ReadPtr);
+ OutHeaderSize = Offset;
for (uint64_t ChunkSize : ChunkSizes)
{
- IoBuffer Chunk(IoBuffer::Wrap, ReadPtr, ChunkSize);
+ IoBuffer Chunk(BlockPayload.AsIoBuffer(), Offset, ChunkSize);
IoHash AttachmentRawHash;
uint64_t AttachmentRawSize;
CompressedBuffer CompressedChunk = CompressedBuffer::FromCompressed(SharedBuffer(Chunk), AttachmentRawHash, AttachmentRawSize);
-
+ ZEN_ASSERT_SLOW(IoHash::HashBuffer(CompressedChunk.DecompressToComposite()) == AttachmentRawHash);
if (!CompressedChunk)
{
ZEN_ERROR("Invalid chunk in block");
return false;
}
Visitor(std::move(CompressedChunk), AttachmentRawHash);
- ReadPtr += ChunkSize;
- ZEN_ASSERT(ReadPtr <= BlockView.GetDataEnd());
+ Offset += ChunkSize;
+ ZEN_ASSERT(Offset <= BlockView.GetSize());
}
return true;
};
diff --git a/src/zenutil/chunkedcontent.cpp b/src/zenutil/chunkedcontent.cpp
new file mode 100644
index 000000000..a41b71972
--- /dev/null
+++ b/src/zenutil/chunkedcontent.cpp
@@ -0,0 +1,865 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/chunkedcontent.h>
+
+#include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
+#include <zencore/logging.h>
+#include <zencore/scopeguard.h>
+#include <zencore/timer.h>
+
+#include <zenutil/chunkedfile.h>
+#include <zenutil/chunkingcontroller.h>
+#include <zenutil/parallellwork.h>
+#include <zenutil/workerpools.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_set.h>
+#include <gsl/gsl-lite.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+using namespace std::literals;
+
+namespace {
+ void AddCunkSequence(ChunkingStatistics& Stats,
+ ChunkedContentData& InOutChunkedContent,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& ChunkHashToChunkIndex,
+ const IoHash& RawHash,
+ std::span<const uint32_t> ChunkSequence,
+ std::span<const IoHash> ChunkHashes,
+ std::span<const uint64_t> ChunkRawSizes)
+ {
+ ZEN_ASSERT(ChunkHashes.size() == ChunkRawSizes.size());
+ InOutChunkedContent.ChunkCounts.push_back(gsl::narrow<uint32_t>(ChunkSequence.size()));
+ InOutChunkedContent.ChunkOrders.reserve(InOutChunkedContent.ChunkOrders.size() + ChunkSequence.size());
+
+ for (uint32_t ChunkedSequenceIndex : ChunkSequence)
+ {
+ const IoHash& ChunkHash = ChunkHashes[ChunkedSequenceIndex];
+ if (auto It = ChunkHashToChunkIndex.find(ChunkHash); It != ChunkHashToChunkIndex.end())
+ {
+ uint32_t ChunkIndex = gsl::narrow<uint32_t>(It->second);
+ InOutChunkedContent.ChunkOrders.push_back(ChunkIndex);
+ }
+ else
+ {
+ uint32_t ChunkIndex = gsl::narrow<uint32_t>(InOutChunkedContent.ChunkHashes.size());
+ ChunkHashToChunkIndex.insert_or_assign(ChunkHash, ChunkIndex);
+ InOutChunkedContent.ChunkHashes.push_back(ChunkHash);
+ InOutChunkedContent.ChunkRawSizes.push_back(ChunkRawSizes[ChunkedSequenceIndex]);
+ InOutChunkedContent.ChunkOrders.push_back(ChunkIndex);
+ Stats.UniqueChunksFound++;
+ Stats.UniqueBytesFound += ChunkRawSizes[ChunkedSequenceIndex];
+ }
+ }
+ InOutChunkedContent.SequenceRawHashes.push_back(RawHash);
+ Stats.UniqueSequencesFound++;
+ }
+
+ void AddCunkSequence(ChunkingStatistics& Stats,
+ ChunkedContentData& InOutChunkedContent,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& ChunkHashToChunkIndex,
+ const IoHash& RawHash,
+ const uint64_t RawSize)
+ {
+ InOutChunkedContent.ChunkCounts.push_back(1);
+
+ if (auto It = ChunkHashToChunkIndex.find(RawHash); It != ChunkHashToChunkIndex.end())
+ {
+ uint32_t ChunkIndex = gsl::narrow<uint32_t>(It->second);
+ InOutChunkedContent.ChunkOrders.push_back(ChunkIndex);
+ }
+ else
+ {
+ uint32_t ChunkIndex = gsl::narrow<uint32_t>(InOutChunkedContent.ChunkHashes.size());
+ ChunkHashToChunkIndex.insert_or_assign(RawHash, ChunkIndex);
+ InOutChunkedContent.ChunkHashes.push_back(RawHash);
+ InOutChunkedContent.ChunkRawSizes.push_back(RawSize);
+ InOutChunkedContent.ChunkOrders.push_back(ChunkIndex);
+ Stats.UniqueChunksFound++;
+ Stats.UniqueBytesFound += RawSize;
+ }
+ InOutChunkedContent.SequenceRawHashes.push_back(RawHash);
+ Stats.UniqueSequencesFound++;
+ }
+
+ IoHash HashOneFile(ChunkingStatistics& Stats,
+ const ChunkingController& InChunkingController,
+ ChunkedFolderContent& OutChunkedContent,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& ChunkHashToChunkIndex,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& RawHashToSequenceRawHashIndex,
+ RwLock& Lock,
+ const std::filesystem::path& FolderPath,
+ uint32_t PathIndex)
+ {
+ const uint64_t RawSize = OutChunkedContent.RawSizes[PathIndex];
+ const std::filesystem::path& Path = OutChunkedContent.Paths[PathIndex];
+
+ if (RawSize == 0)
+ {
+ return IoHash::Zero;
+ }
+ else
+ {
+ ChunkedInfoWithSource Chunked;
+ const bool DidChunking =
+ InChunkingController.ProcessFile((FolderPath / Path).make_preferred(), RawSize, Chunked, Stats.BytesHashed);
+ if (DidChunking)
+ {
+ Lock.WithExclusiveLock([&]() {
+ if (!RawHashToSequenceRawHashIndex.contains(Chunked.Info.RawHash))
+ {
+ RawHashToSequenceRawHashIndex.insert(
+ {Chunked.Info.RawHash, gsl::narrow<uint32_t>(OutChunkedContent.ChunkedContent.SequenceRawHashes.size())});
+ std::vector<uint64_t> ChunkSizes;
+ ChunkSizes.reserve(Chunked.ChunkSources.size());
+ for (const ChunkSource& Source : Chunked.ChunkSources)
+ {
+ ChunkSizes.push_back(Source.Size);
+ }
+ AddCunkSequence(Stats,
+ OutChunkedContent.ChunkedContent,
+ ChunkHashToChunkIndex,
+ Chunked.Info.RawHash,
+ Chunked.Info.ChunkSequence,
+ Chunked.Info.ChunkHashes,
+ ChunkSizes);
+ Stats.UniqueSequencesFound++;
+ }
+ });
+ Stats.FilesChunked++;
+ return Chunked.Info.RawHash;
+ }
+ else
+ {
+ IoBuffer Buffer = IoBufferBuilder::MakeFromFile((FolderPath / Path).make_preferred());
+ const IoHash Hash = IoHash::HashBuffer(Buffer, &Stats.BytesHashed);
+
+ Lock.WithExclusiveLock([&]() {
+ if (!RawHashToSequenceRawHashIndex.contains(Hash))
+ {
+ RawHashToSequenceRawHashIndex.insert(
+ {Hash, gsl::narrow<uint32_t>(OutChunkedContent.ChunkedContent.SequenceRawHashes.size())});
+ AddCunkSequence(Stats, OutChunkedContent.ChunkedContent, ChunkHashToChunkIndex, Hash, RawSize);
+ Stats.UniqueSequencesFound++;
+ }
+ });
+ return Hash;
+ }
+ }
+ }
+
+ std::string PathCompareString(const std::filesystem::path& Path) { return ToLower(Path.generic_string()); }
+
+} // namespace
+
+std::string_view FolderContentSourcePlatformNames[(size_t)SourcePlatform::_Count] = {"Windows"sv, "Linux"sv, "MacOS"sv};
+
+std::string_view
+ToString(SourcePlatform Platform)
+{
+ return FolderContentSourcePlatformNames[(size_t)Platform];
+}
+
+SourcePlatform
+FromString(std::string_view Platform, SourcePlatform Default)
+{
+ for (size_t Index = 0; Index < (size_t)SourcePlatform::_Count; Index++)
+ {
+ if (Platform == FolderContentSourcePlatformNames[Index])
+ {
+ return (SourcePlatform)Index;
+ }
+ }
+ return Default;
+}
+
+SourcePlatform
+GetSourceCurrentPlatform()
+{
+#if ZEN_PLATFORM_WINDOWS
+ return SourcePlatform::Windows;
+#endif
+#if ZEN_PLATFORM_MAC
+ return SourcePlatform::MacOS;
+#endif
+#if ZEN_PLATFORM_LINUX
+ return SourcePlatform::Linux;
+#endif
+}
+
+bool
+FolderContent::AreFileAttributesEqual(const uint32_t Lhs, const uint32_t Rhs)
+{
+#if ZEN_PLATFORM_WINDOWS
+ return (Lhs & 0xff) == (Rhs & 0xff);
+#endif
+#if ZEN_PLATFORM_MAC
+ return Lhs == Rhs;
+#endif
+#if ZEN_PLATFORM_LINUX
+ return Lhs == Rhs;
+#endif
+}
+
+bool
+FolderContent::operator==(const FolderContent& Rhs) const
+{
+ if ((Platform == Rhs.Platform) && (RawSizes == Rhs.RawSizes) && (Attributes == Rhs.Attributes) &&
+ (ModificationTicks == Rhs.ModificationTicks) && (Paths.size() == Rhs.Paths.size()))
+ {
+ size_t PathCount = 0;
+ for (size_t PathIndex = 0; PathIndex < PathCount; PathIndex++)
+ {
+ if (Paths[PathIndex].generic_string() != Rhs.Paths[PathIndex].generic_string())
+ {
+ return false;
+ }
+ }
+ return true;
+ }
+ return false;
+}
+
+bool
+FolderContent::AreKnownFilesEqual(const FolderContent& Rhs) const
+{
+ tsl::robin_map<std::string, size_t> RhsPathToIndex;
+ const size_t RhsPathCount = Rhs.Paths.size();
+ RhsPathToIndex.reserve(RhsPathCount);
+ for (size_t RhsPathIndex = 0; RhsPathIndex < RhsPathCount; RhsPathIndex++)
+ {
+ RhsPathToIndex.insert({Rhs.Paths[RhsPathIndex].generic_string(), RhsPathIndex});
+ }
+ const size_t PathCount = Paths.size();
+ for (size_t PathIndex = 0; PathIndex < PathCount; PathIndex++)
+ {
+ if (auto It = RhsPathToIndex.find(Paths[PathIndex].generic_string()); It != RhsPathToIndex.end())
+ {
+ const size_t RhsPathIndex = It->second;
+ if ((RawSizes[PathIndex] != Rhs.RawSizes[RhsPathIndex]) ||
+ (!AreFileAttributesEqual(Attributes[PathIndex], Rhs.Attributes[RhsPathIndex])) ||
+ (ModificationTicks[PathIndex] != Rhs.ModificationTicks[RhsPathIndex]))
+ {
+ return false;
+ }
+ }
+ else
+ {
+ return false;
+ }
+ }
+ return true;
+}
+
+void
+FolderContent::UpdateState(const FolderContent& Rhs, std::vector<uint32_t>& OutPathIndexesOufOfDate)
+{
+ tsl::robin_map<std::string, uint32_t> RhsPathToIndex;
+ const uint32_t RhsPathCount = gsl::narrow<uint32_t>(Rhs.Paths.size());
+ RhsPathToIndex.reserve(RhsPathCount);
+ for (uint32_t RhsPathIndex = 0; RhsPathIndex < RhsPathCount; RhsPathIndex++)
+ {
+ RhsPathToIndex.insert({Rhs.Paths[RhsPathIndex].generic_string(), RhsPathIndex});
+ }
+ uint32_t PathCount = gsl::narrow<uint32_t>(Paths.size());
+ for (uint32_t PathIndex = 0; PathIndex < PathCount;)
+ {
+ if (auto It = RhsPathToIndex.find(Paths[PathIndex].generic_string()); It != RhsPathToIndex.end())
+ {
+ const uint32_t RhsPathIndex = It->second;
+
+ if ((RawSizes[PathIndex] != Rhs.RawSizes[RhsPathIndex]) ||
+ (ModificationTicks[PathIndex] != Rhs.ModificationTicks[RhsPathIndex]))
+ {
+ RawSizes[PathIndex] = Rhs.RawSizes[RhsPathIndex];
+ ModificationTicks[PathIndex] = Rhs.ModificationTicks[RhsPathIndex];
+ OutPathIndexesOufOfDate.push_back(PathIndex);
+ }
+ Attributes[PathIndex] = Rhs.Attributes[RhsPathIndex];
+ PathIndex++;
+ }
+ else
+ {
+ Paths.erase(Paths.begin() + PathIndex);
+ RawSizes.erase(RawSizes.begin() + PathIndex);
+ Attributes.erase(Attributes.begin() + PathIndex);
+ ModificationTicks.erase(ModificationTicks.begin() + PathIndex);
+ PathCount--;
+ }
+ }
+}
+
+FolderContent
+GetUpdatedContent(const FolderContent& Old, const FolderContent& New, std::vector<std::filesystem::path>& OutDeletedPathIndexes)
+{
+ FolderContent Result = {.Platform = Old.Platform};
+ tsl::robin_map<std::string, uint32_t> NewPathToIndex;
+ const uint32_t NewPathCount = gsl::narrow<uint32_t>(New.Paths.size());
+ NewPathToIndex.reserve(NewPathCount);
+ for (uint32_t NewPathIndex = 0; NewPathIndex < NewPathCount; NewPathIndex++)
+ {
+ NewPathToIndex.insert({New.Paths[NewPathIndex].generic_string(), NewPathIndex});
+ }
+ uint32_t OldPathCount = gsl::narrow<uint32_t>(Old.Paths.size());
+ for (uint32_t OldPathIndex = 0; OldPathIndex < OldPathCount; OldPathIndex++)
+ {
+ if (auto It = NewPathToIndex.find(Old.Paths[OldPathIndex].generic_string()); It != NewPathToIndex.end())
+ {
+ const uint32_t NewPathIndex = It->second;
+
+ if ((Old.RawSizes[OldPathIndex] != New.RawSizes[NewPathIndex]) ||
+ (Old.ModificationTicks[OldPathIndex] != New.ModificationTicks[NewPathIndex]))
+ {
+ Result.Paths.push_back(New.Paths[NewPathIndex]);
+ Result.RawSizes.push_back(New.RawSizes[NewPathIndex]);
+ Result.Attributes.push_back(New.Attributes[NewPathIndex]);
+ Result.ModificationTicks.push_back(New.ModificationTicks[NewPathIndex]);
+ }
+ }
+ else
+ {
+ OutDeletedPathIndexes.push_back(Old.Paths[OldPathIndex]);
+ }
+ }
+ return Result;
+}
+
+void
+SaveFolderContentToCompactBinary(const FolderContent& Content, CbWriter& Output)
+{
+ Output.AddString("platform"sv, ToString(Content.Platform));
+ compactbinary_helpers::WriteArray(Content.Paths, "paths"sv, Output);
+ compactbinary_helpers::WriteArray(Content.RawSizes, "rawSizes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.Attributes, "attributes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.ModificationTicks, "modificationTimes"sv, Output);
+}
+
+FolderContent
+LoadFolderContentToCompactBinary(CbObjectView Input)
+{
+ FolderContent Content;
+ Content.Platform = FromString(Input["platform"sv].AsString(), GetSourceCurrentPlatform());
+ compactbinary_helpers::ReadArray("paths"sv, Input, Content.Paths);
+ compactbinary_helpers::ReadArray("rawSizes"sv, Input, Content.RawSizes);
+ compactbinary_helpers::ReadArray("attributes"sv, Input, Content.Attributes);
+ compactbinary_helpers::ReadArray("modificationTimes"sv, Input, Content.ModificationTicks);
+ return Content;
+}
+
+FolderContent
+GetFolderContent(GetFolderContentStatistics& Stats,
+ const std::filesystem::path& RootPath,
+ std::function<bool(const std::string_view& RelativePath)>&& AcceptDirectory,
+ std::function<bool(std::string_view RelativePath, uint64_t Size, uint32_t Attributes)>&& AcceptFile,
+ WorkerThreadPool& WorkerPool,
+ int32_t UpdateInteralMS,
+ std::function<void(bool IsAborted, std::ptrdiff_t PendingWork)>&& UpdateCallback,
+ std::atomic<bool>& AbortFlag)
+{
+ Stopwatch Timer;
+ auto _ = MakeGuard([&Stats, &Timer]() { Stats.ElapsedWallTimeUS = Timer.GetElapsedTimeUs(); });
+
+ FolderContent Content;
+ struct AsyncVisitor : public GetDirectoryContentVisitor
+ {
+ AsyncVisitor(GetFolderContentStatistics& Stats,
+ std::atomic<bool>& AbortFlag,
+ FolderContent& Content,
+ std::function<bool(const std::string_view& RelativePath)>&& AcceptDirectory,
+ std::function<bool(std::string_view RelativePath, uint64_t Size, uint32_t Attributes)>&& AcceptFile)
+ : m_Stats(Stats)
+ , m_AbortFlag(AbortFlag)
+ , m_FoundContent(Content)
+ , m_AcceptDirectory(std::move(AcceptDirectory))
+ , m_AcceptFile(std::move(AcceptFile))
+ {
+ }
+ virtual void AsyncVisitDirectory(const std::filesystem::path& RelativeRoot, DirectoryContent&& Content) override
+ {
+ if (!m_AbortFlag)
+ {
+ m_Stats.FoundFileCount += Content.FileNames.size();
+ for (uint64_t FileSize : Content.FileSizes)
+ {
+ m_Stats.FoundFileByteCount += FileSize;
+ }
+ std::string RelativeDirectoryPath = RelativeRoot.generic_string();
+ if (m_AcceptDirectory(RelativeDirectoryPath))
+ {
+ std::vector<std::filesystem::path> Paths;
+ std::vector<uint64_t> RawSizes;
+ std::vector<uint32_t> Attributes;
+ std::vector<uint64_t> ModificatonTicks;
+ Paths.reserve(Content.FileNames.size());
+ RawSizes.reserve(Content.FileNames.size());
+ Attributes.reserve(Content.FileNames.size());
+ ModificatonTicks.reserve(Content.FileModificationTicks.size());
+
+ for (size_t FileIndex = 0; FileIndex < Content.FileNames.size(); FileIndex++)
+ {
+ const std::filesystem::path& FileName = Content.FileNames[FileIndex];
+ std::string RelativePath = (RelativeRoot / FileName).generic_string();
+ std::replace(RelativePath.begin(), RelativePath.end(), '\\', '/');
+ if (m_AcceptFile(RelativePath, Content.FileSizes[FileIndex], Content.FileAttributes[FileIndex]))
+ {
+ Paths.emplace_back(std::move(RelativePath));
+ RawSizes.emplace_back(Content.FileSizes[FileIndex]);
+ Attributes.emplace_back(Content.FileAttributes[FileIndex]);
+ ModificatonTicks.emplace_back(Content.FileModificationTicks[FileIndex]);
+
+ m_Stats.AcceptedFileCount++;
+ m_Stats.AcceptedFileByteCount += Content.FileSizes[FileIndex];
+ }
+ }
+ m_Lock.WithExclusiveLock([&]() {
+ m_FoundContent.Paths.insert(m_FoundContent.Paths.end(), Paths.begin(), Paths.end());
+ m_FoundContent.RawSizes.insert(m_FoundContent.RawSizes.end(), RawSizes.begin(), RawSizes.end());
+ m_FoundContent.Attributes.insert(m_FoundContent.Attributes.end(), Attributes.begin(), Attributes.end());
+ m_FoundContent.ModificationTicks.insert(m_FoundContent.ModificationTicks.end(),
+ ModificatonTicks.begin(),
+ ModificatonTicks.end());
+ });
+ }
+ }
+ }
+
+ GetFolderContentStatistics& m_Stats;
+ std::atomic<bool>& m_AbortFlag;
+ RwLock m_Lock;
+ FolderContent& m_FoundContent;
+ std::function<bool(const std::string_view& RelativePath)> m_AcceptDirectory;
+ std::function<bool(std::string_view RelativePath, uint64_t Size, uint32_t Attributes)> m_AcceptFile;
+ } Visitor(Stats, AbortFlag, Content, std::move(AcceptDirectory), std::move(AcceptFile));
+
+ Latch PendingWork(1);
+ GetDirectoryContent(RootPath,
+ DirectoryContentFlags::IncludeFiles | DirectoryContentFlags::Recursive | DirectoryContentFlags::IncludeFileSizes |
+ DirectoryContentFlags::IncludeAttributes | DirectoryContentFlags::IncludeModificationTick,
+ Visitor,
+ WorkerPool,
+ PendingWork);
+ PendingWork.CountDown();
+ while (!PendingWork.Wait(UpdateInteralMS))
+ {
+ UpdateCallback(AbortFlag.load(), PendingWork.Remaining());
+ }
+ std::vector<size_t> Order;
+ size_t PathCount = Content.Paths.size();
+ Order.resize(Content.Paths.size());
+ std::vector<std::string> Parents;
+ Parents.reserve(PathCount);
+ std::vector<std::string> Filenames;
+ Filenames.reserve(PathCount);
+ for (size_t OrderIndex = 0; OrderIndex < PathCount; OrderIndex++)
+ {
+ Order[OrderIndex] = OrderIndex;
+ Parents.emplace_back(Content.Paths[OrderIndex].parent_path().generic_string());
+ Filenames.emplace_back(Content.Paths[OrderIndex].filename().generic_string());
+ }
+ std::sort(Order.begin(), Order.end(), [&Parents, &Filenames](size_t Lhs, size_t Rhs) {
+ const std::string& LhsParent = Parents[Lhs];
+ const std::string& RhsParent = Parents[Rhs];
+ if (LhsParent < RhsParent)
+ {
+ return true;
+ }
+ else if (LhsParent > RhsParent)
+ {
+ return false;
+ }
+ return Filenames[Lhs] < Filenames[Rhs];
+ });
+ FolderContent OrderedContent;
+ OrderedContent.Paths.reserve(PathCount);
+ OrderedContent.RawSizes.reserve(PathCount);
+ OrderedContent.Attributes.reserve(PathCount);
+ OrderedContent.ModificationTicks.reserve(PathCount);
+ for (size_t OrderIndex : Order)
+ {
+ OrderedContent.Paths.emplace_back(std::move(Content.Paths[OrderIndex]));
+ OrderedContent.RawSizes.emplace_back(Content.RawSizes[OrderIndex]);
+ OrderedContent.Attributes.emplace_back(Content.Attributes[OrderIndex]);
+ OrderedContent.ModificationTicks.emplace_back(Content.ModificationTicks[OrderIndex]);
+ }
+ return OrderedContent;
+}
+
+void
+SaveChunkedFolderContentToCompactBinary(const ChunkedFolderContent& Content, CbWriter& Output)
+{
+ Output.AddString("platform"sv, ToString(Content.Platform));
+ compactbinary_helpers::WriteArray(Content.Paths, "paths"sv, Output);
+ compactbinary_helpers::WriteArray(Content.RawSizes, "rawSizes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.Attributes, "attributes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.RawHashes, "rawHashes"sv, Output);
+
+ Output.BeginObject("chunkedContent");
+ compactbinary_helpers::WriteArray(Content.ChunkedContent.SequenceRawHashes, "sequenceRawHashes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.ChunkedContent.ChunkCounts, "chunkCounts"sv, Output);
+ compactbinary_helpers::WriteArray(Content.ChunkedContent.ChunkOrders, "chunkOrders"sv, Output);
+ compactbinary_helpers::WriteArray(Content.ChunkedContent.ChunkHashes, "chunkHashes"sv, Output);
+ compactbinary_helpers::WriteArray(Content.ChunkedContent.ChunkRawSizes, "chunkRawSizes"sv, Output);
+ Output.EndObject(); // chunkedContent
+}
+
+ChunkedFolderContent
+LoadChunkedFolderContentToCompactBinary(CbObjectView Input)
+{
+ ChunkedFolderContent Content;
+ Content.Platform = FromString(Input["platform"sv].AsString(), GetSourceCurrentPlatform());
+ compactbinary_helpers::ReadArray("paths"sv, Input, Content.Paths);
+ compactbinary_helpers::ReadArray("rawSizes"sv, Input, Content.RawSizes);
+ compactbinary_helpers::ReadArray("attributes"sv, Input, Content.Attributes);
+ compactbinary_helpers::ReadArray("rawHashes"sv, Input, Content.RawHashes);
+
+ CbObjectView ChunkedContentView = Input["chunkedContent"sv].AsObjectView();
+ compactbinary_helpers::ReadArray("sequenceRawHashes"sv, ChunkedContentView, Content.ChunkedContent.SequenceRawHashes);
+ compactbinary_helpers::ReadArray("chunkCounts"sv, ChunkedContentView, Content.ChunkedContent.ChunkCounts);
+ compactbinary_helpers::ReadArray("chunkOrders"sv, ChunkedContentView, Content.ChunkedContent.ChunkOrders);
+ compactbinary_helpers::ReadArray("chunkHashes"sv, ChunkedContentView, Content.ChunkedContent.ChunkHashes);
+ compactbinary_helpers::ReadArray("chunkRawSizes"sv, ChunkedContentView, Content.ChunkedContent.ChunkRawSizes);
+ return Content;
+}
+
+ChunkedFolderContent
+MergeChunkedFolderContents(const ChunkedFolderContent& Base, std::span<const ChunkedFolderContent> Overlays)
+{
+ ZEN_ASSERT(!Overlays.empty());
+
+ ChunkedFolderContent Result;
+ const size_t BasePathCount = Base.Paths.size();
+ Result.Paths.reserve(BasePathCount);
+ Result.RawSizes.reserve(BasePathCount);
+ Result.Attributes.reserve(BasePathCount);
+ Result.RawHashes.reserve(BasePathCount);
+
+ const size_t BaseChunkCount = Base.ChunkedContent.ChunkHashes.size();
+ Result.ChunkedContent.SequenceRawHashes.reserve(Base.ChunkedContent.SequenceRawHashes.size());
+ Result.ChunkedContent.ChunkCounts.reserve(BaseChunkCount);
+ Result.ChunkedContent.ChunkHashes.reserve(BaseChunkCount);
+ Result.ChunkedContent.ChunkRawSizes.reserve(BaseChunkCount);
+ Result.ChunkedContent.ChunkOrders.reserve(Base.ChunkedContent.ChunkOrders.size());
+
+ tsl::robin_map<std::string, std::filesystem::path> GenericPathToActualPath;
+ for (const std::filesystem::path& Path : Base.Paths)
+ {
+ GenericPathToActualPath.insert({PathCompareString(Path), Path});
+ }
+ for (const ChunkedFolderContent& Overlay : Overlays)
+ {
+ for (const std::filesystem::path& Path : Overlay.Paths)
+ {
+ GenericPathToActualPath.insert({PathCompareString(Path), Path});
+ }
+ }
+
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> RawHashToSequenceRawHashIndex;
+
+ auto BuildOverlayPaths = [](std::span<const ChunkedFolderContent> Overlays) -> tsl::robin_set<std::string> {
+ tsl::robin_set<std::string> Result;
+ for (const ChunkedFolderContent& OverlayContent : Overlays)
+ {
+ for (const std::filesystem::path& Path : OverlayContent.Paths)
+ {
+ Result.insert(PathCompareString(Path));
+ }
+ }
+ return Result;
+ };
+
+ auto AddContent = [&BuildOverlayPaths](ChunkedFolderContent& Result,
+ const ChunkedFolderContent& OverlayContent,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& ChunkHashToChunkIndex,
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher>& RawHashToSequenceRawHashIndex,
+ const tsl::robin_map<std::string, std::filesystem::path>& GenericPathToActualPath,
+ std::span<const ChunkedFolderContent> Overlays) {
+ const ChunkedContentLookup OverlayLookup = BuildChunkedContentLookup(OverlayContent);
+ tsl::robin_set<std::string> BaseOverlayPaths = BuildOverlayPaths(Overlays);
+ for (uint32_t PathIndex = 0; PathIndex < OverlayContent.Paths.size(); PathIndex++)
+ {
+ std::string GenericPath = PathCompareString(OverlayContent.Paths[PathIndex]);
+ if (!BaseOverlayPaths.contains(GenericPath))
+ {
+ // This asset will not be overridden by a later layer - add it
+
+ const std::filesystem::path OriginalPath = GenericPathToActualPath.at(GenericPath);
+ Result.Paths.push_back(OriginalPath);
+ const IoHash& RawHash = OverlayContent.RawHashes[PathIndex];
+ Result.RawSizes.push_back(OverlayContent.RawSizes[PathIndex]);
+ Result.Attributes.push_back(OverlayContent.Attributes[PathIndex]);
+ Result.RawHashes.push_back(RawHash);
+
+ if (OverlayContent.RawSizes[PathIndex] > 0)
+ {
+ if (!RawHashToSequenceRawHashIndex.contains(RawHash))
+ {
+ RawHashToSequenceRawHashIndex.insert(
+ {RawHash, gsl::narrow<uint32_t>(Result.ChunkedContent.SequenceRawHashes.size())});
+ const uint32_t SequenceRawHashIndex = OverlayLookup.RawHashToSequenceRawHashIndex.at(RawHash);
+ const uint32_t OrderIndexOffset = OverlayLookup.SequenceRawHashIndexChunkOrderOffset[SequenceRawHashIndex];
+ const uint32_t ChunkCount = OverlayContent.ChunkedContent.ChunkCounts[SequenceRawHashIndex];
+ ChunkingStatistics Stats;
+ std::span<const uint32_t> OriginalChunkOrder =
+ std::span<const uint32_t>(OverlayContent.ChunkedContent.ChunkOrders).subspan(OrderIndexOffset, ChunkCount);
+ AddCunkSequence(Stats,
+ Result.ChunkedContent,
+ ChunkHashToChunkIndex,
+ RawHash,
+ OriginalChunkOrder,
+ OverlayContent.ChunkedContent.ChunkHashes,
+ OverlayContent.ChunkedContent.ChunkRawSizes);
+ Stats.UniqueSequencesFound++;
+ }
+ }
+ }
+ }
+ };
+
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> MergedChunkHashToChunkIndex;
+ AddContent(Result, Base, MergedChunkHashToChunkIndex, RawHashToSequenceRawHashIndex, GenericPathToActualPath, Overlays);
+ for (uint32_t OverlayIndex = 0; OverlayIndex < Overlays.size(); OverlayIndex++)
+ {
+ AddContent(Result,
+ Overlays[OverlayIndex],
+ MergedChunkHashToChunkIndex,
+ RawHashToSequenceRawHashIndex,
+ GenericPathToActualPath,
+ Overlays.subspan(OverlayIndex + 1));
+ }
+ return Result;
+}
+
+ChunkedFolderContent
+DeletePathsFromChunkedContent(const ChunkedFolderContent& BaseContent, std::span<const std::filesystem::path> DeletedPaths)
+{
+ ZEN_ASSERT(DeletedPaths.size() <= BaseContent.Paths.size());
+ ChunkedFolderContent Result = {.Platform = BaseContent.Platform};
+ if (DeletedPaths.size() < BaseContent.Paths.size())
+ {
+ tsl::robin_set<std::string> DeletedPathSet;
+ DeletedPathSet.reserve(DeletedPaths.size());
+ for (const std::filesystem::path& DeletedPath : DeletedPaths)
+ {
+ DeletedPathSet.insert(PathCompareString(DeletedPath));
+ }
+ const ChunkedContentLookup BaseLookup = BuildChunkedContentLookup(BaseContent);
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> ChunkHashToChunkIndex;
+
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> RawHashToSequenceRawHashIndex;
+ for (uint32_t PathIndex = 0; PathIndex < BaseContent.Paths.size(); PathIndex++)
+ {
+ const std::filesystem::path& Path = BaseContent.Paths[PathIndex];
+ if (!DeletedPathSet.contains(PathCompareString(Path)))
+ {
+ const IoHash& RawHash = BaseContent.RawHashes[PathIndex];
+ const uint64_t RawSize = BaseContent.RawSizes[PathIndex];
+ Result.Paths.push_back(Path);
+ Result.RawSizes.push_back(RawSize);
+ Result.Attributes.push_back(BaseContent.Attributes[PathIndex]);
+ Result.RawHashes.push_back(RawHash);
+ if (RawSize > 0)
+ {
+ if (!RawHashToSequenceRawHashIndex.contains(RawHash))
+ {
+ RawHashToSequenceRawHashIndex.insert(
+ {RawHash, gsl::narrow<uint32_t>(Result.ChunkedContent.SequenceRawHashes.size())});
+ const uint32_t SequenceRawHashIndex = BaseLookup.RawHashToSequenceRawHashIndex.at(RawHash);
+ const uint32_t OrderIndexOffset = BaseLookup.SequenceRawHashIndexChunkOrderOffset[SequenceRawHashIndex];
+ const uint32_t ChunkCount = BaseContent.ChunkedContent.ChunkCounts[SequenceRawHashIndex];
+ ChunkingStatistics Stats;
+ std::span<const uint32_t> OriginalChunkOrder =
+ std::span<const uint32_t>(BaseContent.ChunkedContent.ChunkOrders).subspan(OrderIndexOffset, ChunkCount);
+ AddCunkSequence(Stats,
+ Result.ChunkedContent,
+ ChunkHashToChunkIndex,
+ RawHash,
+ OriginalChunkOrder,
+ BaseContent.ChunkedContent.ChunkHashes,
+ BaseContent.ChunkedContent.ChunkRawSizes);
+ Stats.UniqueSequencesFound++;
+ }
+ }
+ }
+ }
+ }
+ return Result;
+}
+
+ChunkedFolderContent
+ChunkFolderContent(ChunkingStatistics& Stats,
+ WorkerThreadPool& WorkerPool,
+ const std::filesystem::path& RootPath,
+ const FolderContent& Content,
+ const ChunkingController& InChunkingController,
+ int32_t UpdateInteralMS,
+ std::function<void(bool IsAborted, std::ptrdiff_t PendingWork)>&& UpdateCallback,
+ std::atomic<bool>& AbortFlag)
+{
+ Stopwatch Timer;
+ auto _ = MakeGuard([&Stats, &Timer]() { Stats.ElapsedWallTimeUS = Timer.GetElapsedTimeUs(); });
+
+ ChunkedFolderContent Result = {.Platform = Content.Platform,
+ .Paths = Content.Paths,
+ .RawSizes = Content.RawSizes,
+ .Attributes = Content.Attributes};
+ const size_t ItemCount = Result.Paths.size();
+ Result.RawHashes.resize(ItemCount, IoHash::Zero);
+ Result.ChunkedContent.SequenceRawHashes.reserve(ItemCount); // Up to 1 per file, maybe less
+ Result.ChunkedContent.ChunkCounts.reserve(ItemCount); // Up to one per file
+ Result.ChunkedContent.ChunkOrders.reserve(ItemCount); // At least 1 per file, maybe more
+ Result.ChunkedContent.ChunkHashes.reserve(ItemCount); // At least 1 per file, maybe more
+ Result.ChunkedContent.ChunkRawSizes.reserve(ItemCount); // At least 1 per file, maybe more
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> ChunkHashToChunkIndex;
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> RawHashToChunkSequenceIndex;
+ RawHashToChunkSequenceIndex.reserve(ItemCount);
+ ChunkHashToChunkIndex.reserve(ItemCount);
+ {
+ std::vector<uint32_t> Order;
+ Order.resize(ItemCount);
+ for (uint32_t I = 0; I < ItemCount; I++)
+ {
+ Order[I] = I;
+ }
+
+ // Handle the biggest files first so we don't end up with one straggling large file at the end
+ // std::sort(Order.begin(), Order.end(), [&](uint32_t Lhs, uint32_t Rhs) { return Result.RawSizes[Lhs] > Result.RawSizes[Rhs];
+ //});
+
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> RawHashToSequenceRawHashIndex;
+ RawHashToSequenceRawHashIndex.reserve(ItemCount);
+
+ RwLock Lock;
+
+ ParallellWork Work(AbortFlag);
+
+ for (uint32_t PathIndex : Order)
+ {
+ if (Work.IsAborted())
+ {
+ break;
+ }
+ Work.ScheduleWork(
+ WorkerPool, // GetSyncWorkerPool()
+ [&, PathIndex](std::atomic<bool>& AbortFlag) {
+ if (!AbortFlag)
+ {
+ IoHash RawHash = HashOneFile(Stats,
+ InChunkingController,
+ Result,
+ ChunkHashToChunkIndex,
+ RawHashToSequenceRawHashIndex,
+ Lock,
+ RootPath,
+ PathIndex);
+ Lock.WithExclusiveLock([&]() { Result.RawHashes[PathIndex] = RawHash; });
+ Stats.FilesProcessed++;
+ }
+ },
+ [&, PathIndex](const std::exception& Ex, std::atomic<bool>& AbortFlag) {
+ ZEN_CONSOLE("Failed scanning file {}. Reason: {}", Result.Paths[PathIndex], Ex.what());
+ AbortFlag = true;
+ });
+ }
+
+ Work.Wait(UpdateInteralMS, [&](bool IsAborted, std::ptrdiff_t PendingWork) {
+ ZEN_UNUSED(IsAborted);
+ ZEN_UNUSED(PendingWork);
+ UpdateCallback(Work.IsAborted(), Work.PendingWork().Remaining());
+ });
+ }
+ return Result;
+}
+
+ChunkedContentLookup
+BuildChunkedContentLookup(const ChunkedFolderContent& Content)
+{
+ struct ChunkLocationReference
+ {
+ uint32_t ChunkIndex;
+ ChunkedContentLookup::ChunkLocation Location;
+ };
+
+ ChunkedContentLookup Result;
+ {
+ const uint32_t SequenceRawHashesCount = gsl::narrow<uint32_t>(Content.ChunkedContent.SequenceRawHashes.size());
+ Result.RawHashToSequenceRawHashIndex.reserve(SequenceRawHashesCount);
+ Result.SequenceRawHashIndexChunkOrderOffset.reserve(SequenceRawHashesCount);
+ uint32_t OrderOffset = 0;
+ for (uint32_t SequenceRawHashIndex = 0; SequenceRawHashIndex < Content.ChunkedContent.SequenceRawHashes.size();
+ SequenceRawHashIndex++)
+ {
+ Result.RawHashToSequenceRawHashIndex.insert(
+ {Content.ChunkedContent.SequenceRawHashes[SequenceRawHashIndex], SequenceRawHashIndex});
+ Result.SequenceRawHashIndexChunkOrderOffset.push_back(OrderOffset);
+ OrderOffset += Content.ChunkedContent.ChunkCounts[SequenceRawHashIndex];
+ }
+ }
+
+ std::vector<ChunkLocationReference> Locations;
+ Locations.reserve(Content.ChunkedContent.ChunkOrders.size());
+ for (uint32_t PathIndex = 0; PathIndex < Content.Paths.size(); PathIndex++)
+ {
+ if (Content.RawSizes[PathIndex] > 0)
+ {
+ const IoHash& RawHash = Content.RawHashes[PathIndex];
+ uint32_t SequenceRawHashIndex = Result.RawHashToSequenceRawHashIndex.at(RawHash);
+ const uint32_t OrderOffset = Result.SequenceRawHashIndexChunkOrderOffset[SequenceRawHashIndex];
+ const uint32_t ChunkCount = Content.ChunkedContent.ChunkCounts[SequenceRawHashIndex];
+ uint64_t LocationOffset = 0;
+ for (size_t OrderIndex = OrderOffset; OrderIndex < OrderOffset + ChunkCount; OrderIndex++)
+ {
+ uint32_t ChunkIndex = Content.ChunkedContent.ChunkOrders[OrderIndex];
+
+ Locations.push_back(ChunkLocationReference{ChunkIndex, ChunkedContentLookup::ChunkLocation{PathIndex, LocationOffset}});
+
+ LocationOffset += Content.ChunkedContent.ChunkRawSizes[ChunkIndex];
+ }
+ ZEN_ASSERT(LocationOffset == Content.RawSizes[PathIndex]);
+ }
+ }
+
+ std::sort(Locations.begin(), Locations.end(), [](const ChunkLocationReference& Lhs, const ChunkLocationReference& Rhs) {
+ if (Lhs.ChunkIndex < Rhs.ChunkIndex)
+ {
+ return true;
+ }
+ if (Lhs.ChunkIndex > Rhs.ChunkIndex)
+ {
+ return false;
+ }
+ if (Lhs.Location.PathIndex < Rhs.Location.PathIndex)
+ {
+ return true;
+ }
+ if (Lhs.Location.PathIndex > Rhs.Location.PathIndex)
+ {
+ return false;
+ }
+ return Lhs.Location.Offset < Rhs.Location.Offset;
+ });
+
+ Result.ChunkLocations.reserve(Locations.size());
+ const uint32_t ChunkCount = gsl::narrow<uint32_t>(Content.ChunkedContent.ChunkHashes.size());
+ Result.ChunkHashToChunkIndex.reserve(ChunkCount);
+ size_t RangeOffset = 0;
+ for (uint32_t ChunkIndex = 0; ChunkIndex < ChunkCount; ChunkIndex++)
+ {
+ Result.ChunkHashToChunkIndex.insert({Content.ChunkedContent.ChunkHashes[ChunkIndex], ChunkIndex});
+ uint32_t Count = 0;
+ while (Locations[RangeOffset + Count].ChunkIndex == ChunkIndex)
+ {
+ Result.ChunkLocations.push_back(Locations[RangeOffset + Count].Location);
+ Count++;
+ }
+ Result.ChunkLocationOffset.push_back(RangeOffset);
+ Result.ChunkLocationCounts.push_back(Count);
+ RangeOffset += Count;
+ }
+
+ return Result;
+}
+
+} // namespace zen
diff --git a/src/zenutil/chunkingcontroller.cpp b/src/zenutil/chunkingcontroller.cpp
new file mode 100644
index 000000000..bc0e57b14
--- /dev/null
+++ b/src/zenutil/chunkingcontroller.cpp
@@ -0,0 +1,265 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/chunkingcontroller.h>
+
+#include <zencore/basicfile.h>
+#include <zencore/compactbinarybuilder.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_map.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+using namespace std::literals;
+
+namespace {
+ std::vector<std::string> ReadStringArray(CbArrayView StringArray)
+ {
+ std::vector<std::string> Result;
+ Result.reserve(StringArray.Num());
+ for (CbFieldView FieldView : StringArray)
+ {
+ Result.emplace_back(FieldView.AsString());
+ }
+ return Result;
+ }
+
+ ChunkedParams ReadChunkParams(CbObjectView Params)
+ {
+ bool UseThreshold = Params["UseThreshold"sv].AsBool(true);
+ size_t MinSize = Params["MinSize"sv].AsUInt64(DefaultChunkedParams.MinSize);
+ size_t MaxSize = Params["MaxSize"sv].AsUInt64(DefaultChunkedParams.MaxSize);
+ size_t AvgSize = Params["AvgSize"sv].AsUInt64(DefaultChunkedParams.AvgSize);
+
+ return ChunkedParams{.UseThreshold = UseThreshold, .MinSize = MinSize, .MaxSize = MaxSize, .AvgSize = AvgSize};
+ }
+
+} // namespace
+
+class BasicChunkingController : public ChunkingController
+{
+public:
+ BasicChunkingController(std::span<const std::string_view> ExcludeExtensions,
+ uint64_t ChunkFileSizeLimit,
+ const ChunkedParams& ChunkingParams)
+ : m_ChunkExcludeExtensions(ExcludeExtensions.begin(), ExcludeExtensions.end())
+ , m_ChunkFileSizeLimit(ChunkFileSizeLimit)
+ , m_ChunkingParams(ChunkingParams)
+ {
+ }
+
+ BasicChunkingController(CbObjectView Parameters)
+ : m_ChunkExcludeExtensions(ReadStringArray(Parameters["ChunkExcludeExtensions"sv].AsArrayView()))
+ , m_ChunkFileSizeLimit(Parameters["ChunkFileSizeLimit"sv].AsUInt64(DefaultChunkingFileSizeLimit))
+ , m_ChunkingParams(ReadChunkParams(Parameters["ChunkingParams"sv].AsObjectView()))
+ {
+ }
+
+ virtual bool ProcessFile(const std::filesystem::path& InputPath,
+ uint64_t RawSize,
+ ChunkedInfoWithSource& OutChunked,
+ std::atomic<uint64_t>& BytesProcessed) const override
+ {
+ const bool ExcludeFromChunking =
+ std::find(m_ChunkExcludeExtensions.begin(), m_ChunkExcludeExtensions.end(), InputPath.extension()) !=
+ m_ChunkExcludeExtensions.end();
+
+ if (ExcludeFromChunking || (RawSize < m_ChunkFileSizeLimit))
+ {
+ return false;
+ }
+
+ BasicFile Buffer(InputPath, BasicFile::Mode::kRead);
+ OutChunked = ChunkData(Buffer, 0, RawSize, m_ChunkingParams, &BytesProcessed);
+ return true;
+ }
+
+ virtual std::string_view GetName() const override { return Name; }
+
+ virtual CbObject GetParameters() const override
+ {
+ CbObjectWriter Writer;
+ Writer.BeginArray("ChunkExcludeExtensions"sv);
+ {
+ for (const std::string& Extension : m_ChunkExcludeExtensions)
+ {
+ Writer.AddString(Extension);
+ }
+ }
+ Writer.EndArray(); // ChunkExcludeExtensions
+ Writer.AddInteger("ChunkFileSizeLimit"sv, m_ChunkFileSizeLimit);
+ Writer.BeginObject("ChunkingParams"sv);
+ {
+ Writer.AddBool("UseThreshold"sv, m_ChunkingParams.UseThreshold);
+
+ Writer.AddInteger("MinSize"sv, (uint64_t)m_ChunkingParams.MinSize);
+ Writer.AddInteger("MaxSize"sv, (uint64_t)m_ChunkingParams.MaxSize);
+ Writer.AddInteger("AvgSize"sv, (uint64_t)m_ChunkingParams.AvgSize);
+ }
+ Writer.EndObject(); // ChunkingParams
+ return Writer.Save();
+ }
+ static constexpr std::string_view Name = "BasicChunkingController"sv;
+
+protected:
+ const std::vector<std::string> m_ChunkExcludeExtensions;
+ const uint64_t m_ChunkFileSizeLimit;
+ const ChunkedParams m_ChunkingParams;
+};
+
+class ChunkingControllerWithFixedChunking : public ChunkingController
+{
+public:
+ ChunkingControllerWithFixedChunking(std::span<const std::string_view> FixedChunkingExtensions,
+ uint64_t ChunkFileSizeLimit,
+ const ChunkedParams& ChunkingParams,
+ uint32_t FixedChunkingChunkSize)
+ : m_FixedChunkingExtensions(FixedChunkingExtensions.begin(), FixedChunkingExtensions.end())
+ , m_ChunkFileSizeLimit(ChunkFileSizeLimit)
+ , m_ChunkingParams(ChunkingParams)
+ , m_FixedChunkingChunkSize(FixedChunkingChunkSize)
+ {
+ }
+
+ ChunkingControllerWithFixedChunking(CbObjectView Parameters)
+ : m_FixedChunkingExtensions(ReadStringArray(Parameters["FixedChunkingExtensions"sv].AsArrayView()))
+ , m_ChunkFileSizeLimit(Parameters["ChunkFileSizeLimit"sv].AsUInt64(DefaultChunkingFileSizeLimit))
+ , m_ChunkingParams(ReadChunkParams(Parameters["ChunkingParams"sv].AsObjectView()))
+ , m_FixedChunkingChunkSize(Parameters["FixedChunkingChunkSize"sv].AsUInt32(16u * 1024u * 1024u))
+ {
+ }
+
+ virtual bool ProcessFile(const std::filesystem::path& InputPath,
+ uint64_t RawSize,
+ ChunkedInfoWithSource& OutChunked,
+ std::atomic<uint64_t>& BytesProcessed) const override
+ {
+ if (RawSize < m_ChunkFileSizeLimit)
+ {
+ return false;
+ }
+ const bool FixedChunking = std::find(m_FixedChunkingExtensions.begin(), m_FixedChunkingExtensions.end(), InputPath.extension()) !=
+ m_FixedChunkingExtensions.end();
+
+ if (FixedChunking)
+ {
+ IoHashStream FullHash;
+ IoBuffer Source = IoBufferBuilder::MakeFromFile(InputPath);
+ uint64_t Offset = 0;
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> ChunkHashToChunkIndex;
+ ChunkHashToChunkIndex.reserve(1 + (RawSize / m_FixedChunkingChunkSize));
+ while (Offset < RawSize)
+ {
+ uint64_t ChunkSize = std::min<uint64_t>(RawSize - Offset, m_FixedChunkingChunkSize);
+ IoBuffer Chunk(Source, Offset, ChunkSize);
+ MemoryView ChunkData = Chunk.GetView();
+ FullHash.Append(ChunkData);
+
+ IoHash ChunkHash = IoHash::HashBuffer(ChunkData);
+ if (auto It = ChunkHashToChunkIndex.find(ChunkHash); It != ChunkHashToChunkIndex.end())
+ {
+ OutChunked.Info.ChunkSequence.push_back(It->second);
+ }
+ else
+ {
+ uint32_t ChunkIndex = gsl::narrow<uint32_t>(OutChunked.Info.ChunkHashes.size());
+ OutChunked.Info.ChunkHashes.push_back(ChunkHash);
+ OutChunked.Info.ChunkSequence.push_back(ChunkIndex);
+ OutChunked.ChunkSources.push_back({.Offset = Offset, .Size = gsl::narrow<uint32_t>(ChunkSize)});
+ }
+ Offset += ChunkSize;
+ BytesProcessed.fetch_add(ChunkSize);
+ }
+ OutChunked.Info.RawSize = RawSize;
+ OutChunked.Info.RawHash = FullHash.GetHash();
+ return true;
+ }
+ else
+ {
+ BasicFile Buffer(InputPath, BasicFile::Mode::kRead);
+ OutChunked = ChunkData(Buffer, 0, RawSize, m_ChunkingParams, &BytesProcessed);
+ return true;
+ }
+ }
+
+ virtual std::string_view GetName() const override { return Name; }
+
+ virtual CbObject GetParameters() const override
+ {
+ CbObjectWriter Writer;
+ Writer.BeginArray("FixedChunkingExtensions");
+ {
+ for (const std::string& Extension : m_FixedChunkingExtensions)
+ {
+ Writer.AddString(Extension);
+ }
+ }
+ Writer.EndArray(); // ChunkExcludeExtensions
+ Writer.AddInteger("ChunkFileSizeLimit"sv, m_ChunkFileSizeLimit);
+ Writer.BeginObject("ChunkingParams"sv);
+ {
+ Writer.AddBool("UseThreshold"sv, m_ChunkingParams.UseThreshold);
+
+ Writer.AddInteger("MinSize"sv, (uint64_t)m_ChunkingParams.MinSize);
+ Writer.AddInteger("MaxSize"sv, (uint64_t)m_ChunkingParams.MaxSize);
+ Writer.AddInteger("AvgSize"sv, (uint64_t)m_ChunkingParams.AvgSize);
+ }
+ Writer.EndObject(); // ChunkingParams
+ Writer.AddInteger("FixedChunkingChunkSize"sv, m_FixedChunkingChunkSize);
+ return Writer.Save();
+ }
+
+ static constexpr std::string_view Name = "ChunkingControllerWithFixedChunking"sv;
+
+protected:
+ const std::vector<std::string> m_FixedChunkingExtensions;
+ const uint64_t m_ChunkFileSizeLimit;
+ const ChunkedParams m_ChunkingParams;
+ const uint32_t m_FixedChunkingChunkSize;
+};
+
+std::unique_ptr<ChunkingController>
+CreateBasicChunkingController(std::span<const std::string_view> ExcludeExtensions,
+ uint64_t ChunkFileSizeLimit,
+ const ChunkedParams& ChunkingParams)
+{
+ return std::make_unique<BasicChunkingController>(ExcludeExtensions, ChunkFileSizeLimit, ChunkingParams);
+}
+std::unique_ptr<ChunkingController>
+CreateBasicChunkingController(CbObjectView Parameters)
+{
+ return std::make_unique<BasicChunkingController>(Parameters);
+}
+
+std::unique_ptr<ChunkingController>
+CreateChunkingControllerWithFixedChunking(std::span<const std::string_view> FixedChunkingExtensions,
+ uint64_t ChunkFileSizeLimit,
+ const ChunkedParams& ChunkingParams,
+ uint32_t FixedChunkingChunkSize)
+{
+ return std::make_unique<ChunkingControllerWithFixedChunking>(FixedChunkingExtensions,
+ ChunkFileSizeLimit,
+ ChunkingParams,
+ FixedChunkingChunkSize);
+}
+std::unique_ptr<ChunkingController>
+CreateChunkingControllerWithFixedChunking(CbObjectView Parameters)
+{
+ return std::make_unique<ChunkingControllerWithFixedChunking>(Parameters);
+}
+
+std::unique_ptr<ChunkingController>
+CreateChunkingController(std::string_view Name, CbObjectView Parameters)
+{
+ if (Name == BasicChunkingController::Name)
+ {
+ return CreateBasicChunkingController(Parameters);
+ }
+ else if (Name == ChunkingControllerWithFixedChunking::Name)
+ {
+ return CreateChunkingControllerWithFixedChunking(Parameters);
+ }
+ return {};
+}
+
+} // namespace zen
diff --git a/src/zenutil/filebuildstorage.cpp b/src/zenutil/filebuildstorage.cpp
new file mode 100644
index 000000000..78ebcdd55
--- /dev/null
+++ b/src/zenutil/filebuildstorage.cpp
@@ -0,0 +1,616 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/filebuildstorage.h>
+
+#include <zencore/basicfile.h>
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinaryvalidation.h>
+#include <zencore/fmtutils.h>
+#include <zencore/scopeguard.h>
+#include <zencore/timer.h>
+
+namespace zen {
+
+using namespace std::literals;
+
+class FileBuildStorage : public BuildStorage
+{
+public:
+ explicit FileBuildStorage(const std::filesystem::path& StoragePath,
+ BuildStorage::Statistics& Stats,
+ bool EnableJsonOutput,
+ double LatencySec,
+ double DelayPerKBSec)
+ : m_StoragePath(StoragePath)
+ , m_Stats(Stats)
+ , m_EnableJsonOutput(EnableJsonOutput)
+ , m_LatencySec(LatencySec)
+ , m_DelayPerKBSec(DelayPerKBSec)
+ {
+ CreateDirectories(GetBuildsFolder());
+ CreateDirectories(GetBlobsFolder());
+ CreateDirectories(GetBlobsMetadataFolder());
+ }
+
+ virtual ~FileBuildStorage() {}
+
+ virtual CbObject ListBuilds(CbObject Query) override
+ {
+ ZEN_UNUSED(Query);
+
+ SimulateLatency(Query.GetSize(), 0);
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BuildFolder = GetBuildsFolder();
+ DirectoryContent Content;
+ GetDirectoryContent(BuildFolder, DirectoryContentFlags::IncludeDirs, Content);
+ CbObjectWriter Writer;
+ Writer.BeginArray("results");
+ {
+ for (const std::filesystem::path& BuildPath : Content.Directories)
+ {
+ Oid BuildId = Oid::TryFromHexString(BuildPath.stem().string());
+ if (BuildId != Oid::Zero)
+ {
+ Writer.BeginObject();
+ {
+ Writer.AddObjectId("buildId", BuildId);
+ Writer.AddObject("metadata", ReadBuild(BuildId)["metadata"sv].AsObjectView());
+ }
+ Writer.EndObject();
+ }
+ }
+ }
+ Writer.EndArray(); // builds
+ Writer.Save();
+ SimulateLatency(Writer.GetSaveSize(), 0);
+ return Writer.Save();
+ }
+
+ virtual CbObject PutBuild(const Oid& BuildId, const CbObject& MetaData) override
+ {
+ SimulateLatency(MetaData.GetSize(), 0);
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ CbObjectWriter BuildObject;
+ BuildObject.AddObject("metadata", MetaData);
+ BuildObject.AddInteger("chunkSize"sv, 32u * 1024u * 1024u);
+ WriteBuild(BuildId, BuildObject.Save());
+
+ CbObjectWriter BuildResponse;
+ BuildResponse.AddInteger("chunkSize"sv, 32u * 1024u * 1024u);
+ BuildResponse.Save();
+
+ SimulateLatency(0, BuildResponse.GetSaveSize());
+ return BuildResponse.Save();
+ }
+
+ virtual CbObject GetBuild(const Oid& BuildId) override
+ {
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ CbObject Build = ReadBuild(BuildId);
+ SimulateLatency(0, Build.GetSize());
+ return Build;
+ }
+
+ virtual void FinalizeBuild(const Oid& BuildId) override
+ {
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ ZEN_UNUSED(BuildId);
+ SimulateLatency(0, 0);
+ }
+
+ virtual std::pair<IoHash, std::vector<IoHash>> PutBuildPart(const Oid& BuildId,
+ const Oid& BuildPartId,
+ std::string_view PartName,
+ const CbObject& MetaData) override
+ {
+ SimulateLatency(MetaData.GetSize(), 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BuildPartDataPath = GetBuildPartPath(BuildId, BuildPartId);
+ CreateDirectories(BuildPartDataPath.parent_path());
+
+ TemporaryFile::SafeWriteFile(BuildPartDataPath, MetaData.GetView());
+ m_WrittenBytes += MetaData.GetSize();
+ WriteAsJson(BuildPartDataPath, MetaData);
+
+ IoHash RawHash = IoHash::HashBuffer(MetaData.GetView());
+
+ CbObjectWriter Writer;
+ {
+ CbObject BuildObject = ReadBuild(BuildId);
+ CbObjectView PartsObject = BuildObject["parts"sv].AsObjectView();
+ CbObjectView MetaDataView = BuildObject["metadata"sv].AsObjectView();
+
+ Writer.AddObject("metadata"sv, MetaDataView);
+ Writer.BeginObject("parts"sv);
+ {
+ for (CbFieldView PartView : PartsObject)
+ {
+ if (PartView.GetName() != PartName)
+ {
+ Writer.AddObjectId(PartView.GetName(), PartView.AsObjectId());
+ }
+ }
+ Writer.AddObjectId(PartName, BuildPartId);
+ }
+ Writer.EndObject(); // parts
+ }
+ WriteBuild(BuildId, Writer.Save());
+
+ std::vector<IoHash> NeededAttachments = GetNeededAttachments(MetaData);
+
+ SimulateLatency(0, sizeof(IoHash) * NeededAttachments.size());
+
+ return std::make_pair(RawHash, std::move(NeededAttachments));
+ }
+
+ virtual CbObject GetBuildPart(const Oid& BuildId, const Oid& BuildPartId) override
+ {
+ SimulateLatency(0, 0);
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BuildPartDataPath = GetBuildPartPath(BuildId, BuildPartId);
+
+ IoBuffer Payload = ReadFile(BuildPartDataPath).Flatten();
+ m_Stats.TotalBytesRead += Payload.GetSize();
+
+ ZEN_ASSERT(ValidateCompactBinary(Payload.GetView(), CbValidateMode::Default) == CbValidateError::None);
+
+ CbObject BuildPartObject = CbObject(SharedBuffer(Payload));
+
+ SimulateLatency(0, BuildPartObject.GetSize());
+
+ return BuildPartObject;
+ }
+
+ virtual std::vector<IoHash> FinalizeBuildPart(const Oid& BuildId, const Oid& BuildPartId, const IoHash& PartHash) override
+ {
+ SimulateLatency(0, 0);
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BuildPartDataPath = GetBuildPartPath(BuildId, BuildPartId);
+ IoBuffer Payload = ReadFile(BuildPartDataPath).Flatten();
+ m_Stats.TotalBytesRead += Payload.GetSize();
+ IoHash RawHash = IoHash::HashBuffer(Payload.GetView());
+ if (RawHash != PartHash)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed finalizing build part {}: Expected hash {}, got {}", BuildPartId, PartHash, RawHash));
+ }
+
+ CbObject BuildPartObject = CbObject(SharedBuffer(Payload));
+ std::vector<IoHash> NeededAttachments(GetNeededAttachments(BuildPartObject));
+
+ SimulateLatency(0, NeededAttachments.size() * sizeof(IoHash));
+
+ return NeededAttachments;
+ }
+
+ virtual void PutBuildBlob(const Oid& BuildId,
+ const IoHash& RawHash,
+ ZenContentType ContentType,
+ const CompositeBuffer& Payload) override
+ {
+ ZEN_UNUSED(BuildId);
+ ZEN_ASSERT(ContentType == ZenContentType::kCompressedBinary);
+ SimulateLatency(Payload.GetSize(), 0);
+
+ ZEN_ASSERT_SLOW(ValidateCompressedBuffer(RawHash, Payload));
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BlockPath = GetBlobPayloadPath(RawHash);
+ if (!std::filesystem::is_regular_file(BlockPath))
+ {
+ CreateDirectories(BlockPath.parent_path());
+ TemporaryFile::SafeWriteFile(BlockPath, Payload.Flatten().GetView());
+ }
+ m_Stats.TotalBytesWritten += Payload.GetSize();
+ SimulateLatency(0, 0);
+ }
+
+ virtual std::vector<std::function<void()>> PutLargeBuildBlob(const Oid& BuildId,
+ const IoHash& RawHash,
+ ZenContentType ContentType,
+ uint64_t PayloadSize,
+ std::function<IoBuffer(uint64_t Offset, uint64_t Size)>&& Transmitter,
+ std::function<void(uint64_t, bool)>&& OnSentBytes) override
+ {
+ ZEN_UNUSED(BuildId);
+ ZEN_UNUSED(ContentType);
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BlockPath = GetBlobPayloadPath(RawHash);
+ if (!std::filesystem::is_regular_file(BlockPath))
+ {
+ CreateDirectories(BlockPath.parent_path());
+
+ struct WorkloadData
+ {
+ std::function<IoBuffer(uint64_t Offset, uint64_t Size)> Transmitter;
+ std::function<void(uint64_t, bool)> OnSentBytes;
+ TemporaryFile TempFile;
+ std::atomic<size_t> PartsLeft;
+ };
+
+ std::shared_ptr<WorkloadData> Workload(std::make_shared<WorkloadData>());
+ Workload->Transmitter = std::move(Transmitter);
+ Workload->OnSentBytes = std::move(OnSentBytes);
+ std::error_code Ec;
+ Workload->TempFile.CreateTemporary(BlockPath.parent_path(), Ec);
+
+ if (Ec)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed opening temporary file '{}': {} ({})", Workload->TempFile.GetPath(), Ec.message(), Ec.value()));
+ }
+
+ std::vector<std::function<void()>> WorkItems;
+ uint64_t Offset = 0;
+ while (Offset < PayloadSize)
+ {
+ uint64_t Size = Min(32u * 1024u * 1024u, PayloadSize - Offset);
+
+ WorkItems.push_back([this, RawHash, BlockPath, Workload, Offset, Size]() {
+ IoBuffer PartPayload = Workload->Transmitter(Offset, Size);
+ SimulateLatency(PartPayload.GetSize(), 0);
+
+ std::error_code Ec;
+ Workload->TempFile.Write(PartPayload, Offset, Ec);
+ if (Ec)
+ {
+ throw std::runtime_error(fmt::format("Failed writing to temporary file '{}': {} ({})",
+ Workload->TempFile.GetPath(),
+ Ec.message(),
+ Ec.value()));
+ }
+ uint64_t BytesWritten = PartPayload.GetSize();
+ m_Stats.TotalBytesWritten += BytesWritten;
+ const bool IsLastPart = Workload->PartsLeft.fetch_sub(1) == 1;
+ if (IsLastPart)
+ {
+ Workload->TempFile.Flush();
+ ZEN_ASSERT_SLOW(ValidateCompressedBuffer(RawHash, CompositeBuffer(Workload->TempFile.ReadAll())));
+ Workload->TempFile.MoveTemporaryIntoPlace(BlockPath, Ec);
+ if (Ec)
+ {
+ throw std::runtime_error(fmt::format("Failed moving temporary file '{}' to '{}': {} ({})",
+ Workload->TempFile.GetPath(),
+ BlockPath,
+ Ec.message(),
+ Ec.value()));
+ }
+ }
+ Workload->OnSentBytes(BytesWritten, IsLastPart);
+ SimulateLatency(0, 0);
+ });
+
+ Offset += Size;
+ }
+ Workload->PartsLeft.store(WorkItems.size());
+
+ SimulateLatency(0, 0);
+ return WorkItems;
+ }
+ SimulateLatency(0, 0);
+ return {};
+ }
+
+ virtual IoBuffer GetBuildBlob(const Oid& BuildId, const IoHash& RawHash) override
+ {
+ ZEN_UNUSED(BuildId);
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BlockPath = GetBlobPayloadPath(RawHash);
+ if (std::filesystem::is_regular_file(BlockPath))
+ {
+ IoBuffer Payload = ReadFile(BlockPath).Flatten();
+ ZEN_ASSERT_SLOW(ValidateCompressedBuffer(RawHash, CompositeBuffer(SharedBuffer(Payload))));
+ m_Stats.TotalBytesRead += Payload.GetSize();
+ Payload.SetContentType(ZenContentType::kCompressedBinary);
+ SimulateLatency(0, Payload.GetSize());
+ return Payload;
+ }
+ SimulateLatency(0, 0);
+ return IoBuffer{};
+ }
+
+ virtual std::vector<std::function<void()>> GetLargeBuildBlob(
+ const Oid& BuildId,
+ const IoHash& RawHash,
+ uint64_t ChunkSize,
+ std::function<void(uint64_t Offset, const IoBuffer& Chunk, uint64_t BytesRemaining)>&& Receiver) override
+ {
+ ZEN_UNUSED(BuildId);
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BlockPath = GetBlobPayloadPath(RawHash);
+ if (std::filesystem::is_regular_file(BlockPath))
+ {
+ struct WorkloadData
+ {
+ std::atomic<uint64_t> BytesRemaining;
+ IoBuffer BlobFile;
+ std::function<void(uint64_t Offset, const IoBuffer& Chunk, uint64_t BytesRemaining)> Receiver;
+ };
+
+ std::shared_ptr<WorkloadData> Workload(std::make_shared<WorkloadData>());
+ Workload->BlobFile = IoBufferBuilder::MakeFromFile(BlockPath);
+ const uint64_t BlobSize = Workload->BlobFile.GetSize();
+
+ Workload->Receiver = std::move(Receiver);
+ Workload->BytesRemaining = BlobSize;
+
+ std::vector<std::function<void()>> WorkItems;
+ uint64_t Offset = 0;
+ while (Offset < BlobSize)
+ {
+ uint64_t Size = Min(ChunkSize, BlobSize - Offset);
+ WorkItems.push_back([this, BlockPath, Workload, Offset, Size]() {
+ SimulateLatency(0, 0);
+ IoBuffer PartPayload(Workload->BlobFile, Offset, Size);
+ m_Stats.TotalBytesRead += PartPayload.GetSize();
+ uint64_t ByteRemaning = Workload->BytesRemaining.fetch_sub(Size);
+ Workload->Receiver(Offset, PartPayload, ByteRemaning);
+ SimulateLatency(Size, PartPayload.GetSize());
+ });
+
+ Offset += Size;
+ }
+ SimulateLatency(0, 0);
+ return WorkItems;
+ }
+ return {};
+ }
+
+ virtual void PutBlockMetadata(const Oid& BuildId, const IoHash& BlockRawHash, const CbObject& MetaData) override
+ {
+ ZEN_UNUSED(BuildId);
+
+ SimulateLatency(MetaData.GetSize(), 0);
+
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ const std::filesystem::path BlockMetaDataPath = GetBlobMetadataPath(BlockRawHash);
+ CreateDirectories(BlockMetaDataPath.parent_path());
+ TemporaryFile::SafeWriteFile(BlockMetaDataPath, MetaData.GetView());
+ m_Stats.TotalBytesWritten += MetaData.GetSize();
+ WriteAsJson(BlockMetaDataPath, MetaData);
+ SimulateLatency(0, 0);
+ }
+
+ virtual std::vector<ChunkBlockDescription> FindBlocks(const Oid& BuildId) override
+ {
+ ZEN_UNUSED(BuildId);
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ DirectoryContent Content;
+ GetDirectoryContent(GetBlobsMetadataFolder(), DirectoryContentFlags::IncludeFiles, Content);
+ std::vector<ChunkBlockDescription> Result;
+ for (const std::filesystem::path& MetaDataFile : Content.Files)
+ {
+ IoHash ChunkHash;
+ if (IoHash::TryParse(MetaDataFile.stem().string(), ChunkHash))
+ {
+ std::filesystem::path BlockPath = GetBlobPayloadPath(ChunkHash);
+ if (std::filesystem::is_regular_file(BlockPath))
+ {
+ IoBuffer BlockMetaDataPayload = ReadFile(MetaDataFile).Flatten();
+
+ m_Stats.TotalBytesRead += BlockMetaDataPayload.GetSize();
+
+ CbObject BlockObject = CbObject(SharedBuffer(BlockMetaDataPayload));
+ Result.emplace_back(ParseChunkBlockDescription(BlockObject));
+ }
+ }
+ }
+ SimulateLatency(0, sizeof(IoHash) * Result.size());
+ return Result;
+ }
+
+ virtual std::vector<ChunkBlockDescription> GetBlockMetadata(const Oid& BuildId, std::span<const IoHash> BlockHashes) override
+ {
+ ZEN_UNUSED(BuildId);
+ SimulateLatency(0, 0);
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ m_Stats.TotalRequestCount++;
+
+ std::vector<ChunkBlockDescription> Result;
+ for (const IoHash& BlockHash : BlockHashes)
+ {
+ std::filesystem::path MetaDataFile = GetBlobMetadataPath(BlockHash);
+ if (std::filesystem::is_regular_file(MetaDataFile))
+ {
+ IoBuffer BlockMetaDataPayload = ReadFile(MetaDataFile).Flatten();
+
+ m_Stats.TotalBytesRead += BlockMetaDataPayload.GetSize();
+
+ CbObject BlockObject = CbObject(SharedBuffer(BlockMetaDataPayload));
+ Result.emplace_back(ParseChunkBlockDescription(BlockObject));
+ }
+ }
+ SimulateLatency(sizeof(BlockHashes) * BlockHashes.size(), sizeof(ChunkBlockDescription) * Result.size());
+ return Result;
+ }
+
+protected:
+ std::filesystem::path GetBuildsFolder() const { return m_StoragePath / "builds"; }
+ std::filesystem::path GetBlobsFolder() const { return m_StoragePath / "blobs"; }
+ std::filesystem::path GetBlobsMetadataFolder() const { return m_StoragePath / "blocks"; }
+ std::filesystem::path GetBuildFolder(const Oid& BuildId) const { return GetBuildsFolder() / BuildId.ToString(); }
+
+ std::filesystem::path GetBuildPath(const Oid& BuildId) const { return GetBuildFolder(BuildId) / "metadata.cb"; }
+
+ std::filesystem::path GetBuildPartFolder(const Oid& BuildId, const Oid& BuildPartId) const
+ {
+ return GetBuildFolder(BuildId) / "parts" / BuildPartId.ToString();
+ }
+
+ std::filesystem::path GetBuildPartPath(const Oid& BuildId, const Oid& BuildPartId) const
+ {
+ return GetBuildPartFolder(BuildId, BuildPartId) / "metadata.cb";
+ }
+
+ std::filesystem::path GetBlobPayloadPath(const IoHash& RawHash) const { return GetBlobsFolder() / fmt::format("{}.cbz", RawHash); }
+
+ std::filesystem::path GetBlobMetadataPath(const IoHash& RawHash) const
+ {
+ return GetBlobsMetadataFolder() / fmt::format("{}.cb", RawHash);
+ }
+
+ void SimulateLatency(uint64_t ReceiveSize, uint64_t SendSize)
+ {
+ double SleepSec = m_LatencySec;
+ if (m_DelayPerKBSec > 0.0)
+ {
+ SleepSec += m_DelayPerKBSec * (double(SendSize + ReceiveSize) / 1024u);
+ }
+ if (SleepSec > 0)
+ {
+ Sleep(int(SleepSec * 1000));
+ }
+ }
+
+ void WriteAsJson(const std::filesystem::path& OriginalPath, CbObjectView Data) const
+ {
+ if (m_EnableJsonOutput)
+ {
+ ExtendableStringBuilder<128> SB;
+ CompactBinaryToJson(Data, SB);
+ std::filesystem::path JsonPath = OriginalPath;
+ JsonPath.replace_extension(".json");
+ std::string_view JsonMetaData = SB.ToView();
+ TemporaryFile::SafeWriteFile(JsonPath, MemoryView(JsonMetaData.data(), JsonMetaData.length()));
+ }
+ }
+
+ void WriteBuild(const Oid& BuildId, CbObjectView Data)
+ {
+ const std::filesystem::path BuildDataPath = GetBuildPath(BuildId);
+ CreateDirectories(BuildDataPath.parent_path());
+ TemporaryFile::SafeWriteFile(BuildDataPath, Data.GetView());
+ m_Stats.TotalBytesWritten += Data.GetSize();
+ WriteAsJson(BuildDataPath, Data);
+ }
+
+ CbObject ReadBuild(const Oid& BuildId)
+ {
+ const std::filesystem::path BuildDataPath = GetBuildPath(BuildId);
+ FileContents Content = ReadFile(BuildDataPath);
+ if (Content.ErrorCode)
+ {
+ throw std::runtime_error(fmt::format("Failed reading build '{}' from '{}': {} ({})",
+ BuildId,
+ BuildDataPath,
+ Content.ErrorCode.message(),
+ Content.ErrorCode.value()));
+ }
+ IoBuffer Payload = Content.Flatten();
+ m_Stats.TotalBytesRead += Payload.GetSize();
+ ZEN_ASSERT(ValidateCompactBinary(Payload.GetView(), CbValidateMode::Default) == CbValidateError::None);
+ CbObject BuildObject = CbObject(SharedBuffer(Payload));
+ return BuildObject;
+ }
+
+ std::vector<IoHash> GetNeededAttachments(CbObjectView BuildPartObject)
+ {
+ std::vector<IoHash> NeededAttachments;
+ BuildPartObject.IterateAttachments([&](CbFieldView FieldView) {
+ const IoHash AttachmentHash = FieldView.AsBinaryAttachment();
+ const std::filesystem::path BlockPath = GetBlobPayloadPath(AttachmentHash);
+ if (!std::filesystem::is_regular_file(BlockPath))
+ {
+ NeededAttachments.push_back(AttachmentHash);
+ }
+ });
+ return NeededAttachments;
+ }
+
+ bool ValidateCompressedBuffer(const IoHash& RawHash, const CompositeBuffer& Payload)
+ {
+ IoHash VerifyHash;
+ uint64_t VerifySize;
+ CompressedBuffer ValidateBuffer = CompressedBuffer::FromCompressed(Payload, VerifyHash, VerifySize);
+ if (!ValidateBuffer)
+ {
+ return false;
+ }
+ if (VerifyHash != RawHash)
+ {
+ return false;
+ }
+ CompositeBuffer Decompressed = ValidateBuffer.DecompressToComposite();
+ if (!Decompressed)
+ {
+ return false;
+ }
+ IoHash Hash = IoHash::HashBuffer(Decompressed);
+ if (Hash != RawHash)
+ {
+ return false;
+ }
+ return true;
+ }
+
+private:
+ const std::filesystem::path m_StoragePath;
+ BuildStorage::Statistics& m_Stats;
+ const bool m_EnableJsonOutput = false;
+ std::atomic<uint64_t> m_WrittenBytes;
+
+ const double m_LatencySec = 0.0;
+ const double m_DelayPerKBSec = 0.0;
+};
+
+std::unique_ptr<BuildStorage>
+CreateFileBuildStorage(const std::filesystem::path& StoragePath,
+ BuildStorage::Statistics& Stats,
+ bool EnableJsonOutput,
+ double LatencySec,
+ double DelayPerKBSec)
+{
+ return std::make_unique<FileBuildStorage>(StoragePath, Stats, EnableJsonOutput, LatencySec, DelayPerKBSec);
+}
+
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/buildstorage.h b/src/zenutil/include/zenutil/buildstorage.h
new file mode 100644
index 000000000..9c236310f
--- /dev/null
+++ b/src/zenutil/include/zenutil/buildstorage.h
@@ -0,0 +1,55 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinary.h>
+#include <zenutil/chunkblock.h>
+
+namespace zen {
+
+class BuildStorage
+{
+public:
+ struct Statistics
+ {
+ std::atomic<uint64_t> TotalBytesRead = 0;
+ std::atomic<uint64_t> TotalBytesWritten = 0;
+ std::atomic<uint64_t> TotalRequestCount = 0;
+ std::atomic<uint64_t> TotalRequestTimeUs = 0;
+ std::atomic<uint64_t> TotalExecutionTimeUs = 0;
+ };
+
+ virtual ~BuildStorage() {}
+
+ virtual CbObject ListBuilds(CbObject Query) = 0;
+ virtual CbObject PutBuild(const Oid& BuildId, const CbObject& MetaData) = 0;
+ virtual CbObject GetBuild(const Oid& BuildId) = 0;
+ virtual void FinalizeBuild(const Oid& BuildId) = 0;
+
+ virtual std::pair<IoHash, std::vector<IoHash>> PutBuildPart(const Oid& BuildId,
+ const Oid& BuildPartId,
+ std::string_view PartName,
+ const CbObject& MetaData) = 0;
+ virtual CbObject GetBuildPart(const Oid& BuildId, const Oid& BuildPartId) = 0;
+ virtual std::vector<IoHash> FinalizeBuildPart(const Oid& BuildId, const Oid& BuildPartId, const IoHash& PartHash) = 0;
+ virtual void PutBuildBlob(const Oid& BuildId, const IoHash& RawHash, ZenContentType ContentType, const CompositeBuffer& Payload) = 0;
+ virtual std::vector<std::function<void()>> PutLargeBuildBlob(const Oid& BuildId,
+ const IoHash& RawHash,
+ ZenContentType ContentType,
+ uint64_t PayloadSize,
+ std::function<IoBuffer(uint64_t Offset, uint64_t Size)>&& Transmitter,
+ std::function<void(uint64_t, bool)>&& OnSentBytes) = 0;
+
+ virtual IoBuffer GetBuildBlob(const Oid& BuildId, const IoHash& RawHash) = 0;
+ virtual std::vector<std::function<void()>> GetLargeBuildBlob(
+ const Oid& BuildId,
+ const IoHash& RawHash,
+ uint64_t ChunkSize,
+ std::function<void(uint64_t Offset, const IoBuffer& Chunk, uint64_t BytesRemaining)>&& Receiver) = 0;
+
+ virtual void PutBlockMetadata(const Oid& BuildId, const IoHash& BlockRawHash, const CbObject& MetaData) = 0;
+ virtual std::vector<ChunkBlockDescription> FindBlocks(const Oid& BuildId) = 0;
+ virtual std::vector<ChunkBlockDescription> GetBlockMetadata(const Oid& BuildId, std::span<const IoHash> BlockHashes) = 0;
+};
+
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/chunkblock.h b/src/zenutil/include/zenutil/chunkblock.h
index 9b7414629..21107fb7c 100644
--- a/src/zenutil/include/zenutil/chunkblock.h
+++ b/src/zenutil/include/zenutil/chunkblock.h
@@ -12,21 +12,28 @@
namespace zen {
-struct ChunkBlockDescription
+struct ThinChunkBlockDescription
{
- IoHash BlockHash;
- std::vector<IoHash> ChunkHashes;
+ IoHash BlockHash;
+ std::vector<IoHash> ChunkRawHashes;
+};
+
+struct ChunkBlockDescription : public ThinChunkBlockDescription
+{
+ uint64_t HeaderSize;
std::vector<uint32_t> ChunkRawLengths;
+ std::vector<uint32_t> ChunkCompressedLengths;
};
std::vector<ChunkBlockDescription> ParseChunkBlockDescriptionList(const CbObjectView& BlocksObject);
ChunkBlockDescription ParseChunkBlockDescription(const CbObjectView& BlockObject);
CbObject BuildChunkBlockDescription(const ChunkBlockDescription& Block, CbObjectView MetaData);
-
+ChunkBlockDescription GetChunkBlockDescription(const SharedBuffer& BlockPayload, const IoHash& RawHash);
typedef std::function<std::pair<uint64_t, CompressedBuffer>(const IoHash& RawHash)> FetchChunkFunc;
CompressedBuffer GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks, ChunkBlockDescription& OutBlock);
bool IterateChunkBlock(const SharedBuffer& BlockPayload,
- std::function<void(CompressedBuffer&& Chunk, const IoHash& AttachmentHash)> Visitor);
+ std::function<void(CompressedBuffer&& Chunk, const IoHash& AttachmentHash)> Visitor,
+ uint64_t& OutHeaderSize);
} // namespace zen
diff --git a/src/zenutil/include/zenutil/chunkedcontent.h b/src/zenutil/include/zenutil/chunkedcontent.h
new file mode 100644
index 000000000..15c687462
--- /dev/null
+++ b/src/zenutil/include/zenutil/chunkedcontent.h
@@ -0,0 +1,256 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/iohash.h>
+
+#include <filesystem>
+#include <vector>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_map.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+class CbWriter;
+class ChunkingController;
+class WorkerThreadPool;
+
+enum class SourcePlatform
+{
+ Windows = 0,
+ Linux = 1,
+ MacOS = 2,
+ _Count
+};
+
+std::string_view ToString(SourcePlatform Platform);
+SourcePlatform FromString(std::string_view Platform, SourcePlatform Default);
+SourcePlatform GetSourceCurrentPlatform();
+
+struct FolderContent
+{
+ SourcePlatform Platform = GetSourceCurrentPlatform();
+ std::vector<std::filesystem::path> Paths;
+ std::vector<uint64_t> RawSizes;
+ std::vector<uint32_t> Attributes;
+ std::vector<uint64_t> ModificationTicks;
+
+ bool operator==(const FolderContent& Rhs) const;
+
+ bool AreKnownFilesEqual(const FolderContent& Rhs) const;
+ void UpdateState(const FolderContent& Rhs, std::vector<uint32_t>& PathIndexesOufOfDate);
+ static bool AreFileAttributesEqual(const uint32_t Lhs, const uint32_t Rhs);
+};
+
+FolderContent GetUpdatedContent(const FolderContent& Old,
+ const FolderContent& New,
+ std::vector<std::filesystem::path>& OutDeletedPathIndexes);
+
+void SaveFolderContentToCompactBinary(const FolderContent& Content, CbWriter& Output);
+FolderContent LoadFolderContentToCompactBinary(CbObjectView Input);
+
+struct GetFolderContentStatistics
+{
+ std::atomic<uint64_t> FoundFileCount = 0;
+ std::atomic<uint64_t> FoundFileByteCount = 0;
+ std::atomic<uint64_t> AcceptedFileCount = 0;
+ std::atomic<uint64_t> AcceptedFileByteCount = 0;
+ uint64_t ElapsedWallTimeUS = 0;
+};
+
+FolderContent GetFolderContent(GetFolderContentStatistics& Stats,
+ const std::filesystem::path& RootPath,
+ std::function<bool(const std::string_view& RelativePath)>&& AcceptDirectory,
+ std::function<bool(std::string_view RelativePath, uint64_t Size, uint32_t Attributes)>&& AcceptFile,
+ WorkerThreadPool& WorkerPool,
+ int32_t UpdateInteralMS,
+ std::function<void(bool IsAborted, std::ptrdiff_t PendingWork)>&& UpdateCallback,
+ std::atomic<bool>& AbortFlag);
+
+struct ChunkedContentData
+{
+ // To describe one asset with a particular RawHash, find the index of the hash in SequenceRawHashes
+ // ChunkCounts for that index will be the number of indexes in ChunkOrders that describe
+ // the sequence of chunks required to reconstruct the asset.
+ // Offset into ChunkOrders is based on how many entries in ChunkOrders the previous [n - 1] SequenceRawHashes uses
+ std::vector<IoHash> SequenceRawHashes; // Raw hash for Chunk sequence
+ std::vector<uint32_t> ChunkCounts; // Chunk count of ChunkOrder for SequenceRawHashes[n]
+ std::vector<uint32_t> ChunkOrders; // Chunk sequence indexed into ChunkHashes, ChunkCounts[n] indexes per SequenceRawHashes[n]
+ std::vector<IoHash> ChunkHashes; // Unique chunk hashes
+ std::vector<uint64_t> ChunkRawSizes; // Unique chunk raw size for ChunkHash[n]
+};
+
+struct ChunkedFolderContent
+{
+ SourcePlatform Platform = GetSourceCurrentPlatform();
+ std::vector<std::filesystem::path> Paths;
+ std::vector<uint64_t> RawSizes;
+ std::vector<uint32_t> Attributes;
+ std::vector<IoHash> RawHashes;
+ ChunkedContentData ChunkedContent;
+};
+
+void SaveChunkedFolderContentToCompactBinary(const ChunkedFolderContent& Content, CbWriter& Output);
+ChunkedFolderContent LoadChunkedFolderContentToCompactBinary(CbObjectView Input);
+
+ChunkedFolderContent MergeChunkedFolderContents(const ChunkedFolderContent& Base, std::span<const ChunkedFolderContent> Overlays);
+ChunkedFolderContent DeletePathsFromChunkedContent(const ChunkedFolderContent& Base, std::span<const std::filesystem::path> DeletedPaths);
+
+struct ChunkingStatistics
+{
+ std::atomic<uint64_t> FilesProcessed = 0;
+ std::atomic<uint64_t> FilesChunked = 0;
+ std::atomic<uint64_t> BytesHashed = 0;
+ std::atomic<uint64_t> UniqueChunksFound = 0;
+ std::atomic<uint64_t> UniqueSequencesFound = 0;
+ std::atomic<uint64_t> UniqueBytesFound = 0;
+ uint64_t ElapsedWallTimeUS = 0;
+};
+
+ChunkedFolderContent ChunkFolderContent(ChunkingStatistics& Stats,
+ WorkerThreadPool& WorkerPool,
+ const std::filesystem::path& RootPath,
+ const FolderContent& Content,
+ const ChunkingController& InChunkingController,
+ int32_t UpdateInteralMS,
+ std::function<void(bool IsAborted, std::ptrdiff_t PendingWork)>&& UpdateCallback,
+ std::atomic<bool>& AbortFlag);
+
+struct ChunkedContentLookup
+{
+ struct ChunkLocation
+ {
+ uint32_t PathIndex;
+ uint64_t Offset;
+ };
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> ChunkHashToChunkIndex;
+ tsl::robin_map<IoHash, uint32_t, IoHash::Hasher> RawHashToSequenceRawHashIndex;
+ std::vector<uint32_t> SequenceRawHashIndexChunkOrderOffset;
+ std::vector<ChunkLocation> ChunkLocations;
+ std::vector<size_t> ChunkLocationOffset; // ChunkLocations[ChunkLocationOffset[ChunkIndex]] -> start of sources for ChunkIndex
+ std::vector<uint32_t> ChunkLocationCounts; // ChunkLocationCounts[ChunkIndex] count of chunk locations for ChunkIndex
+};
+
+ChunkedContentLookup BuildChunkedContentLookup(const ChunkedFolderContent& Content);
+
+inline std::pair<size_t, uint32_t>
+GetChunkLocationRange(const ChunkedContentLookup& Lookup, uint32_t ChunkIndex)
+{
+ return std::make_pair(Lookup.ChunkLocationOffset[ChunkIndex], Lookup.ChunkLocationCounts[ChunkIndex]);
+}
+
+inline std::span<const ChunkedContentLookup::ChunkLocation>
+GetChunkLocations(const ChunkedContentLookup& Lookup, uint32_t ChunkIndex)
+{
+ std::pair<size_t, uint32_t> Range = GetChunkLocationRange(Lookup, ChunkIndex);
+ return std::span<const ChunkedContentLookup::ChunkLocation>(Lookup.ChunkLocations).subspan(Range.first, Range.second);
+}
+
+namespace compactbinary_helpers {
+ template<typename Type>
+ void WriteArray(std::span<const Type> Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ Output.BeginArray(ArrayName);
+ for (const Type Value : Values)
+ {
+ Output << Value;
+ }
+ Output.EndArray();
+ }
+
+ template<typename Type>
+ void WriteArray(const std::vector<Type>& Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ WriteArray(std::span<const Type>(Values), ArrayName, Output);
+ }
+
+ template<>
+ inline void WriteArray(std::span<const std::filesystem::path> Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ Output.BeginArray(ArrayName);
+ for (const std::filesystem::path& Path : Values)
+ {
+ Output.AddString((const char*)Path.generic_u8string().c_str());
+ }
+ Output.EndArray();
+ }
+
+ template<>
+ inline void WriteArray(const std::vector<std::filesystem::path>& Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ WriteArray(std::span<const std::filesystem::path>(Values), ArrayName, Output);
+ }
+
+ inline void WriteBinaryAttachmentArray(std::span<const IoHash> Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ Output.BeginArray(ArrayName);
+ for (const IoHash& Hash : Values)
+ {
+ Output.AddBinaryAttachment(Hash);
+ }
+ Output.EndArray();
+ }
+
+ inline void WriteBinaryAttachmentArray(const std::vector<IoHash>& Values, std::string_view ArrayName, CbWriter& Output)
+ {
+ WriteArray(std::span<const IoHash>(Values), ArrayName, Output);
+ }
+
+ inline void ReadArray(std::string_view ArrayName, CbObjectView Input, std::vector<uint32_t>& Result)
+ {
+ CbArrayView Array = Input[ArrayName].AsArrayView();
+ Result.reserve(Array.Num());
+ for (CbFieldView ItemView : Array)
+ {
+ Result.push_back(ItemView.AsUInt32());
+ }
+ }
+
+ inline void ReadArray(std::string_view ArrayName, CbObjectView Input, std::vector<uint64_t>& Result)
+ {
+ CbArrayView Array = Input[ArrayName].AsArrayView();
+ Result.reserve(Array.Num());
+ for (CbFieldView ItemView : Array)
+ {
+ Result.push_back(ItemView.AsUInt64());
+ }
+ }
+
+ inline void ReadArray(std::string_view ArrayName, CbObjectView Input, std::vector<std::filesystem::path>& Result)
+ {
+ CbArrayView Array = Input[ArrayName].AsArrayView();
+ Result.reserve(Array.Num());
+ for (CbFieldView ItemView : Array)
+ {
+ std::u8string_view U8Path = ItemView.AsU8String();
+ Result.push_back(std::filesystem::path(U8Path));
+ }
+ }
+
+ inline void ReadArray(std::string_view ArrayName, CbObjectView Input, std::vector<IoHash>& Result)
+ {
+ CbArrayView Array = Input[ArrayName].AsArrayView();
+ Result.reserve(Array.Num());
+ for (CbFieldView ItemView : Array)
+ {
+ Result.push_back(ItemView.AsHash());
+ }
+ }
+
+ inline void ReadBinaryAttachmentArray(std::string_view ArrayName, CbObjectView Input, std::vector<IoHash>& Result)
+ {
+ CbArrayView Array = Input[ArrayName].AsArrayView();
+ Result.reserve(Array.Num());
+ for (CbFieldView ItemView : Array)
+ {
+ Result.push_back(ItemView.AsBinaryAttachment());
+ }
+ }
+
+} // namespace compactbinary_helpers
+
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/chunkingcontroller.h b/src/zenutil/include/zenutil/chunkingcontroller.h
new file mode 100644
index 000000000..fe4fc1bb5
--- /dev/null
+++ b/src/zenutil/include/zenutil/chunkingcontroller.h
@@ -0,0 +1,55 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinary.h>
+
+#include <zenutil/chunkedfile.h>
+
+#include <atomic>
+#include <filesystem>
+
+namespace zen {
+
+const std::vector<std::string_view> DefaultChunkingExcludeExtensions = {".exe", ".dll", ".pdb", ".self"};
+
+const ChunkedParams DefaultChunkedParams = {.MinSize = ((8u * 1u) * 1024u) - 128u,
+ .MaxSize = 128u * 1024u,
+ .AvgSize = ((8u * 4u) * 1024u) + 128u};
+
+const size_t DefaultChunkingFileSizeLimit = DefaultChunkedParams.MaxSize;
+
+const uint32_t DefaultFixedChunkingChunkSize = 16u * 1024u * 1024u;
+
+struct ChunkedInfoWithSource;
+
+class ChunkingController
+{
+public:
+ virtual ~ChunkingController() {}
+
+ // Return true if the input file was processed. If true is returned OutChunked will contain the chunked info
+ virtual bool ProcessFile(const std::filesystem::path& InputPath,
+ uint64_t RawSize,
+ ChunkedInfoWithSource& OutChunked,
+ std::atomic<uint64_t>& BytesProcessed) const = 0;
+ virtual std::string_view GetName() const = 0;
+ virtual CbObject GetParameters() const = 0;
+};
+
+std::unique_ptr<ChunkingController> CreateBasicChunkingController(
+ std::span<const std::string_view> ExcludeExtensions = DefaultChunkingExcludeExtensions,
+ uint64_t ChunkFileSizeLimit = DefaultChunkingFileSizeLimit,
+ const ChunkedParams& ChunkingParams = DefaultChunkedParams);
+std::unique_ptr<ChunkingController> CreateBasicChunkingController(CbObjectView Parameters);
+
+std::unique_ptr<ChunkingController> CreateChunkingControllerWithFixedChunking(
+ std::span<const std::string_view> ExcludeExtensions = DefaultChunkingExcludeExtensions,
+ uint64_t ChunkFileSizeLimit = DefaultChunkingFileSizeLimit,
+ const ChunkedParams& ChunkingParams = DefaultChunkedParams,
+ uint32_t FixedChunkingChunkSize = DefaultFixedChunkingChunkSize);
+std::unique_ptr<ChunkingController> CreateChunkingControllerWithFixedChunking(CbObjectView Parameters);
+
+std::unique_ptr<ChunkingController> CreateChunkingController(std::string_view Name, CbObjectView Parameters);
+
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/filebuildstorage.h b/src/zenutil/include/zenutil/filebuildstorage.h
new file mode 100644
index 000000000..c95fb32e6
--- /dev/null
+++ b/src/zenutil/include/zenutil/filebuildstorage.h
@@ -0,0 +1,16 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/logging.h>
+#include <zenutil/buildstorage.h>
+
+namespace zen {
+class HttpClient;
+
+std::unique_ptr<BuildStorage> CreateFileBuildStorage(const std::filesystem::path& StoragePath,
+ BuildStorage::Statistics& Stats,
+ bool EnableJsonOutput,
+ double LatencySec = 0.0,
+ double DelayPerKBSec = 0.0);
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/jupiter/jupiterbuildstorage.h b/src/zenutil/include/zenutil/jupiter/jupiterbuildstorage.h
new file mode 100644
index 000000000..89fc70140
--- /dev/null
+++ b/src/zenutil/include/zenutil/jupiter/jupiterbuildstorage.h
@@ -0,0 +1,17 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/logging.h>
+#include <zenutil/buildstorage.h>
+
+namespace zen {
+class HttpClient;
+
+std::unique_ptr<BuildStorage> CreateJupiterBuildStorage(LoggerRef InLog,
+ HttpClient& InHttpClient,
+ BuildStorage::Statistics& Stats,
+ std::string_view Namespace,
+ std::string_view Bucket,
+ const std::filesystem::path& TempFolderPath);
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/parallellwork.h b/src/zenutil/include/zenutil/parallellwork.h
new file mode 100644
index 000000000..7a8218c51
--- /dev/null
+++ b/src/zenutil/include/zenutil/parallellwork.h
@@ -0,0 +1,69 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/thread.h>
+#include <zencore/workthreadpool.h>
+
+#include <atomic>
+
+namespace zen {
+
+class ParallellWork
+{
+public:
+ ParallellWork(std::atomic<bool>& AbortFlag) : m_AbortFlag(AbortFlag), m_PendingWork(1) {}
+
+ ~ParallellWork()
+ {
+ // Make sure to call Wait before destroying
+ ZEN_ASSERT(m_PendingWork.Remaining() == 0);
+ }
+
+ void ScheduleWork(WorkerThreadPool& WorkerPool,
+ std::function<void(std::atomic<bool>& AbortFlag)>&& Work,
+ std::function<void(const std::exception& Ex, std::atomic<bool>& AbortFlag)>&& OnError)
+ {
+ m_PendingWork.AddCount(1);
+ try
+ {
+ WorkerPool.ScheduleWork([this, Work = std::move(Work), OnError = std::move(OnError)] {
+ try
+ {
+ Work(m_AbortFlag);
+ }
+ catch (const std::exception& Ex)
+ {
+ OnError(Ex, m_AbortFlag);
+ }
+ m_PendingWork.CountDown();
+ });
+ }
+ catch (const std::exception&)
+ {
+ m_PendingWork.CountDown();
+ throw;
+ }
+ }
+
+ void Abort() { m_AbortFlag = true; }
+
+ bool IsAborted() const { return m_AbortFlag.load(); }
+
+ void Wait(int32_t UpdateInteralMS, std::function<void(bool IsAborted, std::ptrdiff_t PendingWork)>&& UpdateCallback)
+ {
+ ZEN_ASSERT(m_PendingWork.Remaining() > 0);
+ m_PendingWork.CountDown();
+ while (!m_PendingWork.Wait(UpdateInteralMS))
+ {
+ UpdateCallback(m_AbortFlag.load(), m_PendingWork.Remaining());
+ }
+ }
+ Latch& PendingWork() { return m_PendingWork; }
+
+private:
+ std::atomic<bool>& m_AbortFlag;
+ Latch m_PendingWork;
+};
+
+} // namespace zen
diff --git a/src/zenutil/jupiter/jupiterbuildstorage.cpp b/src/zenutil/jupiter/jupiterbuildstorage.cpp
new file mode 100644
index 000000000..481e9146f
--- /dev/null
+++ b/src/zenutil/jupiter/jupiterbuildstorage.cpp
@@ -0,0 +1,371 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/jupiter/jupiterbuildstorage.h>
+
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/fmtutils.h>
+#include <zencore/scopeguard.h>
+#include <zencore/timer.h>
+#include <zenutil/jupiter/jupitersession.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_map.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+using namespace std::literals;
+
+class JupiterBuildStorage : public BuildStorage
+{
+public:
+ JupiterBuildStorage(LoggerRef InLog,
+ HttpClient& InHttpClient,
+ Statistics& Stats,
+ std::string_view Namespace,
+ std::string_view Bucket,
+ const std::filesystem::path& TempFolderPath)
+ : m_Session(InLog, InHttpClient)
+ , m_Stats(Stats)
+ , m_Namespace(Namespace)
+ , m_Bucket(Bucket)
+ , m_TempFolderPath(TempFolderPath)
+ {
+ }
+ virtual ~JupiterBuildStorage() {}
+
+ virtual CbObject ListBuilds(CbObject Query) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ IoBuffer Payload = Query.GetBuffer().AsIoBuffer();
+ Payload.SetContentType(ZenContentType::kCbObject);
+ JupiterResult ListResult = m_Session.ListBuilds(m_Namespace, m_Bucket, Payload);
+ AddStatistic(ListResult);
+ if (!ListResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed listing builds: {} ({})", ListResult.Reason, ListResult.ErrorCode));
+ }
+ return PayloadToJson("Failed listing builds"sv, ListResult.Response);
+ }
+
+ virtual CbObject PutBuild(const Oid& BuildId, const CbObject& MetaData) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ IoBuffer Payload = MetaData.GetBuffer().AsIoBuffer();
+ Payload.SetContentType(ZenContentType::kCbObject);
+ JupiterResult PutResult = m_Session.PutBuild(m_Namespace, m_Bucket, BuildId, Payload);
+ AddStatistic(PutResult);
+ if (!PutResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed creating build: {} ({})", PutResult.Reason, PutResult.ErrorCode));
+ }
+ return PayloadToJson(fmt::format("Failed creating build: {}", BuildId), PutResult.Response);
+ }
+
+ virtual CbObject GetBuild(const Oid& BuildId) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult GetBuildResult = m_Session.GetBuild(m_Namespace, m_Bucket, BuildId);
+ AddStatistic(GetBuildResult);
+ if (!GetBuildResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed fetching build: {} ({})", GetBuildResult.Reason, GetBuildResult.ErrorCode));
+ }
+ return PayloadToJson(fmt::format("Failed fetching build {}:", BuildId), GetBuildResult.Response);
+ }
+
+ virtual void FinalizeBuild(const Oid& BuildId) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult FinalizeBuildResult = m_Session.FinalizeBuild(m_Namespace, m_Bucket, BuildId);
+ AddStatistic(FinalizeBuildResult);
+ if (!FinalizeBuildResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed finalizing build part: {} ({})", FinalizeBuildResult.Reason, FinalizeBuildResult.ErrorCode));
+ }
+ }
+
+ virtual std::pair<IoHash, std::vector<IoHash>> PutBuildPart(const Oid& BuildId,
+ const Oid& BuildPartId,
+ std::string_view PartName,
+ const CbObject& MetaData) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ IoBuffer Payload = MetaData.GetBuffer().AsIoBuffer();
+ Payload.SetContentType(ZenContentType::kCbObject);
+ PutBuildPartResult PutPartResult = m_Session.PutBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId, PartName, Payload);
+ AddStatistic(PutPartResult);
+ if (!PutPartResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed creating build part: {} ({})", PutPartResult.Reason, PutPartResult.ErrorCode));
+ }
+ return std::make_pair(PutPartResult.RawHash, std::move(PutPartResult.Needs));
+ }
+
+ virtual CbObject GetBuildPart(const Oid& BuildId, const Oid& BuildPartId) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult GetBuildPartResult = m_Session.GetBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId);
+ AddStatistic(GetBuildPartResult);
+ if (!GetBuildPartResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed fetching build part {}: {} ({})",
+ BuildPartId,
+ GetBuildPartResult.Reason,
+ GetBuildPartResult.ErrorCode));
+ }
+ return PayloadToJson(fmt::format("Failed fetching build part {}:", BuildPartId), GetBuildPartResult.Response);
+ }
+
+ virtual std::vector<IoHash> FinalizeBuildPart(const Oid& BuildId, const Oid& BuildPartId, const IoHash& PartHash) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ FinalizeBuildPartResult FinalizePartResult = m_Session.FinalizeBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId, PartHash);
+ AddStatistic(FinalizePartResult);
+ if (!FinalizePartResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed finalizing build part: {} ({})", FinalizePartResult.Reason, FinalizePartResult.ErrorCode));
+ }
+ return std::move(FinalizePartResult.Needs);
+ }
+
+ virtual void PutBuildBlob(const Oid& BuildId,
+ const IoHash& RawHash,
+ ZenContentType ContentType,
+ const CompositeBuffer& Payload) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult PutBlobResult = m_Session.PutBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, ContentType, Payload);
+ AddStatistic(PutBlobResult);
+ if (!PutBlobResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PutBlobResult.Reason, PutBlobResult.ErrorCode));
+ }
+ }
+
+ virtual std::vector<std::function<void()>> PutLargeBuildBlob(const Oid& BuildId,
+ const IoHash& RawHash,
+ ZenContentType ContentType,
+ uint64_t PayloadSize,
+ std::function<IoBuffer(uint64_t Offset, uint64_t Size)>&& Transmitter,
+ std::function<void(uint64_t, bool)>&& OnSentBytes) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ std::vector<std::function<JupiterResult(bool&)>> WorkItems;
+ JupiterResult PutMultipartBlobResult = m_Session.PutMultipartBuildBlob(m_Namespace,
+ m_Bucket,
+ BuildId,
+ RawHash,
+ ContentType,
+ PayloadSize,
+ std::move(Transmitter),
+ WorkItems);
+ AddStatistic(PutMultipartBlobResult);
+ if (!PutMultipartBlobResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed putting build part: {} ({})", PutMultipartBlobResult.Reason, PutMultipartBlobResult.ErrorCode));
+ }
+ OnSentBytes(PutMultipartBlobResult.SentBytes, WorkItems.empty());
+
+ std::vector<std::function<void()>> WorkList;
+ for (auto& WorkItem : WorkItems)
+ {
+ WorkList.emplace_back([this, WorkItem = std::move(WorkItem), OnSentBytes]() {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ bool IsComplete = false;
+ JupiterResult PartResult = WorkItem(IsComplete);
+ AddStatistic(PartResult);
+ if (!PartResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+ }
+ OnSentBytes(PartResult.SentBytes, IsComplete);
+ });
+ }
+ return WorkList;
+ }
+
+ virtual IoBuffer GetBuildBlob(const Oid& BuildId, const IoHash& RawHash) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult GetBuildBlobResult = m_Session.GetBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, m_TempFolderPath);
+ AddStatistic(GetBuildBlobResult);
+ if (!GetBuildBlobResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed fetching build blob {}: {} ({})", RawHash, GetBuildBlobResult.Reason, GetBuildBlobResult.ErrorCode));
+ }
+ return std::move(GetBuildBlobResult.Response);
+ }
+
+ virtual std::vector<std::function<void()>> GetLargeBuildBlob(
+ const Oid& BuildId,
+ const IoHash& RawHash,
+ uint64_t ChunkSize,
+ std::function<void(uint64_t Offset, const IoBuffer& Chunk, uint64_t BytesRemaining)>&& Receiver) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ std::vector<std::function<JupiterResult()>> WorkItems;
+ JupiterResult GetMultipartBlobResult =
+ m_Session.GetMultipartBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, ChunkSize, std::move(Receiver), WorkItems);
+
+ AddStatistic(GetMultipartBlobResult);
+ if (!GetMultipartBlobResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed getting build part: {} ({})", GetMultipartBlobResult.Reason, GetMultipartBlobResult.ErrorCode));
+ }
+ std::vector<std::function<void()>> WorkList;
+ for (auto& WorkItem : WorkItems)
+ {
+ WorkList.emplace_back([this, WorkItem = std::move(WorkItem)]() {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult PartResult = WorkItem();
+ AddStatistic(PartResult);
+ if (!PartResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed getting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+ }
+ });
+ }
+ return WorkList;
+ }
+
+ virtual void PutBlockMetadata(const Oid& BuildId, const IoHash& BlockRawHash, const CbObject& MetaData) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ IoBuffer Payload = MetaData.GetBuffer().AsIoBuffer();
+ Payload.SetContentType(ZenContentType::kCbObject);
+ JupiterResult PutMetaResult = m_Session.PutBlockMetadata(m_Namespace, m_Bucket, BuildId, BlockRawHash, Payload);
+ AddStatistic(PutMetaResult);
+ if (!PutMetaResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed putting build block metadata: {} ({})", PutMetaResult.Reason, PutMetaResult.ErrorCode));
+ }
+ }
+
+ virtual std::vector<ChunkBlockDescription> FindBlocks(const Oid& BuildId) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ JupiterResult FindResult = m_Session.FindBlocks(m_Namespace, m_Bucket, BuildId);
+ AddStatistic(FindResult);
+ if (!FindResult.Success)
+ {
+ throw std::runtime_error(fmt::format("Failed fetching known blocks: {} ({})", FindResult.Reason, FindResult.ErrorCode));
+ }
+ return ParseChunkBlockDescriptionList(PayloadToJson("Failed fetching known blocks"sv, FindResult.Response));
+ }
+
+ virtual std::vector<ChunkBlockDescription> GetBlockMetadata(const Oid& BuildId, std::span<const IoHash> BlockHashes) override
+ {
+ Stopwatch ExecutionTimer;
+ auto _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+ CbObjectWriter Request;
+
+ Request.BeginArray("blocks"sv);
+ for (const IoHash& BlockHash : BlockHashes)
+ {
+ Request.AddHash(BlockHash);
+ }
+ Request.EndArray();
+
+ IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer();
+ Payload.SetContentType(ZenContentType::kCbObject);
+ JupiterResult GetBlockMetadataResult = m_Session.GetBlockMetadata(m_Namespace, m_Bucket, BuildId, Payload);
+ AddStatistic(GetBlockMetadataResult);
+ if (!GetBlockMetadataResult.Success)
+ {
+ throw std::runtime_error(
+ fmt::format("Failed fetching block metadatas: {} ({})", GetBlockMetadataResult.Reason, GetBlockMetadataResult.ErrorCode));
+ }
+ std::vector<ChunkBlockDescription> UnorderedList =
+ ParseChunkBlockDescriptionList(PayloadToJson("Failed fetching block metadatas", GetBlockMetadataResult.Response));
+ tsl::robin_map<IoHash, size_t, IoHash::Hasher> BlockDescriptionLookup;
+ for (size_t DescriptionIndex = 0; DescriptionIndex < UnorderedList.size(); DescriptionIndex++)
+ {
+ const ChunkBlockDescription& Description = UnorderedList[DescriptionIndex];
+ BlockDescriptionLookup.insert_or_assign(Description.BlockHash, DescriptionIndex);
+ }
+ std::vector<ChunkBlockDescription> SortedBlockDescriptions;
+ SortedBlockDescriptions.reserve(BlockDescriptionLookup.size());
+ for (const IoHash& BlockHash : BlockHashes)
+ {
+ if (auto It = BlockDescriptionLookup.find(BlockHash); It != BlockDescriptionLookup.end())
+ {
+ SortedBlockDescriptions.push_back(std::move(UnorderedList[It->second]));
+ }
+ }
+ return SortedBlockDescriptions;
+ }
+
+private:
+ static CbObject PayloadToJson(std::string_view Context, const IoBuffer& Payload)
+ {
+ if (Payload.GetContentType() == ZenContentType::kJSON)
+ {
+ std::string_view Json(reinterpret_cast<const char*>(Payload.GetData()), Payload.GetSize());
+ return LoadCompactBinaryFromJson(Json).AsObject();
+ }
+ else if (Payload.GetContentType() == ZenContentType::kCbObject)
+ {
+ return LoadCompactBinaryObject(Payload);
+ }
+ else if (Payload.GetContentType() == ZenContentType::kCompressedBinary)
+ {
+ IoHash RawHash;
+ uint64_t RawSize;
+ return LoadCompactBinaryObject(CompressedBuffer::FromCompressed(SharedBuffer(Payload), RawHash, RawSize));
+ }
+ else
+ {
+ throw std::runtime_error(
+ fmt::format("{}: {} ({})", "Unsupported response format", Context, ToString(Payload.GetContentType())));
+ }
+ }
+
+ void AddStatistic(const JupiterResult& Result)
+ {
+ m_Stats.TotalBytesWritten += Result.SentBytes;
+ m_Stats.TotalBytesRead += Result.ReceivedBytes;
+ m_Stats.TotalRequestTimeUs += uint64_t(Result.ElapsedSeconds * 1000000.0);
+ m_Stats.TotalRequestCount++;
+ }
+
+ JupiterSession m_Session;
+ Statistics& m_Stats;
+ const std::string m_Namespace;
+ const std::string m_Bucket;
+ const std::filesystem::path m_TempFolderPath;
+};
+
+std::unique_ptr<BuildStorage>
+CreateJupiterBuildStorage(LoggerRef InLog,
+ HttpClient& InHttpClient,
+ BuildStorage::Statistics& Stats,
+ std::string_view Namespace,
+ std::string_view Bucket,
+ const std::filesystem::path& TempFolderPath)
+{
+ return std::make_unique<JupiterBuildStorage>(InLog, InHttpClient, Stats, Namespace, Bucket, TempFolderPath);
+}
+
+} // namespace zen