aiden
/
duckstation
mirror of https://github.com/stenzek/duckstation
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

ObjectArchive: Synchronize access internally

Browse Source 
Reduces the locked portion to not include compression/decompression.
pull/3797/merge
Stenzek 1 week ago
parent 60b06c0e28
commit 3151c5256b
No known key found for this signature in database
3 changed files with 196 additions and 119 deletions
Show Stats Download Patch File Download Diff File

2
src/util-tests/object_archive_tests.cpp
Unescape Escape View File

@ -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();
}

261
src/util/object_archive.cpp
Unescape Escape View File

@ -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;
ERROR_LOG("Failed to open existing object archive index '{}': {}", Path::GetFileName(index_filename),
open_error.GetDescription());
if (was_invalidated)
*was_invalidated = 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);
}
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();
// 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;
}
m_index_file = index_file;
m_blob_file = blob_file;
ERROR_LOG("Failed to open existing object archive index '{}': {}", Path::GetFileName(index_path),
open_error.GetDescription());
}
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<u8>(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,6 +356,17 @@ ObjectArchive::KeySpan ObjectArchive::GetKeySpan(const CacheIndexData& data) con
std::optional<ObjectArchive::ObjectData> ObjectArchive::Lookup(KeySpan key, Error* error)
{
std::optional<ObjectArchive::ObjectData> data;
CompressType compress_type;
u32 uncompressed_size;
{
// 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); });
@ -333,23 +376,26 @@ std::optional<ObjectArchive::ObjectData> ObjectArchive::Lookup(KeySpan key, Erro
return std::nullopt;
}
ObjectData data(iter->compressed_size);
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]]
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;
}
if (iter->compress_type == CompressType::Uncompressed)
return std::optional<ObjectData>(std::move(data));
compress_type = iter->compress_type;
uncompressed_size = iter->uncompressed_size;
}
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::ObjectData> 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,20 +424,25 @@ bool ObjectArchive::Insert(KeySpan key, std::span<const u8> data, CompressType c
Error::SetStringView(error, "Invalid key size.");
return false;
}
else if (!IsOpen()) [[unlikely]]
// Lookup once before compress, and again afterwards because we don't hold the lock.
{
Error::SetStringView(error, "Archive is not open.");
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,
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); });
if (iter != m_index.end() && KeyEqual(GetKeySpan(*iter), key))
iter != m_index.end() && KeyEqual(GetKeySpan(*iter), key))
{
Error::SetStringView(error, ERROR_DESCRIPTION_ALREADY_EXISTS);
return false;
}
}
DynamicHeapArray<u8> compress_buffer;
const void* write_data = data.data();
@ -407,6 +462,22 @@ bool ObjectArchive::Insert(KeySpan key, std::span<const u8> 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<const u8> 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;

14
src/util/object_archive.h
Unescape Escape View File

@ -7,6 +7,7 @@
#include "common/heap_array.h"
#include <mutex>
#include <optional>
#include <span>
#include <string>
@ -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<CacheIndexData>;
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;
};

Write Preview
Loading…
Cancel
Save