mirror of
https://github.com/MyLuxy/overdrive.git
synced 2026-10-19 02:10:55 -05:00
fix: extraction receipts no longer limit how many files a game has
This commit is contained in:
1 parent
4e7d4fa070
commit
02992ccb7c
8 files changed
+317
-68
No files matched your search
@@ -1,5 +1,14 @@
|
||||
# Changelog
|
||||
|
||||
## Unreleased — extraction receipts of any size
|
||||
|
||||
- An extraction's receipt (one checksum line per file) is written to disk as it grows, through a 1 MiB
|
||||
buffer, instead of kept whole in memory, and a later run reads it back a line at a time. A game packed
|
||||
as a folder (issue #52: ASTRO BOT, Resident Evil Requiem) holds more files than the 16 MiB a receipt
|
||||
could take, and failed with "Extraction receipt exceeds the metadata limit." Memory now stays the same
|
||||
whatever the number of files; an archive may hold up to 1,000,000 entries (was 100,000). The receipt
|
||||
file keeps its format, so extractions made before still recover.
|
||||
|
||||
## Unreleased — PS5 packages install
|
||||
|
||||
- The package installer payload takes a PS5 package (`\x7F` `FIH`) as well as a PS4 one
|
||||
|
||||
@@ -35,7 +35,9 @@ Writer::~Writer() {
|
||||
stop_threads();
|
||||
}
|
||||
|
||||
bool Writer::start(std::size_t ring_bytes) {
|
||||
bool Writer::start(std::size_t ring_bytes, Receipt& receipt) {
|
||||
receipt_ = &receipt;
|
||||
first_line_ = receipt.lines();
|
||||
if (ring_bytes < 1024 * 1024) return false;
|
||||
ring_ = static_cast<unsigned char*>(std::malloc(ring_bytes));
|
||||
if (!ring_) return false;
|
||||
@@ -222,8 +224,7 @@ void Writer::write_loop() {
|
||||
}
|
||||
if (!skip && digest.size() != 64) fail("Cannot verify extracted file.");
|
||||
if (!failed()) {
|
||||
receipt_ += digest + "|" + name + "\n";
|
||||
if (receipt_.size() > max_metadata_bytes) fail("Extraction receipt exceeds the metadata limit.");
|
||||
if (!receipt_->add(digest, name)) fail("Cannot write extraction receipt.");
|
||||
const auto folder = path.size() - name.size() - 1;
|
||||
if (!path.ends_with("/" + name) || (!root_.empty() && path.compare(0, folder, root_)))
|
||||
fail("Extracted file outside its folder.");
|
||||
@@ -256,11 +257,14 @@ std::string Writer::readback(const std::string& file) {
|
||||
}
|
||||
|
||||
// Every file is on disk before the receipt: inputs are deleted once extraction completes.
|
||||
void Writer::sync_loop(const std::vector<std::size_t>& lines, std::atomic<std::size_t>& next, std::atomic<std::size_t>& synced_files, const Progress& progress) {
|
||||
void Writer::sync_loop(Receipt::Lines* lines, std::mutex& reading, std::atomic<std::size_t>& synced_files, std::size_t total, const Progress& progress) {
|
||||
name_thread("archive-sync");
|
||||
for (std::size_t index; !failed() && (index = next++) < lines.size();) {
|
||||
const auto at = receipt_.find('|', lines[index]) + 1;
|
||||
const auto file = root_ + "/" + receipt_.substr(at, receipt_.find('\n', at) - at);
|
||||
for (std::string line; !failed();) {
|
||||
{
|
||||
std::lock_guard lock(reading);
|
||||
if (!lines || !lines->next(line)) break;
|
||||
}
|
||||
const auto file = root_ + "/" + line.substr(line.find('|') + 1);
|
||||
const auto syncing = std::chrono::steady_clock::now();
|
||||
// fsync needs no write access; a filesystem that wants it gets the file opened for writing.
|
||||
int fd = open(file.c_str(), O_RDONLY | O_NOFOLLOW);
|
||||
@@ -275,7 +279,7 @@ void Writer::sync_loop(const std::vector<std::size_t>& lines, std::atomic<std::s
|
||||
timings_.sync_us += elapsed_us(syncing);
|
||||
++timings_.syncs;
|
||||
if (!synced) fail(std::strerror(error));
|
||||
else if (progress) progress(++synced_files, lines.size());
|
||||
else if (progress) progress(++synced_files, total);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -291,21 +295,22 @@ void Writer::stop_threads() {
|
||||
ring_ = nullptr;
|
||||
}
|
||||
|
||||
std::string Writer::finish(std::string& receipt, const Progress& progress) {
|
||||
std::string Writer::finish(const Progress& progress) {
|
||||
stop_threads();
|
||||
std::vector<std::size_t> lines;
|
||||
for (std::size_t at = 0; at < receipt_.size(); at = receipt_.find('\n', at) + 1) lines.push_back(at);
|
||||
std::atomic<std::size_t> next{0}, synced{0};
|
||||
if (progress && !failed()) progress(0, lines.size());
|
||||
const std::size_t total = receipt_ ? receipt_->lines() - first_line_ : 0;
|
||||
Receipt::Lines* lines = nullptr;
|
||||
if (!failed() && total && !receipt_->lines_from(first_line_, lines)) fail("Cannot read extraction receipt.");
|
||||
std::mutex reading;
|
||||
std::atomic<std::size_t> synced{0};
|
||||
if (progress && !failed()) progress(0, total);
|
||||
// Without threads, the files are synced here, one after another.
|
||||
std::array<WorkerThread, syncers> threads;
|
||||
for (auto& thread : threads)
|
||||
if (!failed() && lines.size() > 1) thread.start([&] { sync_loop(lines, next, synced, progress); }, thread_stack);
|
||||
sync_loop(lines, next, synced, progress);
|
||||
if (!failed() && total > 1) thread.start([&] { sync_loop(lines, reading, synced, total, progress); }, thread_stack);
|
||||
sync_loop(lines, reading, synced, total, progress);
|
||||
for (auto& thread : threads) thread.join();
|
||||
if (cancelled_) return {};
|
||||
if (failed_ || aborted_) return error_.empty() ? "Extraction stopped." : error_;
|
||||
receipt += receipt_;
|
||||
return {};
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
// Additional attribution term: see LICENSE-ATTRIBUTION.
|
||||
#pragma once
|
||||
#include "extraction_receipt.hpp"
|
||||
#include "worker_thread.hpp"
|
||||
#include <array>
|
||||
#include <atomic>
|
||||
@@ -35,17 +36,18 @@ public:
|
||||
~Writer();
|
||||
Writer(const Writer&) = delete;
|
||||
Writer& operator=(const Writer&) = delete;
|
||||
// False when the ring or a thread cannot be had; the caller then writes inline.
|
||||
bool start(std::size_t ring_bytes);
|
||||
// False when the ring or a thread cannot be had; the caller then writes inline. Each file's line goes
|
||||
// to `receipt` once it is written and hashed.
|
||||
bool start(std::size_t ring_bytes, Receipt& receipt);
|
||||
// Takes ownership of `fd` even on failure. `path` is the file's absolute path, `name` its receipt name.
|
||||
bool begin(int fd, std::string path, std::string name);
|
||||
bool append(const void* bytes, std::size_t size, std::uint64_t offset);
|
||||
bool end();
|
||||
// Waits until every file is written, hashed, synced and closed, appends their receipt lines, and
|
||||
// returns the first error (empty on success or cancellation). `progress` gets the files synced of the
|
||||
// total once the writes are done and after each fsync, from any of the syncing threads.
|
||||
// Waits until every file is written, hashed, synced and closed (their receipt lines are in the receipt
|
||||
// given to start()), and returns the first error (empty on success or cancellation). `progress` gets the
|
||||
// files synced of the total once the writes are done and after each fsync, from any of the syncing threads.
|
||||
using Progress = std::function<void(std::size_t synced, std::size_t total)>;
|
||||
std::string finish(std::string& receipt, const Progress& progress = {});
|
||||
std::string finish(const Progress& progress = {});
|
||||
// The first error, or empty when the job was cancelled.
|
||||
std::string error();
|
||||
|
||||
@@ -62,8 +64,8 @@ private:
|
||||
bool sealed = true, busy = false;
|
||||
};
|
||||
void write_loop();
|
||||
// fsyncs the receipt's files from `next` on, one at a time, until none is left or one fails.
|
||||
void sync_loop(const std::vector<std::size_t>& lines, std::atomic<std::size_t>& next, std::atomic<std::size_t>& synced_files, const Progress& progress);
|
||||
// fsyncs this writer's files, read in turn from the receipt, one at a time, until none is left or one fails.
|
||||
void sync_loop(Receipt::Lines* lines, std::mutex& reading, std::atomic<std::size_t>& synced_files, std::size_t total, const Progress& progress);
|
||||
bool push(Command command);
|
||||
std::string readback(const std::string& file);
|
||||
bool failed() const { return failed_.load() || aborted_.load() || cancelled_.load(); }
|
||||
@@ -78,8 +80,11 @@ private:
|
||||
std::uint64_t head_ = 0, tail_ = 0;
|
||||
std::size_t queued_files_ = 0;
|
||||
std::deque<Command> commands_;
|
||||
// The receipt's names are relative to this folder, taken from the first file.
|
||||
std::string receipt_, error_, root_;
|
||||
// The receipt's names are relative to this folder, taken from the first file. This writer's lines start
|
||||
// at `first_line_` of `receipt_`.
|
||||
std::string error_, root_;
|
||||
Receipt* receipt_ = nullptr;
|
||||
std::uint64_t first_line_ = 0;
|
||||
std::mutex mutex_;
|
||||
std::condition_variable work_, space_;
|
||||
std::atomic<bool> failed_{false}, aborted_{false};
|
||||
|
||||
@@ -291,17 +291,13 @@ bool recover(Job& job, const std::string& identity) {
|
||||
const auto root = job.request.destination;
|
||||
struct stat st;
|
||||
const auto receipt = root + "/" + receipt_name;
|
||||
if (lstat(receipt.c_str(), &st) || !S_ISREG(st.st_mode) || static_cast<std::uint64_t>(st.st_size) > max_metadata_bytes) return false;
|
||||
std::string metadata;
|
||||
if (!read_metadata(receipt, metadata)) return false;
|
||||
std::size_t cursor = 0;
|
||||
auto next_line = [&](std::string& line) {
|
||||
const auto end = metadata.find('\n', cursor);
|
||||
if (end == std::string::npos) return false;
|
||||
line = metadata.substr(cursor, end - cursor);
|
||||
cursor = end + 1;
|
||||
return true;
|
||||
};
|
||||
if (lstat(receipt.c_str(), &st) || !S_ISREG(st.st_mode)) return false;
|
||||
// Read a line at a time: a game folder's receipt is far bigger than anything kept in memory (issue #52).
|
||||
const int fd = open(receipt.c_str(), O_RDONLY | O_NOFOLLOW);
|
||||
if (fd < 0) return false;
|
||||
struct Closer { int fd; ~Closer() { close(fd); } } closer{fd};
|
||||
LineReader reader(fd);
|
||||
auto next_line = [&](std::string& line) { return reader.next(line); };
|
||||
std::string line;
|
||||
if (!next_line(line) || line != "P5AR001:" + identity) return false;
|
||||
Snapshot verified;
|
||||
@@ -319,7 +315,7 @@ bool recover(Job& job, const std::string& identity) {
|
||||
if (installable(path)) verified.artifacts.push_back(path);
|
||||
if (verified.artifacts.size() > max_artifacts) return false;
|
||||
}
|
||||
if (cursor != metadata.size()) return false;
|
||||
if (!reader.clean()) return false;
|
||||
std::lock_guard lock(job.mutex);
|
||||
job.snapshot.written = verified.written;
|
||||
job.snapshot.entries = verified.entries;
|
||||
@@ -472,7 +468,7 @@ archive* open_archive(Job& job, std::string& error, std::unique_ptr<Rar5Password
|
||||
return archive;
|
||||
}
|
||||
// Writes every entry under root and appends its digest line to the receipt.
|
||||
std::string extract(Job& job, archive* archive, const std::string& staging, std::string& receipt, Rar5PasswordReader* decrypted, Writer* writer) {
|
||||
std::string extract(Job& job, archive* archive, const std::string& staging, Receipt& receipt, Rar5PasswordReader* decrypted, Writer* writer) {
|
||||
const auto& request = job.request;
|
||||
archive_entry* entry = nullptr;
|
||||
int result = ARCHIVE_OK;
|
||||
@@ -518,8 +514,7 @@ std::string extract(Job& job, archive* archive, const std::string& staging, std:
|
||||
const auto error = write_entry(job, archive, fd, size, staging + "/" + path, digest, decrypted ? &crc : nullptr);
|
||||
close(fd);
|
||||
if (!error.empty() || job.cancelled) return error;
|
||||
receipt += digest + "|" + path + "\n";
|
||||
if (receipt.size() > max_metadata_bytes) return "Extraction receipt exceeds the metadata limit.";
|
||||
if (!receipt.add(digest, path)) return "Cannot write extraction receipt.";
|
||||
}
|
||||
if (decrypted && !decrypted->verify(original_name, crc)) return "Extracted RAR5 file checksum mismatch.";
|
||||
std::lock_guard lock(job.mutex);
|
||||
@@ -571,7 +566,7 @@ std::string rar_request(const Request& request, const std::string& staging) {
|
||||
}
|
||||
// One file the worker wrote and synced: "<sha256> <size> <path>". It is checked as libarchive's entries
|
||||
// are, since the worker is a separate program.
|
||||
std::string rar_file(Job& job, const std::string& staging, const std::string& report, std::uint64_t& total, std::string& receipt) {
|
||||
std::string rar_file(Job& job, const std::string& staging, const std::string& report, std::uint64_t& total, Receipt& receipt) {
|
||||
const auto space = report.find(' '), next = space == std::string::npos ? space : report.find(' ', space + 1);
|
||||
if (space != 64 || next == std::string::npos) return "Invalid RAR worker output.";
|
||||
const auto digest = report.substr(0, space), path = report.substr(next + 1), size_text = report.substr(space + 1, next - space - 1);
|
||||
@@ -585,8 +580,7 @@ std::string rar_file(Job& job, const std::string& staging, const std::string& re
|
||||
return "Archive contains an unsafe path, link or special file.";
|
||||
if (size > job.request.max_bytes - total) return "Archive exceeds the extraction size limit.";
|
||||
total += size;
|
||||
receipt += digest + "|" + path + "\n";
|
||||
if (receipt.size() > max_metadata_bytes) return "Extraction receipt exceeds the metadata limit.";
|
||||
if (!receipt.add(digest, path)) return "Cannot write extraction receipt.";
|
||||
std::lock_guard lock(job.mutex);
|
||||
if (installable(path)) {
|
||||
if (job.snapshot.artifacts.size() == max_artifacts) return "Archive has too many installable artifacts.";
|
||||
@@ -596,7 +590,7 @@ std::string rar_file(Job& job, const std::string& staging, const std::string& re
|
||||
return {};
|
||||
}
|
||||
// False when the worker could not be started, before anything was written: libarchive extracts instead.
|
||||
bool extract_rar(Job& job, const std::string& staging, std::string& receipt, std::string& error) {
|
||||
bool extract_rar(Job& job, const std::string& staging, Receipt& receipt, std::string& error) {
|
||||
std::string unavailable;
|
||||
const int stream = start_rar_worker(rar_request(job.request, staging), unavailable);
|
||||
if (stream < 0) {
|
||||
@@ -782,7 +776,7 @@ void run(const std::shared_ptr<Job>& pointer) {
|
||||
fail(job, "Cannot record extraction staging ownership.");
|
||||
return;
|
||||
}
|
||||
std::string receipt = "P5AR001:" + identity + "\n";
|
||||
Receipt receipt(staging, "P5AR001:" + identity);
|
||||
std::string error;
|
||||
trace("staging ok", staging.c_str());
|
||||
{
|
||||
@@ -804,7 +798,7 @@ void run(const std::shared_ptr<Job>& pointer) {
|
||||
{
|
||||
// Without the ring or its threads, files are written inline as before.
|
||||
Writer writer(job.cancelled, job.timings);
|
||||
const bool pipelined = writer.start(pipeline_bytes());
|
||||
const bool pipelined = writer.start(pipeline_bytes(), receipt);
|
||||
trace("writer", pipelined ? "pipelined" : "inline");
|
||||
const auto started = std::chrono::steady_clock::now();
|
||||
const auto cpu_started = thread_cpu_us();
|
||||
@@ -812,7 +806,7 @@ void run(const std::shared_ptr<Job>& pointer) {
|
||||
const auto decoded_us = elapsed_us(started), cpu_us = thread_cpu_us() - cpu_started;
|
||||
// On an error or cancel the writer's destructor drops what is queued and closes every file.
|
||||
if (error.empty() && !job.cancelled) enter(job, "finishing");
|
||||
if (pipelined && error.empty() && !job.cancelled) error = writer.finish(receipt, [&](std::size_t synced, std::size_t total) {
|
||||
if (pipelined && error.empty() && !job.cancelled) error = writer.finish([&](std::size_t synced, std::size_t total) {
|
||||
std::lock_guard lock(job.mutex);
|
||||
job.snapshot.synced = synced;
|
||||
job.snapshot.sync_total = total;
|
||||
@@ -839,9 +833,9 @@ void run(const std::shared_ptr<Job>& pointer) {
|
||||
// Every volume exists now: the receipt names them as a run after the download would.
|
||||
const auto full = input_identity(request);
|
||||
if (full.empty()) error = "Archive source is missing or not a regular file.";
|
||||
else receipt.replace(0, receipt.find('\n'), "P5AR001:" + full);
|
||||
else receipt.set_header("P5AR001:" + full);
|
||||
}
|
||||
if (error.empty() && !job.cancelled && !write_metadata(staging, receipt_name, receipt)) error = "Cannot save extraction completion receipt.";
|
||||
if (error.empty() && !job.cancelled && !receipt.commit(receipt_name)) error = "Cannot save extraction completion receipt.";
|
||||
if (error.empty() && !job.cancelled && fsync(root)) error = std::strerror(errno);
|
||||
close(root);
|
||||
if (job.cancelled || !error.empty()) {
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
#include <string>
|
||||
#include <vector>
|
||||
namespace archives {
|
||||
inline constexpr std::uint64_t max_entries = 100000;
|
||||
inline constexpr std::uint64_t max_entries = 1000000;
|
||||
inline constexpr std::size_t max_metadata_bytes = 16 * 1024 * 1024;
|
||||
// `stream`: a RAR set whose later volumes are still downloading. Only the first exists yet; each next one is
|
||||
// read once its final name appears.
|
||||
|
||||
@@ -0,0 +1,170 @@
|
||||
// Copyright (C) 2026 half144 and PS5 React contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
// Additional attribution term: see LICENSE-ATTRIBUTION.
|
||||
#pragma once
|
||||
#include <cerrno>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <fcntl.h>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
namespace archives {
|
||||
// Reads a file's lines in order through a small buffer. `next` is false at the end; `clean` then says
|
||||
// whether the file ended on a newline (a last line cut short is not a line).
|
||||
class LineReader {
|
||||
public:
|
||||
explicit LineReader(int fd, std::uint64_t offset = 0) : fd_(fd), offset_(offset) {}
|
||||
bool next(std::string& line) {
|
||||
while (true) {
|
||||
const auto end = buffer_.find('\n', start_);
|
||||
if (end != std::string::npos) {
|
||||
line.assign(buffer_, start_, end - start_);
|
||||
start_ = end + 1;
|
||||
return true;
|
||||
}
|
||||
buffer_.erase(0, start_);
|
||||
start_ = 0;
|
||||
if (buffer_.size() > max_line) { failed_ = true; return false; }
|
||||
char chunk[64 * 1024];
|
||||
ssize_t got;
|
||||
do got = pread(fd_, chunk, sizeof chunk, static_cast<off_t>(offset_)); while (got < 0 && errno == EINTR);
|
||||
if (got < 0) { failed_ = true; return false; }
|
||||
if (got == 0) return false;
|
||||
offset_ += static_cast<std::uint64_t>(got);
|
||||
buffer_.append(chunk, static_cast<std::size_t>(got));
|
||||
}
|
||||
}
|
||||
bool clean() const { return !failed_ && start_ == buffer_.size(); }
|
||||
|
||||
private:
|
||||
static constexpr std::size_t max_line = 64 * 1024;
|
||||
int fd_;
|
||||
std::uint64_t offset_;
|
||||
std::string buffer_;
|
||||
std::size_t start_ = 0;
|
||||
bool failed_ = false;
|
||||
};
|
||||
|
||||
// The extraction receipt: a "P5AR001:<identity>" header, then one "<sha256>|<path>" line per extracted
|
||||
// file. A game folder holds hundreds of thousands of files (issue #52: ASTRO BOT and Resident Evil
|
||||
// Requiem overran the 16 MiB a receipt kept in memory could take), so its lines go to a file in the
|
||||
// staging folder as they come, through a 1 MiB buffer, and the receipt itself is written from that file
|
||||
// at the end. Memory stays the same whatever the number of files. Used from one thread at a time.
|
||||
class Receipt {
|
||||
public:
|
||||
Receipt(std::string directory, std::string header) : directory_(std::move(directory)), header_(std::move(header)) {}
|
||||
~Receipt() {
|
||||
if (fd_ >= 0) close(fd_);
|
||||
if (fd_ >= 0 || spilled_) unlink(spill_path().c_str());
|
||||
}
|
||||
Receipt(const Receipt&) = delete;
|
||||
Receipt& operator=(const Receipt&) = delete;
|
||||
|
||||
// False when the line could not be kept (the disk is full or failing).
|
||||
bool add(const std::string& digest, const std::string& name) {
|
||||
if (failed_) return false;
|
||||
buffer_ += digest;
|
||||
buffer_ += '|';
|
||||
buffer_ += name;
|
||||
buffer_ += '\n';
|
||||
++lines_;
|
||||
return buffer_.size() < buffer_bytes || flush();
|
||||
}
|
||||
std::uint64_t lines() const { return lines_; }
|
||||
void set_header(std::string header) { header_ = std::move(header); }
|
||||
|
||||
// Reads the lines from line `first` on (counted from 0), in order: what one writer added, for its fsyncs.
|
||||
// Valid until the next add().
|
||||
class Lines {
|
||||
public:
|
||||
bool next(std::string& line) {
|
||||
while (skip_) {
|
||||
if (!reader_.next(line)) return false;
|
||||
--skip_;
|
||||
}
|
||||
return reader_.next(line);
|
||||
}
|
||||
private:
|
||||
friend class Receipt;
|
||||
Lines(int fd, std::uint64_t skip) : reader_(fd), skip_(skip) {}
|
||||
LineReader reader_;
|
||||
std::uint64_t skip_;
|
||||
};
|
||||
bool lines_from(std::uint64_t first, Lines*& out) {
|
||||
out = nullptr;
|
||||
if (!flush()) return false;
|
||||
lines_view_.reset(new Lines(fd_, first));
|
||||
out = lines_view_.get();
|
||||
return true;
|
||||
}
|
||||
|
||||
// Writes <directory>/<name>: the header, then every line, synced. The line file goes with it. False
|
||||
// on any failure, leaving no receipt behind.
|
||||
bool commit(const char* name) {
|
||||
if (!flush()) return false;
|
||||
const auto path = directory_ + "/" + name;
|
||||
const int out = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, 0600);
|
||||
if (out < 0) return false;
|
||||
const auto header = header_ + "\n";
|
||||
bool ok = write_all(out, header.data(), header.size());
|
||||
if (ok && fd_ >= 0) {
|
||||
char chunk[256 * 1024];
|
||||
for (off_t at = 0; ok;) {
|
||||
ssize_t got;
|
||||
do got = pread(fd_, chunk, sizeof chunk, at); while (got < 0 && errno == EINTR);
|
||||
if (got < 0) ok = false;
|
||||
if (got <= 0) break;
|
||||
ok = write_all(out, chunk, static_cast<std::size_t>(got));
|
||||
at += got;
|
||||
}
|
||||
}
|
||||
if (ok && fsync(out)) ok = false;
|
||||
if (close(out)) ok = false;
|
||||
if (!ok) {
|
||||
unlink(path.c_str());
|
||||
return false;
|
||||
}
|
||||
if (fd_ >= 0) {
|
||||
close(fd_);
|
||||
fd_ = -1;
|
||||
unlink(spill_path().c_str());
|
||||
spilled_ = false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private:
|
||||
static constexpr std::size_t buffer_bytes = 1024 * 1024;
|
||||
std::string spill_path() const { return directory_ + "/.ps5-react-receipt-lines"; }
|
||||
static bool write_all(int fd, const char* data, std::size_t size) {
|
||||
while (size) {
|
||||
const auto count = write(fd, data, size);
|
||||
if (count < 0 && errno == EINTR) continue;
|
||||
if (count <= 0) return false;
|
||||
data += count;
|
||||
size -= static_cast<std::size_t>(count);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
bool flush() {
|
||||
if (failed_) return false;
|
||||
if (buffer_.empty()) return true;
|
||||
if (fd_ < 0) {
|
||||
fd_ = open(spill_path().c_str(), O_RDWR | O_CREAT | O_EXCL | O_APPEND | O_NOFOLLOW, 0600);
|
||||
if (fd_ < 0) { failed_ = true; return false; }
|
||||
spilled_ = true;
|
||||
}
|
||||
if (!write_all(fd_, buffer_.data(), buffer_.size())) { failed_ = true; return false; }
|
||||
buffer_.clear();
|
||||
return true;
|
||||
}
|
||||
std::string directory_, header_, buffer_;
|
||||
int fd_ = -1;
|
||||
std::uint64_t lines_ = 0;
|
||||
bool failed_ = false, spilled_ = false;
|
||||
std::unique_ptr<Lines> lines_view_;
|
||||
};
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Real local archive extraction with the same native worker used on PS5."""
|
||||
from pathlib import Path
|
||||
import binascii,hashlib,io,json,lzma,os,random,struct,subprocess,sys,tarfile,tempfile,time,zipfile
|
||||
import binascii,hashlib,io,json,lzma,os,random,shutil,struct,subprocess,sys,tarfile,tempfile,time,zipfile
|
||||
from rar_worker import build_host
|
||||
ROOT=Path(__file__).resolve().parents[1]
|
||||
DATA=bytes(range(256))*1024
|
||||
@@ -232,6 +232,22 @@ def main():
|
||||
with tarfile.open(tar,"w") as t:
|
||||
entry=tarfile.TarInfo("nested/file.bin");entry.size=len(DATA);t.addfile(entry,io.BytesIO(DATA))
|
||||
assert run("tar",[tar])[0]=="completed"
|
||||
# A game folder (issue #52): 150,000 files with long paths, whose receipt is far over the 16 MiB a receipt
|
||||
# kept in memory could take. Extracted through the write ring and inline, then recovered by a later run.
|
||||
many=root/"many.tar";deep="Image0/"+"/".join(["sce_module_resources"]*5)
|
||||
with tarfile.open(many,"w") as t:
|
||||
for i in range(150000):
|
||||
entry=tarfile.TarInfo(f"{deep}/d{i//1000}/asset_{i:06d}_with_a_rather_long_name.bin");entry.size=4;t.addfile(entry,io.BytesIO(i.to_bytes(4,"little")))
|
||||
def many_run(name,**env):
|
||||
return subprocess.check_output([str(binary),str(root/name),str(1<<30),str(many)],text=True,timeout=1800,env={**os.environ,**env}).splitlines()
|
||||
for name,env in (("many-ring",{}),("many-inline",{"ARCHIVE_PIPELINE_BYTES":"0"})):
|
||||
out=many_run(name,**env);assert out[0]=="completed",out
|
||||
receipt=root/name/".ps5-react-extraction"
|
||||
assert receipt.stat().st_size>16*1024*1024 and sum(1 for _ in receipt.open("rb"))==150001
|
||||
assert not (root/name/".ps5-react-receipt-lines").exists()
|
||||
assert (root/name/deep/"d149/asset_149999_with_a_rather_long_name.bin").read_bytes()==(149999).to_bytes(4,"little")
|
||||
out=many_run("many-ring");assert out[0]=="completed",out
|
||||
shutil.rmtree(root/"many-ring");shutil.rmtree(root/"many-inline");many.unlink()
|
||||
# rar-extract: images published at the set root, unsafe names and links refused, a missing middle
|
||||
# volume named as such.
|
||||
p=root/"plain.rar";p.write_bytes(plain_rar5("nested/deeper/game.ffpfsc",DATA))
|
||||
|
||||
@@ -40,11 +40,22 @@ std::vector<unsigned char> slurp(const std::string& path) {
|
||||
return out;
|
||||
}
|
||||
|
||||
// Commits `receipt` to a file and returns its lines after the header, the file then removed.
|
||||
std::string committed(archives::Receipt& receipt, const std::string& dir) {
|
||||
assert(receipt.commit("receipt-out"));
|
||||
const auto bytes = slurp(dir + "/receipt-out");
|
||||
unlink((dir + "/receipt-out").c_str());
|
||||
const std::string text(bytes.begin(), bytes.end());
|
||||
assert(text.rfind("P5AR001:test\n", 0) == 0);
|
||||
return text.substr(text.find('\n') + 1);
|
||||
}
|
||||
|
||||
void round_trip(const std::string& dir, std::mt19937& rng, std::size_t ring) {
|
||||
std::atomic<bool> cancelled{false};
|
||||
archives::Timings timings;
|
||||
archives::Writer writer(cancelled, timings);
|
||||
assert(writer.start(ring));
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
assert(writer.start(ring, receipt));
|
||||
const int files = 1 + rng() % 200;
|
||||
std::vector<std::vector<unsigned char>> contents;
|
||||
std::string expected;
|
||||
@@ -68,15 +79,14 @@ void round_trip(const std::string& dir, std::mt19937& rng, std::size_t ring) {
|
||||
expected += sha(data) + "|" + name + "\n";
|
||||
contents.push_back(std::move(data));
|
||||
}
|
||||
std::string receipt;
|
||||
std::atomic<std::size_t> reports{0}, last_total{0}, most{0};
|
||||
const auto error = writer.finish(receipt, [&](std::size_t synced, std::size_t total) {
|
||||
const auto error = writer.finish([&](std::size_t synced, std::size_t total) {
|
||||
++reports;
|
||||
last_total = total;
|
||||
for (auto seen = most.load(); synced > seen && !most.compare_exchange_weak(seen, synced);) {}
|
||||
});
|
||||
if (!error.empty()) std::fprintf(stderr, "error: %s\n", error.c_str());
|
||||
assert(error.empty() && receipt == expected);
|
||||
assert(error.empty() && committed(receipt, dir) == expected);
|
||||
// One report before the fsyncs and one after each.
|
||||
assert(reports == std::size_t(files) + 1 && last_total == std::size_t(files) && most == std::size_t(files));
|
||||
for (int i = 0; i < files; ++i) {
|
||||
@@ -89,15 +99,15 @@ void write_error(const std::string& dir) {
|
||||
std::atomic<bool> cancelled{false};
|
||||
archives::Timings timings;
|
||||
archives::Writer writer(cancelled, timings);
|
||||
assert(writer.start(1 << 20));
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
assert(writer.start(1 << 20, receipt));
|
||||
const int fd = open((dir + "/ro").c_str(), O_RDONLY | O_CREAT, 0600);
|
||||
assert(writer.begin(fd, dir + "/ro", "ro"));
|
||||
std::vector<unsigned char> data(8 << 20, 7);
|
||||
bool accepted = true;
|
||||
for (int i = 0; i < 64 && accepted; ++i) accepted = writer.append(data.data(), data.size(), std::size_t(i) * data.size());
|
||||
std::string receipt;
|
||||
const auto error = accepted && writer.end() ? writer.finish(receipt) : writer.error();
|
||||
assert(!error.empty() && receipt.empty());
|
||||
const auto error = accepted && writer.end() ? writer.finish() : writer.error();
|
||||
assert(!error.empty() && receipt.lines() == 0);
|
||||
unlink((dir + "/ro").c_str());
|
||||
}
|
||||
|
||||
@@ -106,15 +116,15 @@ void scattered(const std::string& dir) {
|
||||
std::atomic<bool> cancelled{false};
|
||||
archives::Timings timings;
|
||||
archives::Writer writer(cancelled, timings);
|
||||
assert(writer.start(1 << 20));
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
assert(writer.start(1 << 20, receipt));
|
||||
std::vector<unsigned char> data(20000);
|
||||
for (std::size_t i = 0; i < data.size(); ++i) data[i] = static_cast<unsigned char>(i * 7);
|
||||
const int fd = open((dir + "/s").c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0600);
|
||||
assert(fd >= 0 && writer.begin(fd, dir + "/s", "s"));
|
||||
for (std::size_t i = data.size(); i-- > 0;) assert(writer.append(&data[i], 1, i));
|
||||
assert(writer.end());
|
||||
std::string receipt;
|
||||
assert(writer.finish(receipt).empty() && receipt == sha(data) + "|s\n" && slurp(dir + "/s") == data);
|
||||
assert(writer.finish().empty() && committed(receipt, dir) == sha(data) + "|s\n" && slurp(dir + "/s") == data);
|
||||
unlink((dir + "/s").c_str());
|
||||
}
|
||||
|
||||
@@ -123,15 +133,15 @@ void sync_missing(const std::string& dir) {
|
||||
std::atomic<bool> cancelled{false};
|
||||
archives::Timings timings;
|
||||
archives::Writer writer(cancelled, timings);
|
||||
assert(writer.start(1 << 20));
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
assert(writer.start(1 << 20, receipt));
|
||||
std::vector<unsigned char> data(4096, 3);
|
||||
for (const char* name : {"m0", "m1", "m2"}) {
|
||||
const int fd = open((dir + "/" + name).c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0600);
|
||||
assert(fd >= 0 && writer.begin(fd, dir + "/" + name, name) && writer.append(data.data(), data.size(), 0) && writer.end());
|
||||
}
|
||||
unlink((dir + "/m1").c_str());
|
||||
std::string receipt;
|
||||
assert(!writer.finish(receipt).empty() && receipt.empty());
|
||||
assert(!writer.finish().empty());
|
||||
unlink((dir + "/m0").c_str());
|
||||
unlink((dir + "/m2").c_str());
|
||||
}
|
||||
@@ -140,7 +150,8 @@ void cancel_midway(const std::string& dir, std::mt19937& rng) {
|
||||
std::atomic<bool> cancelled{false};
|
||||
archives::Timings timings;
|
||||
archives::Writer writer(cancelled, timings);
|
||||
assert(writer.start(1 << 20));
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
assert(writer.start(1 << 20, receipt));
|
||||
std::vector<unsigned char> data(256 * 1024, 1);
|
||||
const int stop_after = rng() % 40;
|
||||
for (int i = 0; i < 40; ++i) {
|
||||
@@ -154,6 +165,44 @@ void cancel_midway(const std::string& dir, std::mt19937& rng) {
|
||||
}
|
||||
}
|
||||
|
||||
// A game folder's receipt (issue #52): 300,000 lines with long names, far over the 16 MiB a receipt kept
|
||||
// in memory could take, go through the line file and come back whole and in order, from any line on.
|
||||
void many_lines(const std::string& dir) {
|
||||
const std::string tail(140, 'x');
|
||||
{
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
for (int i = 0; i < 300000; ++i) assert(receipt.add(std::string(64, 'a' + i % 6), "Image0/data/" + std::to_string(i) + "/" + tail));
|
||||
archives::Receipt::Lines* lines = nullptr;
|
||||
assert(receipt.lines_from(299998, lines));
|
||||
std::string line;
|
||||
assert(lines->next(line) && line == std::string(64, 'a' + 299998 % 6) + "|Image0/data/299998/" + tail);
|
||||
assert(lines->next(line) && !lines->next(line));
|
||||
receipt.set_header("P5AR001:renamed");
|
||||
assert(receipt.commit("receipt-out"));
|
||||
struct stat st;
|
||||
assert(stat((dir + "/.ps5-react-receipt-lines").c_str(), &st) != 0);
|
||||
}
|
||||
const int fd = open((dir + "/receipt-out").c_str(), O_RDONLY);
|
||||
assert(fd >= 0);
|
||||
archives::LineReader reader(fd);
|
||||
std::string line;
|
||||
assert(reader.next(line) && line == "P5AR001:renamed");
|
||||
int count = 0;
|
||||
for (; reader.next(line); ++count)
|
||||
if (count % 50000 == 0) assert(line == std::string(64, 'a' + count % 6) + "|Image0/data/" + std::to_string(count) + "/" + tail);
|
||||
assert(count == 300000 && reader.clean());
|
||||
close(fd);
|
||||
struct stat st;
|
||||
assert(stat((dir + "/receipt-out").c_str(), &st) == 0 && st.st_size > 16 * 1024 * 1024);
|
||||
unlink((dir + "/receipt-out").c_str());
|
||||
// A receipt dropped before commit leaves no line file behind.
|
||||
{
|
||||
archives::Receipt receipt(dir, "P5AR001:test");
|
||||
for (int i = 0; i < 20000; ++i) assert(receipt.add(std::string(64, 'b'), tail));
|
||||
}
|
||||
assert(stat((dir + "/.ps5-react-receipt-lines").c_str(), &st) != 0);
|
||||
}
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
const std::string dir = argc > 1 ? argv[1] : "/tmp/archive-writer-test";
|
||||
mkdir(dir.c_str(), 0700);
|
||||
@@ -163,8 +212,9 @@ int main(int argc, char** argv) {
|
||||
write_error(dir);
|
||||
scattered(dir);
|
||||
sync_missing(dir);
|
||||
many_lines(dir);
|
||||
for (int round = 0; round < 40; ++round) cancel_midway(dir, rng);
|
||||
for (int i = 0; i < 40; ++i) unlink((dir + "/c" + std::to_string(i)).c_str());
|
||||
assert(open_descriptors() == before);
|
||||
std::puts("PASS: writer round trips, write error, cancel without leaks");
|
||||
std::puts("PASS: writer round trips, write error, cancel without leaks, a 300,000-line receipt");
|
||||
}
|
||||
Reference in new issue
Block a user