From 3151c5256bf43b7a6acff339a7b937d6e41199c9 Mon Sep 17 00:00:00 2001 From: Stenzek Date: Sun, 13 Sep 2026 15:39:11 +1000 Subject: [PATCH] ObjectArchive: Synchronize access internally Reduces the locked portion to not include compression/decompression. --- src/util-tests/object_archive_tests.cpp | 2 +- src/util/object_archive.cpp | 299 +++++++++++++++--------- src/util/object_archive.h | 14 +- 3 files changed, 196 insertions(+), 119 deletions(-) diff --git a/src/util-tests/object_archive_tests.cpp b/src/util-tests/object_archive_tests.cpp index e12ff7347..91b9b5a78 100644 --- a/src/util-tests/object_archive_tests.cpp +++ b/src/util-tests/object_archive_tests.cpp @@ -142,7 +142,7 @@ TEST(ObjectArchive, OpenPathInvalidationStatus) { ObjectArchive archive; ASSERT_TRUE(archive.OpenPath(path.GetPath(), TEST_VERSION, &error, &was_invalidated)) << error.GetDescription(); - EXPECT_FALSE(was_invalidated); + EXPECT_TRUE(was_invalidated); ASSERT_TRUE(archive.Insert(StringToCacheKey("persist"), payload, ObjectArchive::CompressType::Uncompressed, &error)) << error.GetDescription(); } diff --git a/src/util/object_archive.cpp b/src/util/object_archive.cpp index fcb39e62a..46bd0b3de 100644 --- a/src/util/object_archive.cpp +++ b/src/util/object_archive.cpp @@ -62,13 +62,20 @@ ObjectArchive::ObjectArchive() = default; ObjectArchive::~ObjectArchive() { - Close(); + LockedClose(); } +constinit const std::string_view ObjectArchive::ERROR_DESCRIPTION_NOT_OPEN = "Archive is not open."; constinit const std::string_view ObjectArchive::ERROR_DESCRIPTION_DOES_NOT_EXIST = "Key not found in archive."; constinit const std::string_view ObjectArchive::ERROR_DESCRIPTION_ALREADY_EXISTS = "Key already exists in archive."; void ObjectArchive::Close() +{ + std::unique_lock lock(m_mutex); + LockedClose(); +} + +void ObjectArchive::LockedClose() { if (m_index_file) { @@ -86,6 +93,8 @@ void ObjectArchive::Close() bool ObjectArchive::Clear(Error* error) { + std::unique_lock lock(m_mutex); + if (!IsOpen()) return true; @@ -97,7 +106,7 @@ bool ObjectArchive::Clear(Error* error) !FileSystem::FSeek64(m_index_file, 0, SEEK_END, error)) { ERROR_LOG("Failed to seek/truncate object cache"); - Close(); + LockedClose(); return false; } @@ -106,54 +115,68 @@ bool ObjectArchive::Clear(Error* error) return true; } +size_t ObjectArchive::GetSize() const +{ + std::unique_lock lock(m_mutex); + return m_index.size(); +} + bool ObjectArchive::OpenPath(std::string_view base_path, u32 data_version, Error* error, bool* was_invalidated) { - Close(); + std::unique_lock lock(m_mutex); + + LockedClose(); if (was_invalidated) *was_invalidated = false; - const std::string index_filename = fmt::format("{}.idx", base_path); - const std::string blob_filename = fmt::format("{}.bin", base_path); + const std::string index_path = fmt::format("{}.idx", base_path); + const std::string blob_path = fmt::format("{}.bin", base_path); - if (FileSystem::FileExists(index_filename.c_str())) + if (FileSystem::FileExists(index_path.c_str())) { Error open_error; - if (OpenExisting(index_filename, blob_filename, data_version, &open_error)) + bool was_sharing_violation = false; + std::FILE* index_file = FileSystem::OpenCFile(index_path.c_str(), "r+b", &open_error); + if (index_file) [[likely]] + { + std::FILE* blob_file = FileSystem::OpenCFile(blob_path.c_str(), "a+b", &open_error); + if (blob_file) [[likely]] + { + if (ReadExisting(data_version, index_file, blob_file, &open_error)) + return true; + + // ReadExisting() doesn't consume the handles. + std::fclose(blob_file); + std::fclose(index_file); + } + else + { + was_sharing_violation = (errno == EACCES); + std::fclose(index_file); + ERROR_LOG("Blob file '{}' is missing", Path::GetFileName(blob_path)); + } + } + else + { + was_sharing_violation = (errno == EACCES); + } + + // special case here: when there's a sharing violation (i.e. two instances running), + // we don't want to blow away the cache. so just continue without a cache. + if (was_sharing_violation) + { + WARNING_LOG("Failed to open archive index with EACCES, are you running two instances?"); return true; + } - ERROR_LOG("Failed to open existing object archive index '{}': {}", Path::GetFileName(index_filename), + ERROR_LOG("Failed to open existing object archive index '{}': {}", Path::GetFileName(index_path), open_error.GetDescription()); - - if (was_invalidated) - *was_invalidated = true; } - return CreateNew(index_filename, blob_filename, data_version, error); -} - -bool ObjectArchive::OpenFile(std::FILE* index_file, std::FILE* blob_file, u32 data_version, Error* error) -{ - Close(); - - m_index_file = index_file; - m_blob_file = blob_file; - - return ReadExisting(data_version, error); -} - -bool ObjectArchive::CreateFile(std::FILE* index_file, std::FILE* blob_file, u32 data_version, Error* error) -{ - Close(); - - m_index_file = index_file; - m_blob_file = blob_file; - - return CreateNew(data_version, error); -} + if (was_invalidated) + *was_invalidated = true; -bool ObjectArchive::CreateNew(const std::string& index_path, const std::string& blob_path, u32 version, Error* error) -{ if (FileSystem::FileExists(blob_path.c_str())) { WARNING_LOG("Removing existing blob file '{}'", Path::GetFileName(blob_path)); @@ -165,26 +188,27 @@ bool ObjectArchive::CreateNew(const std::string& index_path, const std::string& FileSystem::DeleteFile(index_path.c_str()); } - m_index_file = FileSystem::OpenCFile(index_path.c_str(), "wb", error); - if (!m_index_file) [[unlikely]] + std::FILE* index_file = FileSystem::OpenCFile(index_path.c_str(), "wb", error); + if (!index_file) [[unlikely]] { ERROR_LOG("Failed to open index file '{}' for writing", Path::GetFileName(index_path)); return false; } - m_blob_file = FileSystem::OpenCFile(blob_path.c_str(), "w+b", error); - if (!m_blob_file) [[unlikely]] + std::FILE* blob_file = FileSystem::OpenCFile(blob_path.c_str(), "w+b", error); + if (!blob_file) [[unlikely]] { ERROR_LOG("Failed to open blob file '{}' for writing", Path::GetFileName(blob_path)); - Close(); + std::fclose(index_file); FileSystem::DeleteFile(index_path.c_str()); return false; } - if (!CreateNew(version, error)) + if (!CreateNew(data_version, index_file, blob_file, error)) { ERROR_LOG("Failed to create to index file '{}'", Path::GetFileName(index_path)); - Close(); + std::fclose(blob_file); + std::fclose(index_file); FileSystem::DeleteFile(blob_path.c_str()); FileSystem::DeleteFile(index_path.c_str()); return false; @@ -193,72 +217,76 @@ bool ObjectArchive::CreateNew(const std::string& index_path, const std::string& return true; } -bool ObjectArchive::CreateNew(u32 version, Error* error) +bool ObjectArchive::OpenFile(std::FILE* index_file, std::FILE* blob_file, u32 data_version, Error* error) { - CacheFileHeader file_header; - file_header.signature = EXPECTED_SIGNATURE; - file_header.cache_version = version; - if (std::fwrite(&file_header, sizeof(file_header), 1, m_index_file) != 1) [[unlikely]] + std::unique_lock lock(m_mutex); + + LockedClose(); + + if (!ReadExisting(data_version, index_file, blob_file, error)) { - Error::SetErrno(error, "fwrite() for version failed: ", errno); + // Need to consume the file pointers. + std::fclose(index_file); + std::fclose(blob_file); return false; } return true; } -bool ObjectArchive::OpenExisting(const std::string& index_path, const std::string& blob_path, u32 version, Error* error) +bool ObjectArchive::CreateFile(std::FILE* index_file, std::FILE* blob_file, u32 data_version, Error* error) { - m_index_file = FileSystem::OpenCFile(index_path.c_str(), "r+b", error); - if (!m_index_file) - { - // special case here: when there's a sharing violation (i.e. two instances running), - // we don't want to blow away the cache. so just continue without a cache. - if (errno == EACCES) - { - WARNING_LOG("Failed to open archive index with EACCES, are you running two instances?"); - return true; - } + std::unique_lock lock(m_mutex); + LockedClose(); + + if (!CreateNew(data_version, index_file, blob_file, error)) + { + // Need to consume the file pointers. + std::fclose(index_file); + std::fclose(blob_file); return false; } - m_blob_file = FileSystem::OpenCFile(blob_path.c_str(), "a+b", error); - if (!m_blob_file) [[unlikely]] + return true; +} + +bool ObjectArchive::CreateNew(u32 version, std::FILE* index_file, std::FILE* blob_file, Error* error) +{ + CacheFileHeader file_header; + file_header.signature = EXPECTED_SIGNATURE; + file_header.cache_version = version; + if (std::fwrite(&file_header, sizeof(file_header), 1, index_file) != 1) [[unlikely]] { - ERROR_LOG("Blob file '{}' is missing", Path::GetFileName(blob_path)); - Close(); + Error::SetErrno(error, "fwrite() for version failed: ", errno); return false; } - return ReadExisting(version, error); + m_index_file = index_file; + m_blob_file = blob_file; + return true; } -bool ObjectArchive::ReadExisting(u32 version, Error* error) +bool ObjectArchive::ReadExisting(u32 version, std::FILE* index_file, std::FILE* blob_file, Error* error) { CacheFileHeader file_header; - if (std::fread(&file_header, sizeof(file_header), 1, m_index_file) != 1 || + if (std::fread(&file_header, sizeof(file_header), 1, index_file) != 1 || file_header.signature != EXPECTED_SIGNATURE || file_header.cache_version != version) [[unlikely]] { Error::SetStringFmt(error, "Bad file/data version (expected {}, got {})", version, file_header.cache_version); - Close(); return false; } - const s64 index_file_size = FileSystem::FSize64(m_index_file); + const s64 index_file_size = FileSystem::FSize64(index_file); if (index_file_size < 0 || index_file_size > 1 * 1048576) { Error::SetStringFmt(error, "Index file is too large ({} bytes)", index_file_size); - Close(); return false; } - const s64 blob_file_size = FileSystem::FSize64(m_blob_file, error); + const s64 blob_file_size = FileSystem::FSize64(blob_file, error); if (blob_file_size < 0) - { - Close(); return false; - } // preallocate key storage, this will overshoot a bit since we don't know the actual key sizes, but it should be // good enough to avoid fragmentation and multiple resizes in most cases. @@ -271,17 +299,17 @@ bool ObjectArchive::ReadExisting(u32 version, Error* error) CacheIndexEntryHeader key_header; u32 key_size; - if (std::fread(&key_header, sizeof(key_header), 1, m_index_file) != 1 || - (key_size = key_header.GetKeySize()) == 0 || key_size > MAX_KEY_SIZE || - (key_header.file_offset + key_header.compressed_size) > blob_file_size || + if (std::fread(&key_header, sizeof(key_header), 1, index_file) != 1 || (key_size = key_header.GetKeySize()) == 0 || + key_size > MAX_KEY_SIZE || (key_header.file_offset + key_header.compressed_size) > blob_file_size || key_header.compress_type >= static_cast(CompressType::Count) || - (key.resize(key_size), std::fread(key.data(), key_size, 1, m_index_file)) != 1) [[unlikely]] + (key.resize(key_size), std::fread(key.data(), key_size, 1, index_file)) != 1) [[unlikely]] { - if (std::feof(m_index_file)) + if (std::feof(index_file)) break; Error::SetErrno(error, "fread() failed: ", errno); - Close(); + m_index.clear(); + m_key_pool.clear(); return false; } @@ -292,9 +320,10 @@ bool ObjectArchive::ReadExisting(u32 version, Error* error) } // ensure we don't write before seeking - if (!FileSystem::FSeek64(m_index_file, 0, SEEK_END, error)) + if (!FileSystem::FSeek64(index_file, 0, SEEK_END, error)) { - Close(); + m_index.clear(); + m_key_pool.clear(); return false; } @@ -308,12 +337,15 @@ bool ObjectArchive::ReadExisting(u32 version, Error* error) if (KeyEqual(GetKeySpan(m_index[i - 1]), GetKeySpan(m_index[i]))) { Error::SetStringView(error, "Duplicate key in index file, corrupt file?"); - Close(); + m_index.clear(); + m_key_pool.clear(); return false; } } DEV_LOG("Read {} entries in {:.2f} ms", m_index.size(), timer.GetTimeMilliseconds()); + m_blob_file = blob_file; + m_index_file = index_file; return true; } @@ -324,32 +356,46 @@ ObjectArchive::KeySpan ObjectArchive::GetKeySpan(const CacheIndexData& data) con std::optional ObjectArchive::Lookup(KeySpan key, Error* error) { - const auto iter = - std::lower_bound(m_index.begin(), m_index.end(), key, - [this](const CacheIndexData& entry, KeySpan key) { return KeyLess(GetKeySpan(entry), key); }); - if (iter == m_index.end() || !KeyEqual(GetKeySpan(*iter), key)) + std::optional data; + CompressType compress_type; + u32 uncompressed_size; { - Error::SetStringView(error, ERROR_DESCRIPTION_DOES_NOT_EXIST); - return std::nullopt; - } + // Minimize the time locked, only the lookup+read, not the decompress. + std::unique_lock lock(m_mutex); + if (!IsOpen()) [[unlikely]] + { + Error::SetStringView(error, ERROR_DESCRIPTION_NOT_OPEN); + return std::nullopt; + } + const auto iter = + std::lower_bound(m_index.begin(), m_index.end(), key, + [this](const CacheIndexData& entry, KeySpan key) { return KeyLess(GetKeySpan(entry), key); }); + if (iter == m_index.end() || !KeyEqual(GetKeySpan(*iter), key)) + { + Error::SetStringView(error, ERROR_DESCRIPTION_DOES_NOT_EXIST); + return std::nullopt; + } - ObjectData data(iter->compressed_size); - if (std::fseek(m_blob_file, iter->file_offset, SEEK_SET) != 0 || - std::fread(data.data(), iter->compressed_size, 1, m_blob_file) != 1) [[unlikely]] - { - ERROR_LOG("failed to read {} byte object at offset {}", iter->compressed_size, iter->file_offset); - Error::SetErrno(error, errno); - return std::nullopt; + data.emplace(iter->compressed_size); + if (std::fseek(m_blob_file, iter->file_offset, SEEK_SET) != 0 || + std::fread(data->data(), iter->compressed_size, 1, m_blob_file) != 1) [[unlikely]] + { + ERROR_LOG("failed to read {} byte object at offset {}", iter->compressed_size, iter->file_offset); + Error::SetErrno(error, errno); + return std::nullopt; + } + + compress_type = iter->compress_type; + uncompressed_size = iter->uncompressed_size; } - if (iter->compress_type == CompressType::Uncompressed) - return std::optional(std::move(data)); + if (compress_type == CompressType::Uncompressed) + return data; ObjectData uncompressed_data; - if (!CompressHelpers::DecompressBuffer(uncompressed_data, iter->compress_type, data.cspan(), iter->uncompressed_size, - error)) + if (!CompressHelpers::DecompressBuffer(uncompressed_data, compress_type, data->cspan(), uncompressed_size, error)) { - ERROR_LOG("Decompress {} byte object failed", iter->uncompressed_size); + ERROR_LOG("Decompress {} byte object failed", uncompressed_size); return std::nullopt; } @@ -358,7 +404,11 @@ std::optional ObjectArchive::Lookup(KeySpan key, Erro bool ObjectArchive::Contains(KeySpan key) const { - if (key.empty() || key.size() > MAX_KEY_SIZE || !IsOpen()) [[unlikely]] + if (key.empty() || key.size() > MAX_KEY_SIZE) [[unlikely]] + return false; + + std::unique_lock lock(m_mutex); + if (!IsOpen()) [[unlikely]] return false; const auto iter = @@ -374,19 +424,24 @@ bool ObjectArchive::Insert(KeySpan key, std::span data, CompressType c Error::SetStringView(error, "Invalid key size."); return false; } - else if (!IsOpen()) [[unlikely]] - { - Error::SetStringView(error, "Archive is not open."); - return false; - } - const auto iter = - std::lower_bound(m_index.begin(), m_index.end(), key, - [this](const CacheIndexData& entry, KeySpan key) { return KeyLess(GetKeySpan(entry), key); }); - if (iter != m_index.end() && KeyEqual(GetKeySpan(*iter), key)) + // Lookup once before compress, and again afterwards because we don't hold the lock. { - Error::SetStringView(error, ERROR_DESCRIPTION_ALREADY_EXISTS); - return false; + std::unique_lock lock(m_mutex); + if (!IsOpen()) [[unlikely]] + { + Error::SetStringView(error, ERROR_DESCRIPTION_NOT_OPEN); + return false; + } + + if (const auto iter = std::lower_bound( + m_index.begin(), m_index.end(), key, + [this](const CacheIndexData& entry, KeySpan key) { return KeyLess(GetKeySpan(entry), key); }); + iter != m_index.end() && KeyEqual(GetKeySpan(*iter), key)) + { + Error::SetStringView(error, ERROR_DESCRIPTION_ALREADY_EXISTS); + return false; + } } DynamicHeapArray compress_buffer; @@ -407,6 +462,22 @@ bool ObjectArchive::Insert(KeySpan key, std::span data, CompressType c write_size = compress_buffer.size(); } + // See above. + std::unique_lock lock(m_mutex); + if (!IsOpen()) [[unlikely]] + { + Error::SetStringView(error, ERROR_DESCRIPTION_NOT_OPEN); + return false; + } + const auto iter = + std::lower_bound(m_index.begin(), m_index.end(), key, + [this](const CacheIndexData& entry, KeySpan key) { return KeyLess(GetKeySpan(entry), key); }); + if (iter != m_index.end() && KeyEqual(GetKeySpan(*iter), key)) + { + Error::SetStringView(error, ERROR_DESCRIPTION_ALREADY_EXISTS); + return false; + } + if (!m_blob_file || !FileSystem::FSeek64(m_blob_file, 0, SEEK_END, error)) return false; @@ -449,6 +520,7 @@ bool ObjectArchive::Insert(KeySpan key, std::span data, CompressType c u64 ObjectArchive::GetTotalObjectSize() const { + std::unique_lock lock(m_mutex); u64 total_size = 0; for (const CacheIndexData& entry : m_index) total_size += entry.uncompressed_size; @@ -457,6 +529,7 @@ u64 ObjectArchive::GetTotalObjectSize() const u64 ObjectArchive::GetTotalSize() const { + std::unique_lock lock(m_mutex); u64 total_size = 0; for (const CacheIndexData& entry : m_index) total_size += entry.compressed_size + sizeof(CacheIndexEntryHeader) + entry.key_size; diff --git a/src/util/object_archive.h b/src/util/object_archive.h index 719f372fc..22ab4d106 100644 --- a/src/util/object_archive.h +++ b/src/util/object_archive.h @@ -7,6 +7,7 @@ #include "common/heap_array.h" +#include #include #include #include @@ -26,14 +27,16 @@ public: ~ObjectArchive(); /// Error messages indicating the key does/does not exist in the archive. + static const std::string_view ERROR_DESCRIPTION_NOT_OPEN; static const std::string_view ERROR_DESCRIPTION_DOES_NOT_EXIST; static const std::string_view ERROR_DESCRIPTION_ALREADY_EXISTS; /// Returns true if the archive has been successfully opened or created. + /// NOTE: Not synchronized, because we should only be opening the archive on one thread. bool IsOpen() const { return (m_index_file != nullptr); } /// Returns the number of entries currently stored in the archive. - size_t GetSize() const { return m_index.size(); } + size_t GetSize() const; /// Opens or creates an archive at the given base path. The index and blob files will be named /// "{base_path}.idx" and "{base_path}.bin" respectively. If the files already exist and match @@ -86,10 +89,9 @@ private: }; using CacheIndex = std::vector; - bool CreateNew(const std::string& index_path, const std::string& blob_path, u32 version, Error* error); - bool CreateNew(u32 version, Error* error); - bool OpenExisting(const std::string& index_path, const std::string& blob_path, u32 version, Error* error); - bool ReadExisting(u32 version, Error* error); + bool CreateNew(u32 version, std::FILE* index_file, std::FILE* blob_file, Error* error); + bool ReadExisting(u32 version, std::FILE* index_file, std::FILE* blob_file, Error* error); + void LockedClose(); KeySpan GetKeySpan(const CacheIndexData& data) const; @@ -98,4 +100,6 @@ private: std::FILE* m_index_file = nullptr; std::FILE* m_blob_file = nullptr; + + mutable std::mutex m_mutex; };