From 90aaa75abed5d7088be5f911ce2bd1b4af03445f Mon Sep 17 00:00:00 2001 From: Mihawk <163614687+mihawk-99@users.noreply.github.com> Date: Wed, 7 Oct 2026 08:23:21 -0400 Subject: [PATCH] WebUI: a parallel transfer engine (batches, upload sessions, ranges, six lanes) Measured first: one libmicrohttpd thread with 64 KiB a connection held uploads to 13 MB/s, and four connections to 15.3, on a link where FTP reached 25 / 30.5 MB/s. PS5-Suite's design was reviewed (persistent connection pool, small files per connection, large files as pwrite ranges, big socket buffers) and an engine built from scratch on the WebUI's HTTP (src/webui_transfer.{h,cpp}, docs/WEBUI.md): - a pool of four server threads, 512 KiB a connection, 4 MiB socket buffers, writes in 1 MiB blocks; non-transfer routes under one lock as on the single thread; - PUT /api/upload/batch: many small files in one request (a framed stream), each renamed into place when whole, existing names skipped and named unless replace; - upload sessions (/api/upload/session, /part, /commit): a large file's ranges in any order on any number of connections, committed once covered; - Range downloads (206); every upload still atomic, never over an existing name; - the page plans batches, sessions and single uploads onto six lanes. Evidence: evidence/webui-transfer-engine (base PS5 13.40, PC on Wi-Fi 6). Uploads 26.0 MB/s on one connection and 29.9 MB/s for one file in six parts (FTP: 25 / 30.5); 400 x 32 KiB at 215 files/s (FTP 105). The page against the console: 605 files in 20.1 s, all equal by FTP readback, duplicates refused, a cancelled 200 MiB file left nothing. ThreadSanitizer under a mixed parallel load: no report. Command: build/tmp/transfer/bench.py; node tests/webui_transfer_browser.cjs; tools/verify.sh: all five PASS. --- evidence/webui-transfer-engine/capture.json | 37 ++ .../webui-transfer-engine/expectation.json | 15 + src/webui_ps5.cpp | 526 +++++++++++++++--- src/webui_transfer.cpp | 373 +++++++++++++ src/webui_transfer.h | 177 ++++++ tests/test_webui.py | 116 +++- tests/test_webui_daemon.py | 2 +- tests/test_webui_transfer_browser.py | 65 +++ tests/webui_transfer_browser.cjs | 59 ++ tests/webui_update_fixture.py | 2 +- tools/build-webui-daemon.sh | 2 +- webui/app.js | 190 ++++++- 12 files changed, 1468 insertions(+), 96 deletions(-) create mode 100644 evidence/webui-transfer-engine/capture.json create mode 100644 evidence/webui-transfer-engine/expectation.json create mode 100644 src/webui_transfer.cpp create mode 100644 src/webui_transfer.h create mode 100644 tests/test_webui_transfer_browser.py create mode 100644 tests/webui_transfer_browser.cjs diff --git a/evidence/webui-transfer-engine/capture.json b/evidence/webui-transfer-engine/capture.json new file mode 100644 index 0000000..d41d455 --- /dev/null +++ b/evidence/webui-transfer-engine/capture.json @@ -0,0 +1,37 @@ +{ + "step": "webui-transfer-engine", + "recorded": "2026-10-07T12:17:29Z", + "console_run": "base PS5 13.40 over Wi-Fi 6 (PC side), WebUI daemon, picker showing", + "revision": "working tree on 189b415 (this commit)", + "command": "build/tmp/transfer/bench.py 6769 {baseline|engine} ...; FTP ceiling by ftplib; node tests/webui_transfer_browser.cjs against the console + FTP readback", + "raw_capture": "klog/webui-transfer-engine/raw.txt", + "raw_lines": 25, + "raw_bytes": 4228, + "lines": [ + "link: PC on Wi-Fi 6 (rx 816 / tx 864 Mbit/s, 160 MHz), RTT ~10 ms", + "ftp ceiling: upload 1 x 256 MiB 25.0 MB/s; 4 parallel 30.5 MB/s; download 15.9 MB/s; 400 x 32 KiB: 46 files/s (1 conn), 105 files/s (4 conn)", + "before: {\"label\": \"baseline\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 20.59, \"files\": 1, \"MBps\": 13.0, \"files_per_s\": 0.0, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"upload 4 x 256 MiB, 4 connections\", \"seconds\": 70.17, \"files\": 4, \"MBps\": 15.3, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"upload 400 x 32 KiB, 1 keep-alive connection\", \"seconds\": 5.23, \"files\": 400, \"MBps\": 2.5, \"files_per_s\": 76.4, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"upload 400 x 32 KiB, 8 keep-alive connections\", \"seconds\": 3.42, \"files\": 400, \"MBps\": 3.8, \"files_per_s\": 116.9, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"download 400 x 32 KiB, 1 connection\", \"seconds\": 2.43, \"files\": 400, \"MBps\": 5.4, \"files_per_s\": 164.5, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"download 400 x 32 KiB, 8 connections\", \"seconds\": 0.85, \"files\": 400, \"MBps\": 15.4, \"files_per_s\": 468.9, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"download 1 x 256 MiB, 1 connection\", \"seconds\": 17.82, \"files\": 1, \"MBps\": 15.1, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "before: {\"label\": \"baseline\", \"test\": \"download 4 x 256 MiB, 4 connections\", \"seconds\": 43.62, \"files\": 4, \"MBps\": 24.6, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 10.31, \"files\": 1, \"MBps\": 26.0, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"upload 4 x 256 MiB, 4 connections\", \"seconds\": 37.21, \"files\": 4, \"MBps\": 28.9, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"upload 400 x 32 KiB, 1 keep-alive connection\", \"seconds\": 4.49, \"files\": 400, \"MBps\": 2.9, \"files_per_s\": 89.0, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"upload 400 x 32 KiB, 8 keep-alive connections\", \"seconds\": 1.86, \"files\": 400, \"MBps\": 7.1, \"files_per_s\": 215.4, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"download 400 x 32 KiB, 1 connection\", \"seconds\": 1.86, \"files\": 400, \"MBps\": 7.1, \"files_per_s\": 215.5, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"download 400 x 32 KiB, 8 connections\", \"seconds\": 0.69, \"files\": 400, \"MBps\": 18.9, \"files_per_s\": 576.2, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 1 lane(s), fsync=1\", \"seconds\": 3.15, \"files\": 400, \"MBps\": 4.2, \"files_per_s\": 126.8, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 4 lane(s), fsync=1\", \"seconds\": 1.89, \"files\": 400, \"MBps\": 6.9, \"files_per_s\": 212.0, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 1 lane(s), fsync=0\", \"seconds\": 2.89, \"files\": 400, \"MBps\": 4.5, \"files_per_s\": 138.4, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 4 lane(s), fsync=0\", \"seconds\": 2.24, \"files\": 400, \"MBps\": 5.8, \"files_per_s\": 178.3, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB as 8 parts on 6 connections (commit 0.00 s)\", \"seconds\": 8.96, \"files\": 1, \"MBps\": 29.9, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"download 1 x 256 MiB, 1 connection\", \"seconds\": 13.4, \"files\": 1, \"MBps\": 20.0, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "after: {\"label\": \"engine\", \"test\": \"download 4 x 256 MiB, 4 connections\", \"seconds\": 42.04, \"files\": 4, \"MBps\": 25.5, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}", + "page engine on the console: PASS: 605 files (600 small in 4 batches, 3 single, 2 large in 18 parts) in 20.1 s; duplicates refused; cancel drops the session", + "page engine on the console, FTP readback: 605 entries; mismatched [] ; leftovers []" + ] +} diff --git a/evidence/webui-transfer-engine/expectation.json b/evidence/webui-transfer-engine/expectation.json new file mode 100644 index 0000000..b27b183 --- /dev/null +++ b/evidence/webui-transfer-engine/expectation.json @@ -0,0 +1,15 @@ +{ + "step": "webui-transfer-engine", + "note": "Parallel transfer engine (thread pool, 1 MiB blocks, batches, upload sessions, ranges). Large uploads 13 -> 26-30 MB/s (the link's FTP ceiling); small files 76/117 -> 89/215 files/s (2x FTP); the page's engine delivers 605 files whole on the console.", + "must_contain": [ + "\"label\": \"baseline\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 20.59", + "\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 10.31", + "as 8 parts on 6 connections", + "\"files_per_s\": 215.4", + "605 entries; mismatched [] ; leftovers []" + ], + "must_not_contain": [ + "\"errors\": 1" + ], + "human_check": "none" +} diff --git a/src/webui_ps5.cpp b/src/webui_ps5.cpp index d27a966..b54ffe1 100644 --- a/src/webui_ps5.cpp +++ b/src/webui_ps5.cpp @@ -4,6 +4,7 @@ */ #include "webui_ps5.h" #include "webui_update.h" +#include "webui_transfer.h" #include "ps5_frontend_choice.h" #include "../vendor/retroarch/libretro-common/include/libretro.h" #include @@ -19,7 +20,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -55,20 +58,32 @@ const Setting settings[] = { {"menu_show_advanced_settings", "Show advanced settings", "bool", "false", 0, 0}, {"menu_driver", "Console menu", "menu", "xmb", 0, 0}, }; +// The server answers on a pool of threads (ps5_webui_start). Transfers write on +// their own; everything else (settings, folders, the updater) runs under this lock, +// as it did on the one thread before. +std::mutex state_lock; +ps5_transfer::Sessions sessions; +constexpr std::time_t session_idle_seconds = 15 * 60; +struct BatchState; +enum class Transfer +{ + none, + single, // PUT /api/upload + batch, // PUT /api/upload/batch + part // PUT /api/upload/part +}; struct Request { - int file = -1; + Transfer transfer = Transfer::none; + ps5_transfer::Writer writer; + std::unique_ptr batch; + std::shared_ptr session; + bool replace = false; std::string temporary, destination, body; - uint64_t received = 0, expected = 0; + uint64_t received = 0, expected = 0, offset = 0; unsigned error = 0; const char *message = ""; - ~Request() - { - if (file >= 0) - close(file); - if (!temporary.empty()) - unlink(temporary.c_str()); - } + ~Request(); }; std::string quote(const std::string &text) { @@ -242,6 +257,39 @@ bool content_path(const std::string &relative, std::string &absolute, bool new_l } return true; } +// A path below content/ for a new file, its missing folders created on the way +// (an upload of a folder's files, a batch's subfolders). The leaf is not checked. +bool content_parents(const std::string &relative, std::string &absolute) +{ + if (relative.empty() || !valid_path(relative)) + return false; + absolute = root_path + "/content"; + struct stat st{}; + if (content_stat(absolute.c_str(), &st) || !S_ISDIR(st.st_mode)) + return false; + size_t start = 0; + for (;;) + { + const auto end = relative.find('/', start); + absolute += '/' + relative.substr(start, end - start); + if (end == std::string::npos) + return true; + if (content_stat(absolute.c_str(), &st)) + { + if (errno != ENOENT || (mkdir(absolute.c_str(), 0755) && errno != EEXIST) || + content_stat(absolute.c_str(), &st)) + return false; + } + if (S_ISLNK(st.st_mode) || !S_ISDIR(st.st_mode)) + return false; + start = end + 1; + } +} +bool exists(const std::string &absolute) +{ + struct stat st{}; + return content_stat(absolute.c_str(), &st) == 0 || errno != ENOENT; +} std::string trim(std::string s) { auto first = s.find_first_not_of(" \t\r\n"); @@ -791,17 +839,25 @@ MHD_Result list_content(MHD_Connection *c) // A download is read in blocks rather than handed to MHD as a descriptor: MHD sends a // descriptor with sendfile, which on the console stopped after its first 33 KB // (the daemon, 2026-10-07: curl got 33,300 of 1.9 GB and a broken transfer). +struct DownloadSource +{ + int fd; + uint64_t start; +}; ssize_t read_download(void *cls, uint64_t position, char *buffer, size_t size) { - const ssize_t got = pread(static_cast(reinterpret_cast(cls)), buffer, size, - static_cast(position)); + const auto *source = static_cast(cls); + const ssize_t got = + pread(source->fd, buffer, size, static_cast(source->start + position)); if (got == 0) return MHD_CONTENT_READER_END_OF_STREAM; return got < 0 ? MHD_CONTENT_READER_END_WITH_ERROR : got; } void close_download(void *cls) { - close(static_cast(reinterpret_cast(cls))); + auto *source = static_cast(cls); + close(source->fd); + delete source; } MHD_Result download(MHD_Connection *c) { @@ -817,24 +873,165 @@ MHD_Result download(MHD_Connection *c) close(fd); return error(c, 400, "Choose a file to download."); } - auto *response = MHD_create_response_from_callback( - static_cast(st.st_size), 256 * 1024, read_download, - reinterpret_cast(static_cast(fd)), close_download); - if (!response) + // One byte range (Range: bytes=a-b), so a client can fetch a large file on several + // connections or resume one. + const uint64_t size = static_cast(st.st_size); + uint64_t start = 0, end = size; + const char *range = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Range"); + const bool partial = range && ps5_transfer::parse_range(range, size, start, end); + if (range && !partial && size) { + close(fd); + return error(c, 416, "This part of the file does not exist."); + } + auto *source = new (std::nothrow) DownloadSource{fd, start}; + auto *response = source ? MHD_create_response_from_callback( + end - start, 1024 * 1024, read_download, source, close_download) + : nullptr; + if (!response) + { + delete source; close(fd); return MHD_NO; } + MHD_add_response_header(response, "Accept-Ranges", "bytes"); + if (partial) + { + const std::string content_range = "bytes " + std::to_string(start) + '-' + + std::to_string(end - 1) + '/' + std::to_string(size); + MHD_add_response_header(response, "Content-Range", content_range.c_str()); + } const std::string disposition = "attachment; filename=" + quote(path.substr(path.find_last_of('/') + 1)); MHD_add_response_header(response, "Content-Disposition", disposition.c_str()); MHD_add_response_header(response, "Content-Type", "application/octet-stream"); MHD_add_response_header(response, "X-Content-Type-Options", "nosniff"); MHD_add_response_header(response, "Cache-Control", "no-store"); - auto result = MHD_queue_response(c, 200, response); + auto result = MHD_queue_response(c, partial ? 206 : 200, response); MHD_destroy_response(response); return result; } +// Many small files in one request (src/webui_transfer.h, the batch stream): each is +// written under a temporary name and renamed when whole; one that exists is skipped +// unless replace. Results are counted, not answered one by one. +struct BatchUpload final : ps5_transfer::BatchSink +{ + std::string folder; // below content/, may be empty + bool replace = false, sync = true; + ps5_transfer::Writer writer; + std::string name, temporary, destination; + uint64_t size = 0; + unsigned written = 0, skipped = 0; + uint64_t bytes = 0; + std::vector> failed; + std::vector skipped_names; // the first 1024 + void skip() + { + ++skipped; + if (skipped_names.size() < 1024) + skipped_names.push_back(name); + } + bool fail(const char *why) + { + if (failed.size() < 64) + failed.emplace_back(name, why); + else + failed.back().second = "and more"; + const int fd = writer.release(); + if (fd >= 0) + close(fd); + if (!temporary.empty()) + unlink(temporary.c_str()); + temporary.clear(); + return false; + } + bool begin(const std::string &record, uint64_t record_size) override + { + name = record; + size = record_size; + temporary.clear(); + if (record_size > upload_limit || + !content_parents(folder.empty() ? record : folder + '/' + record, destination)) + return fail("invalid name"); + if (!replace && exists(destination)) + { + skip(); + return false; + } + temporary = destination.substr(0, destination.find_last_of('/') + 1) + ".upload-" + nonce(); + return writer.create(temporary) || fail("cannot create"); + } + bool data(const char *bytes_in, size_t count) override + { + return writer.write(bytes_in, count) || fail("cannot write (storage full?)"); + } + void end() override + { + const bool flushed = writer.flush(); + const uint64_t got = writer.written(); + const int fd = writer.release(); + if (!flushed || fd < 0 || got != size) + { + if (fd >= 0) + close(fd); + fail("incomplete"); + return; + } + const auto result = ps5_transfer::commit(fd, temporary, destination, replace, sync); + close(fd); + if (result == ps5_transfer::Commit::done) + { + temporary.clear(); + ++written; + bytes += size; + } + else if (result == ps5_transfer::Commit::exists) + { + unlink(temporary.c_str()); + temporary.clear(); + skip(); + } + else + fail("cannot finish"); + } + ~BatchUpload() override + { + const int fd = writer.release(); + if (fd >= 0) + close(fd); + if (!temporary.empty()) + unlink(temporary.c_str()); + } +}; +struct BatchState +{ + BatchUpload upload; + ps5_transfer::BatchReader reader{upload}; +}; +Request::~Request() +{ + if (transfer == Transfer::part && session) + { + std::lock_guard guard(session->lock); + if (--session->writers == 0 && session->closing && session->fd >= 0) + { + close(session->fd); + session->fd = -1; + } + } + if (!temporary.empty()) + unlink(temporary.c_str()); +} +bool content_length(MHD_Connection *c, uint64_t &length) +{ + const char *value = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Content-Length"); + char *end = nullptr; + errno = 0; + if (!value || !*value || *value == '-') + return false; + length = std::strtoull(value, &end, 10); + return !errno && !*end; +} void prepare_upload(MHD_Connection *c, Request &r) { auto fail = [&](unsigned code, const char *message) @@ -848,8 +1045,9 @@ void prepare_upload(MHD_Connection *c, Request &r) fail(400, "Choose a valid content folder and filename."); return; } - struct stat st{}; - if (content_stat(r.destination.c_str(), &st) == 0 || errno != ENOENT) + r.transfer = Transfer::single; + r.replace = std::strcmp(arg(c, "existing"), "replace") == 0; + if (!r.replace && exists(r.destination)) { fail(409, "A file with this name already exists. Rename your file first."); return; @@ -876,12 +1074,205 @@ void prepare_upload(MHD_Connection *c, Request &r) } r.temporary = r.destination.substr(0, r.destination.find_last_of('/') + 1) + ".upload-" + nonce(); - r.file = open(r.temporary.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, 0644); - if (r.file < 0) + if (!r.writer.create(r.temporary)) { + r.temporary.clear(); fail(507, "Could not create the upload. Check console storage."); + } +} +void prepare_batch(MHD_Connection *c, Request &r) +{ + r.transfer = Transfer::batch; + if (!content_length(c, r.expected) || r.expected > upload_limit) + { + r.error = 411; + r.message = "A batch needs its size, at most 64 GiB."; return; } + auto state = std::make_unique(); + state->upload.folder = arg(c, "folder"); + state->upload.replace = std::strcmp(arg(c, "existing"), "replace") == 0; + state->upload.sync = std::strcmp(arg(c, "sync"), "0") != 0; + if (!state->upload.folder.empty() && !valid_path(state->upload.folder)) + { + r.error = 400; + r.message = "Choose a valid content folder."; + return; + } + r.batch = std::move(state); +} +void prepare_part(MHD_Connection *c, Request &r) +{ + r.transfer = Transfer::part; + r.session = sessions.find(arg(c, "id")); + char *end = nullptr; + const char *offset = arg(c, "offset"); + errno = 0; + r.offset = std::strtoull(offset, &end, 10); + if (!r.session) + { + r.error = 404; + r.message = "This upload is no longer open. Start it again."; + return; + } + if (!*offset || errno || *end || !content_length(c, r.expected) || r.offset > r.session->size || + r.expected > r.session->size - r.offset) + { + r.session.reset(); + r.error = 416; + r.message = "This part does not fit the file."; + return; + } + std::lock_guard guard(r.session->lock); + if (r.session->closing || r.session->fd < 0) + { + r.session.reset(); + r.error = 409; + r.message = "This upload was closed."; + return; + } + ++r.session->writers; + r.session->touched = std::time(nullptr); + r.writer.attach(r.session->fd, r.offset); +} +std::string names(const std::vector &list) +{ + std::string out = "["; + for (const auto &name : list) + out += (out.size() > 1 ? "," : "") + quote(name); + return out + ']'; +} +// The transfers' own routes: they run beside each other, outside state_lock. +MHD_Result transfer_route(MHD_Connection *c, const std::string &url, const std::string &method, + Request &r) +{ + if (r.transfer == Transfer::single) + { + if (r.received != r.expected) + return error(c, 400, "The upload was incomplete. Try again."); + const int fd = r.writer.release(); + if (fd < 0) + return error(c, 507, "Could not finish writing the file. Check console storage."); + const auto result = ps5_transfer::commit(fd, r.temporary, r.destination, r.replace, true); + close(fd); + if (result == ps5_transfer::Commit::exists) + return error(c, 409, "A file with this name already exists. Rename your file first."); + if (result != ps5_transfer::Commit::done) + return error(c, 500, "Could not finish the upload."); + r.temporary.clear(); + return respond(c, 201, "{\"uploaded\":true,\"bytes\":" + std::to_string(r.received) + '}'); + } + if (r.transfer == Transfer::batch) + { + auto *state = r.batch.get(); + auto &b = state->upload; + std::string failed = "["; + for (const auto &f : b.failed) + failed += (failed.size() > 1 ? "," : "") + std::string("{\"name\":") + quote(f.first) + + ",\"error\":" + quote(f.second) + '}'; + const bool whole = state->reader.finished() && r.received == r.expected; + return respond(c, whole ? 200 : 400, + "{\"complete\":" + std::string(whole ? "true" : "false") + ",\"written\":" + + std::to_string(b.written) + ",\"skipped\":" + std::to_string(b.skipped) + + ",\"skippedNames\":" + names(b.skipped_names) + ",\"bytes\":" + + std::to_string(b.bytes) + ",\"failed\":" + failed + "],\"problem\":" + + quote(whole ? "" + : *state->reader.problem() ? state->reader.problem() + : "The batch ended early.") + + '}'); + } + if (r.transfer == Transfer::part) + { + if (r.received != r.expected || !r.writer.flush()) + return error(c, 507, "The part could not be written. Try it again."); + uint64_t covered; + { + // Released here, not when the request is freed after its answer: a commit + // sent the moment this answer arrives must find no writer left. + std::lock_guard guard(r.session->lock); + r.session->add(r.offset, r.offset + r.received); + r.session->touched = std::time(nullptr); + covered = r.session->covered; + if (--r.session->writers == 0 && r.session->closing && r.session->fd >= 0) + { + close(r.session->fd); + r.session->fd = -1; + } + } + r.session.reset(); + return respond(c, 200, "{\"covered\":" + std::to_string(covered) + '}'); + } + if (method == "POST" && url == "/api/upload/session") + { + sessions.expire(std::time(nullptr), session_idle_seconds); + std::string destination; + char *end = nullptr; + const char *text = arg(c, "size"); + errno = 0; + const uint64_t size = std::strtoull(text, &end, 10); + const bool size_ok = *text && !errno && !*end && size <= upload_limit; + const bool replace = std::strcmp(arg(c, "existing"), "replace") == 0; + if (!size_ok) + return error(c, 413, "Files must be 64 GiB or smaller."); + if (!content_parents(arg(c, "path"), destination)) + return error(c, 400, "Choose a valid content folder and filename."); + if (!replace && exists(destination)) + return error(c, 409, "A file with this name already exists. Rename your file first."); + struct statvfs storage{}; + if (storage_space(storage) && size > uint64_t(storage.f_bavail) * storage.f_frsize) + return error(c, 507, "There is not enough free space on the console."); + const std::string id = nonce(); + const char *why = ""; + auto session = sessions.open( + id, destination.substr(0, destination.find_last_of('/') + 1) + ".upload-" + id, + destination, size, replace, why); + if (!session) + return error(c, 507, why); + return respond(c, 201, + "{\"id\":" + quote(id) + ",\"size\":" + std::to_string(size) + + ",\"partSize\":" + std::to_string(32u << 20) + '}'); + } + if (method == "POST" && url == "/api/upload/commit") + { + auto session = sessions.find(arg(c, "id")); + if (!session) + return error(c, 404, "This upload is no longer open. Start it again."); + { + std::lock_guard guard(session->lock); + if (session->writers || session->covered != session->size) + return respond(c, 409, + "{\"error\":\"Parts are missing.\",\"covered\":" + + std::to_string(session->covered) + '}'); + } + if (sessions.take(session->id) != session) + return error(c, 409, "This upload was already finished."); + const auto result = ps5_transfer::commit(session->fd, session->temporary, + session->destination, session->replace, true); + close(session->fd); + session->fd = -1; + if (result != ps5_transfer::Commit::done) + { + unlink(session->temporary.c_str()); + return result == ps5_transfer::Commit::exists + ? error(c, 409, + "A file with this name already exists. Rename your file first.") + : error(c, 500, "Could not finish the upload."); + } + return respond(c, 201, + "{\"uploaded\":true,\"bytes\":" + std::to_string(session->size) + '}'); + } + if (method == "DELETE" && url == "/api/upload/session") + { + auto session = sessions.take(arg(c, "id")); + if (session) + ps5_transfer::discard(*session); + return respond(c, 200, "{\"cancelled\":" + std::string(session ? "true" : "false") + '}'); + } + return error(c, 404, "This page was not found."); +} +bool is_transfer_route(const std::string &url) +{ + return url.rfind("/api/upload", 0) == 0; } std::string current_frontend() { @@ -961,26 +1352,6 @@ MHD_Result route(MHD_Connection *c, const std::string &url, const std::string &m "This folder already exists or could not be created."); return respond(c, 201, "{\"created\":true}"); } - if (method == "PUT" && url == "/api/upload") - { - if (r.error) - return error(c, r.error, r.message); - if (r.received != r.expected) - return error(c, 400, "The upload was incomplete. Try again."); - if (fsync(r.file)) - return error(c, 507, "Could not finish writing the file. Check console storage."); - close(r.file); - r.file = -1; - struct stat st{}; - // All HTTP handlers run on one MHD thread; concurrent uploads cannot replace a completed - // file. - if (content_stat(r.destination.c_str(), &st) == 0 || errno != ENOENT) - return error(c, 409, "A file with this name already exists. Rename your file first."); - if (rename(r.temporary.c_str(), r.destination.c_str())) - return error(c, 500, "Could not finish the upload."); - r.temporary.clear(); - return respond(c, 201, "{\"uploaded\":true,\"bytes\":" + std::to_string(r.received) + '}'); - } if (method != "GET") return error(c, 405, "This action is not supported."); // Only shipped assets are reachable; no filesystem passthrough. @@ -1019,12 +1390,14 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method if (write_request && (!provided || token != provided)) return error(c, 403, "Your session expired. Reload the page and try again."); if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload") == 0) - { prepare_upload(c, *r); - if (r->error) - return error(c, r->error, r->message); - } - else + else if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload/batch") == 0) + prepare_batch(c, *r); + else if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload/part") == 0) + prepare_part(c, *r); + if (r->error) + return error(c, r->error, r->message); + if (r->transfer == Transfer::none) { const char *length = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Content-Length"); if (length && std::strtoull(length, nullptr, 10) > 16384) @@ -1035,31 +1408,29 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method auto &r = *static_cast(*context); if (*size) { - if (r.file >= 0 && !r.error) + if (r.transfer != Transfer::none && !r.error) { if (*size > r.expected - r.received) { r.error = 413; - r.message = "Upload exceeded its declared file size."; + r.message = "Upload exceeded its declared size."; } - else + else if (r.transfer == Transfer::batch) { - size_t offset = 0; - while (offset < *size) + r.received += *size; + if (!r.batch->reader.feed(data, *size)) { - ssize_t n = write(r.file, data + offset, *size - offset); - if (n < 0 && errno == EINTR) - continue; - if (n <= 0) - { - r.error = 507; - r.message = "The console could not write the upload. Check free space."; - break; - } - offset += size_t(n); - r.received += uint64_t(n); + r.error = 400; + r.message = r.batch->reader.problem(); } } + else if (!r.writer.write(data, *size)) + { + r.error = 507; + r.message = "The console could not write the upload. Check free space."; + } + else + r.received += *size; } else if (!r.error) { @@ -1076,10 +1447,26 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method // never during a body callback. Drain failed streams without buffering. return MHD_YES; } - if (r.error) + if (r.error && !(r.transfer == Transfer::batch && r.batch && r.error == 400)) return error(c, r.error, r.message); + if (r.transfer != Transfer::none || is_transfer_route(url)) + return transfer_route(c, url, method, r); + std::lock_guard guard(state_lock); return route(c, url, method, r); } +constexpr unsigned transfer_threads = 4; +// Large socket buffers for each connection: a lane keeps the link full on its own. +void tune_connection(void *, MHD_Connection *c, void **, MHD_ConnectionNotificationCode code) +{ + if (code != MHD_CONNECTION_NOTIFY_STARTED) + return; + const auto *info = MHD_get_connection_info(c, MHD_CONNECTION_INFO_CONNECTION_FD); + if (!info) + return; + int buffer = 4 << 20; + setsockopt(info->connect_fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof buffer); + setsockopt(info->connect_fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof buffer); +} void completed(void *, MHD_Connection *, void **context, MHD_RequestTerminationCode) { delete static_cast(*context); @@ -1233,12 +1620,16 @@ bool ps5_webui_start(const char *root, unsigned short port, bool apply_settings) listen_port = port; token = nonce(); mkdir((root_path + "/content").c_str(), 0755); + // A pool of threads, so parallel connections progress side by side (one thread + // held a 4-connection upload to 15 MB/s, 2026-10-07); 512 KiB a connection, so a + // body arrives in large pieces; up to 32 connections for the page's lanes. web_daemon = MHD_start_daemon( MHD_USE_INTERNAL_POLLING_THREAD | MHD_USE_ITC | MHD_USE_ERROR_LOG, port, nullptr, nullptr, - handle, nullptr, MHD_OPTION_CONNECTION_LIMIT, unsigned(8), MHD_OPTION_CONNECTION_TIMEOUT, - unsigned(30), MHD_OPTION_CONNECTION_MEMORY_LIMIT, size_t(65536), - MHD_OPTION_THREAD_STACK_SIZE, size_t(256 * 1024), MHD_OPTION_NOTIFY_COMPLETED, completed, - nullptr, MHD_OPTION_END); + handle, nullptr, MHD_OPTION_THREAD_POOL_SIZE, unsigned(transfer_threads), + MHD_OPTION_CONNECTION_LIMIT, unsigned(32), MHD_OPTION_CONNECTION_TIMEOUT, unsigned(30), + MHD_OPTION_CONNECTION_MEMORY_LIMIT, size_t(512 * 1024), MHD_OPTION_THREAD_STACK_SIZE, + size_t(256 * 1024), MHD_OPTION_NOTIFY_COMPLETED, completed, nullptr, + MHD_OPTION_NOTIFY_CONNECTION, tune_connection, nullptr, MHD_OPTION_END); std::fprintf(stderr, "webui: %s port=%u\n", web_daemon ? "listening" : "unavailable", port); return web_daemon != nullptr; } @@ -1250,5 +1641,6 @@ void ps5_webui_stop() web_daemon = nullptr; std::fprintf(stderr, "webui: stopped\n"); } + sessions.clear(); ps5_update::stop(); } diff --git a/src/webui_transfer.cpp b/src/webui_transfer.cpp new file mode 100644 index 0000000..403dd49 --- /dev/null +++ b/src/webui_transfer.cpp @@ -0,0 +1,373 @@ +/* Copyright (C) 2026 Mihawk; SPDX-License-Identifier: GPL-3.0-or-later */ +/* The WebUI's transfer engine (src/webui_transfer.h says what each part is for). */ +#include "webui_transfer.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ps5_transfer +{ +Writer::~Writer() +{ + if (owned_ && fd_ >= 0) + close(fd_); +} + +bool Writer::create(const std::string &path) +{ + fd_ = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, 0644); + owned_ = fd_ >= 0; + offset_ = written_ = 0; + return owned_; +} + +void Writer::attach(int fd, uint64_t offset) +{ + fd_ = fd; + owned_ = false; + offset_ = offset; + written_ = 0; +} + +bool Writer::write(const char *data, size_t size) +{ + if (fd_ < 0) + return false; + if (buffer_.empty()) + buffer_.resize(write_block); + while (size) + { + /* A piece as large as the block goes straight to the file. */ + if (used_ == 0 && size >= write_block) + { + const size_t whole = size - size % write_block; + const ssize_t n = pwrite(fd_, data, whole, static_cast(offset_ + written_)); + if (n < 0 && errno == EINTR) + continue; + if (n <= 0) + return false; + written_ += uint64_t(n); + data += n; + size -= size_t(n); + continue; + } + const size_t take = std::min(size, write_block - used_); + std::memcpy(buffer_.data() + used_, data, take); + used_ += take; + data += take; + size -= take; + if (used_ == write_block && !flush()) + return false; + } + return true; +} + +bool Writer::flush() +{ + size_t done = 0; + while (done < used_) + { + const ssize_t n = pwrite(fd_, buffer_.data() + done, used_ - done, + static_cast(offset_ + written_)); + if (n < 0 && errno == EINTR) + continue; + if (n <= 0) + return false; + done += size_t(n); + written_ += uint64_t(n); + } + used_ = 0; + return true; +} + +int Writer::release() +{ + const int fd = flush() ? fd_ : -1; + if (fd < 0 && owned_ && fd_ >= 0) + close(fd_); + fd_ = -1; + owned_ = false; + return fd; +} + +namespace +{ +std::mutex commit_lock; +} + +Commit commit(int data_fd, const std::string &temporary, const std::string &destination, + bool replace, bool sync) +{ + if (data_fd >= 0 && sync && fsync(data_fd)) + return Commit::failed; + std::lock_guard guard(commit_lock); + if (!replace) + { + /* Opened, not stat'ed: a title cannot use lstat reliably (src/webui_ps5.cpp). */ + const int existing = open(destination.c_str(), O_RDONLY | O_NOFOLLOW | O_NONBLOCK); + if (existing >= 0 || errno != ENOENT) + { + if (existing >= 0) + close(existing); + return Commit::exists; + } + } + return rename(temporary.c_str(), destination.c_str()) == 0 ? Commit::done : Commit::failed; +} + +bool BatchReader::fail(const char *why) +{ + if (accepted_) + sink_.end(); + accepted_ = false; + state_ = State::bad; + problem_ = why; + return false; +} + +bool BatchReader::feed(const char *data, size_t size) +{ + while (size || (state_ == State::data && remaining_ == 0)) + { + switch (state_) + { + case State::bad: + return false; + case State::done: + return size == 0 || fail("The batch has data after its end."); + case State::name_length: + case State::size: + { + const size_t want = state_ == State::name_length ? 2 : 8; + const size_t take = std::min(size, want - have_); + std::memcpy(header_ + have_, data, take); + have_ += take; + data += take; + size -= take; + if (have_ < want) + break; + have_ = 0; + if (state_ == State::name_length) + { + name_length_ = size_t(header_[0]) | size_t(header_[1]) << 8; + if (name_length_ == 0) + state_ = State::done; + else if (name_length_ > 1024) + return fail("A file name in the batch is too long."); + else + { + name_.clear(); + state_ = State::name; + } + } + else + { + remaining_ = 0; + for (int i = 7; i >= 0; --i) + remaining_ = remaining_ << 8 | header_[i]; + accepted_ = sink_.begin(name_, remaining_); + ++records_; + state_ = State::data; + } + break; + } + case State::name: + { + const size_t take = std::min(size, name_length_ - name_.size()); + name_.append(data, take); + data += take; + size -= take; + if (name_.size() == name_length_) + state_ = State::size; + break; + } + case State::data: + { + const size_t take = size_t(std::min(size, remaining_)); + if (accepted_ && take && !sink_.data(data, take)) + accepted_ = false; /* the sink keeps why; the rest is drained */ + data += take; + size -= take; + remaining_ -= take; + if (remaining_ == 0) + { + if (accepted_) + sink_.end(); + accepted_ = false; + state_ = State::name_length; + } + break; + } + } + } + return state_ != State::bad; +} + +void Session::add(uint64_t start, uint64_t end) +{ + if (start >= end) + return; + auto next = ranges.upper_bound(start); + if (next != ranges.begin()) + { + auto previous = std::prev(next); + if (previous->second >= start) + { + start = previous->first; + end = std::max(end, previous->second); + covered -= previous->second - previous->first; + ranges.erase(previous); + } + } + next = ranges.lower_bound(start); + while (next != ranges.end() && next->first <= end) + { + end = std::max(end, next->second); + covered -= next->second - next->first; + next = ranges.erase(next); + } + ranges[start] = end; + covered += end - start; +} + +std::shared_ptr Sessions::open(const std::string &id, const std::string &temporary, + const std::string &destination, uint64_t size, bool replace, + const char *&why) +{ + std::lock_guard guard(lock_); + if (sessions_.size() >= limit) + { + why = "Too many large uploads at once. Wait for one to finish."; + return nullptr; + } + auto session = std::make_shared(); + session->id = id; + session->temporary = temporary; + session->destination = destination; + session->size = size; + session->replace = replace; + session->touched = std::time(nullptr); + session->fd = ::open(temporary.c_str(), O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW, 0644); + if (session->fd < 0) + { + why = "Could not create the upload. Check console storage."; + return nullptr; + } + /* Sized now, so ranges land anywhere; sparse until written. */ + if (ftruncate(session->fd, static_cast(size))) + { + close(session->fd); + unlink(temporary.c_str()); + why = "There is not enough free space on the console."; + return nullptr; + } + sessions_[id] = session; + return session; +} + +std::shared_ptr Sessions::find(const std::string &id) +{ + std::lock_guard guard(lock_); + auto it = sessions_.find(id); + return it == sessions_.end() ? nullptr : it->second; +} + +std::shared_ptr Sessions::take(const std::string &id) +{ + std::lock_guard guard(lock_); + auto it = sessions_.find(id); + if (it == sessions_.end()) + return nullptr; + auto session = it->second; + sessions_.erase(it); + return session; +} + +void discard(Session &session) +{ + std::lock_guard guard(session.lock); + session.closing = true; + if (session.writers == 0 && session.fd >= 0) + { + close(session.fd); + session.fd = -1; + } + unlink(session.temporary.c_str()); +} + +void Sessions::expire(std::time_t now, std::time_t seconds) +{ + std::vector> stale; + { + std::lock_guard guard(lock_); + for (auto it = sessions_.begin(); it != sessions_.end();) + { + std::lock_guard session_guard(it->second->lock); + if (it->second->writers == 0 && now - it->second->touched > seconds) + { + stale.push_back(it->second); + it = sessions_.erase(it); + } + else + ++it; + } + } + for (auto &session : stale) + discard(*session); +} + +void Sessions::clear() +{ + std::map> all; + { + std::lock_guard guard(lock_); + all.swap(sessions_); + } + for (auto &entry : all) + discard(*entry.second); +} + +size_t Sessions::count() +{ + std::lock_guard guard(lock_); + return sessions_.size(); +} + +bool parse_range(const char *header, uint64_t size, uint64_t &start, uint64_t &end) +{ + if (!header || std::strncmp(header, "bytes=", 6) != 0 || size == 0) + return false; + const char *p = header + 6; + char *after = nullptr; + if (*p == '-') + { + /* The last n bytes. */ + const uint64_t last = std::strtoull(p + 1, &after, 10); + if (after == p + 1 || *after || last == 0) + return false; + start = last >= size ? 0 : size - last; + end = size; + return true; + } + start = std::strtoull(p, &after, 10); + if (after == p || *after != '-' || start >= size) + return false; + p = after + 1; + if (*p == '\0') + { + end = size; + return true; + } + const uint64_t last = std::strtoull(p, &after, 10); + if (*after || last < start) + return false; + end = std::min(last + 1, size); + return true; +} +} // namespace ps5_transfer diff --git a/src/webui_transfer.h b/src/webui_transfer.h new file mode 100644 index 0000000..cb9a05d --- /dev/null +++ b/src/webui_transfer.h @@ -0,0 +1,177 @@ +/* Copyright (C) 2026 Mihawk; SPDX-License-Identifier: GPL-3.0-or-later */ +/* The WebUI's transfer engine, apart from HTTP (src/webui_ps5.cpp wires it). + * + * Three ways in, all ending the same way: a file is written under a hidden + * temporary name beside its destination and renamed into place only when every + * byte is there, so a transfer cut short never leaves a partial file behind. + * + * - one file, one request (PUT /api/upload), written in 1 MiB blocks; + * - many small files, one request (PUT /api/upload/batch): a stream of records, + * so thousands of files cost one round trip per batch instead of one per file; + * - one large file, several requests at once (an upload session): byte ranges + * arrive in any order on as many connections as the client opens, each written + * at its own offset, and the file is committed once its ranges cover it. + * + * The server answers on several threads (a pool), so these are thread-safe: a + * session's ranges under its own lock, renames under one commit lock. */ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ps5_transfer +{ +constexpr size_t write_block = 1u << 20; + +/* Writes a stream through a block buffer, from a start offset (pwrite), so one file + * can take several writers at different offsets. */ +class Writer +{ + public: + Writer() = default; + Writer(const Writer &) = delete; + Writer &operator=(const Writer &) = delete; + ~Writer(); + /* A new file, created exclusive; the writer owns and closes it. */ + bool create(const std::string &path); + /* Someone else's descriptor (a session's), from offset. */ + void attach(int fd, uint64_t offset); + bool write(const char *data, size_t size); + bool flush(); + int fd() const + { + return fd_; + } + uint64_t written() const + { + return written_; + } + /* Flushes, then gives the descriptor up (the caller closes an owned one). */ + int release(); + + private: + int fd_ = -1; + bool owned_ = false; + uint64_t offset_ = 0, written_ = 0; + std::vector buffer_; + size_t used_ = 0; +}; + +/* Moves a finished temporary file to its destination: never over an existing file + * unless replace. Every rename goes through one lock, so two writers of the same + * name cannot both pass the existence check. data_fd (or -1) is synced first. */ +enum class Commit +{ + done, + exists, + failed +}; +Commit commit(int data_fd, const std::string &temporary, const std::string &destination, + bool replace, bool sync); + +/* The batch stream: records of + * u16 name length (little-endian), name (a path below the batch folder), + * u64 size, size bytes of data, + * ended by a name length of 0. The reader takes the body in whatever pieces it + * arrives in and calls the sink at each record's start, data and end. */ +class BatchSink +{ + public: + virtual ~BatchSink() = default; + /* A record starts: false skips its data (the reason is the sink's to keep). */ + virtual bool begin(const std::string &name, uint64_t size) = 0; + virtual bool data(const char *bytes, size_t size) = 0; + virtual void end() = 0; +}; +class BatchReader +{ + public: + explicit BatchReader(BatchSink &sink) : sink_(sink) + { + } + /* false once the stream is malformed (an empty or oversized name, data past the + * end marker); the files before it stand. */ + bool feed(const char *data, size_t size); + bool finished() const + { + return state_ == State::done; + } + const char *problem() const + { + return problem_; + } + unsigned records() const + { + return records_; + } + + private: + enum class State + { + name_length, + name, + size, + data, + done, + bad + }; + BatchSink &sink_; + State state_ = State::name_length; + unsigned char header_[8] = {}; + size_t have_ = 0, name_length_ = 0; + std::string name_; + uint64_t remaining_ = 0; + bool accepted_ = false; + unsigned records_ = 0; + const char *problem_ = ""; + bool fail(const char *why); +}; + +/* One large file arriving in ranges (an upload session). */ +struct Session +{ + std::string id, temporary, destination; + uint64_t size = 0; + bool replace = false; + int fd = -1; + std::mutex lock; + std::map ranges; /* start -> end, merged */ + uint64_t covered = 0; + unsigned writers = 0; + bool closing = false; + std::time_t touched = 0; + /* Records [start, end) as received; under lock. */ + void add(uint64_t start, uint64_t end); +}; +class Sessions +{ + public: + /* A new session, its temporary file created and sized; nullptr with why set. */ + std::shared_ptr open(const std::string &id, const std::string &temporary, + const std::string &destination, uint64_t size, bool replace, + const char *&why); + std::shared_ptr find(const std::string &id); + /* Takes the session out (commit or abort); its file is the caller's then. */ + std::shared_ptr take(const std::string &id); + /* Drops sessions idle for longer than seconds (their temporary files deleted). */ + void expire(std::time_t now, std::time_t seconds); + /* Every session gone, as the server stops. */ + void clear(); + size_t count(); + static constexpr size_t limit = 16; + + private: + std::mutex lock_; + std::map> sessions_; +}; +/* Closes and deletes a session's file (abort, expiry). */ +void discard(Session &session); + +/* "bytes=a-b" (one range) against a file's size: false when absent or unusable. */ +bool parse_range(const char *header, uint64_t size, uint64_t &start, uint64_t &end); +} // namespace ps5_transfer diff --git a/tests/test_webui.py b/tests/test_webui.py index 9d7d797..64bcd17 100644 --- a/tests/test_webui.py +++ b/tests/test_webui.py @@ -28,7 +28,7 @@ class WebUI(unittest.TestCase): subprocess.run(['c++', '-std=c++17', '-O1', '-g', '-pthread', '-I'+str(ROOT / '.deps/webui/libmicrohttpd-1.0.10/src/include'), '-I'+str(ROOT / '.deps/native/zlib/zlib-1.3.2/contrib/minizip'), '-I'+str(ROOT / 'vendor/retroarch/deps/mbedtls'), - 'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_update.cpp', archive, update, '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True) + 'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp', 'src/webui_update.cpp', archive, update, '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True) with socket.socket() as s: s.bind(('127.0.0.1', 0)); cls.port = s.getsockname()[1] cls.process = subprocess.Popen([str(cls.binary), str(cls.root), str(cls.port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) @@ -328,3 +328,117 @@ class WebUI(unittest.TestCase): time.sleep(.02) self.assertFalse(list((self.root / 'content').glob('.upload-*'))) self.assertFalse((self.root / 'content/unfinished.iso').exists()) + + # The transfer engine (src/webui_transfer.h): batches, sessions, ranges, parallel. + @staticmethod + def batch(files, end=True): + out = b'' + for name, data in files: + raw = name.encode() + out += len(raw).to_bytes(2, 'little') + raw + len(data).to_bytes(8, 'little') + data + return out + (b'\0\0' if end else b'') + + def test_batch_upload_many_small_files(self): + files = [(f'media/{system}/{i:03}.png', bytes([i % 251]) * (i * 37 % 5000)) + for system in ('snes', 'nes') for i in range(150)] + status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-a', self.batch(files)) + result = json.loads(body) + self.assertEqual((status, result['written'], result['skipped'], result['failed']), (200, 300, 0, [])) + for name, data in files: + self.assertEqual((self.root / 'content/batch-a' / name).read_bytes(), data) + # Existing files are skipped unless replace; a replace writes them again. + changed = [(files[0][0], b'new'), ('media/snes/extra.png', b'x')] + result = json.loads(self.request('PUT', '/api/upload/batch?folder=batch-a', self.batch(changed))[2]) + self.assertEqual((result['written'], result['skipped']), (1, 1)) + self.assertEqual((self.root / 'content/batch-a' / files[0][0]).read_bytes(), files[0][1]) + result = json.loads(self.request('PUT', '/api/upload/batch?folder=batch-a&existing=replace&sync=0', + self.batch(changed))[2]) + self.assertEqual((result['written'], result['skipped']), (2, 0)) + self.assertEqual((self.root / 'content/batch-a' / files[0][0]).read_bytes(), b'new') + self.assertFalse(list((self.root / 'content/batch-a').rglob('.upload-*'))) + + def test_batch_refuses_bad_names_and_bad_streams(self): + bad = [('../escape.png', b'a'), ('.hidden', b'b'), ('a//b', b'c'), ('ok/fine.png', b'd'), ('/abs', b'e')] + status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-b', self.batch(bad)) + result = json.loads(body) + self.assertEqual((status, result['written'], len(result['failed'])), (200, 1, 4)) + self.assertFalse((self.root / 'escape.png').exists()) + self.assertEqual((self.root / 'content/batch-b/ok/fine.png').read_bytes(), b'd') + # Cut short: the files before the cut stand, the one cut is not left behind. + stream = self.batch([('whole.png', b'1' * 100)], end=False) + stream += (8).to_bytes(2, 'little') + b'half.png' + (1000).to_bytes(8, 'little') + b'2' * 10 + status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-c', stream) + result = json.loads(body) + self.assertEqual((status, result['complete'], result['written']), (400, False, 1)) + self.assertFalse((self.root / 'content/batch-c/half.png').exists()) + self.assertFalse(list((self.root / 'content/batch-c').glob('.upload-*'))) + # Data past the end marker, and a folder outside content, are refused. + self.assertEqual(self.request('PUT', '/api/upload/batch?folder=batch-d', self.batch([]) + b'junk')[0], 400) + self.assertEqual(self.request('PUT', '/api/upload/batch?folder=../x', self.batch([]))[0], 400) + self.assertEqual(self.request('PUT', '/api/upload/batch?folder=x', self.batch([]), + headers={'X-RetroArch-Token': 'wrong'})[0], 403) + + def test_session_parts_in_parallel_any_order(self): + import os, random, threading + data = os.urandom(5 * 1024 * 1024 + 123) + status, _, body = self.request('POST', '/api/upload/session?path=' + quote('big/game.iso') + f'&size={len(data)}') + self.assertEqual(status, 201) + sid = json.loads(body)['id'] + # Missing parts: the commit says what is covered and leaves the session open. + status, _, body = self.request('POST', f'/api/upload/commit?id={sid}') + self.assertEqual((status, json.loads(body)['covered']), (409, 0)) + part = 700 * 1024 + ranges = [(o, min(o + part + 4096, len(data))) for o in range(0, len(data), part)] # overlapping + random.Random(7).shuffle(ranges) + results = [] + def send(chunk): + for start, end in chunk: + results.append(self.request('PUT', f'/api/upload/part?id={sid}&offset={start}', data[start:end])[0]) + threads = [threading.Thread(target=send, args=(ranges[i::6],)) for i in range(6)] + [t.start() for t in threads]; [t.join() for t in threads] + self.assertEqual(set(results), {200}) + self.assertFalse((self.root / 'content/big/game.iso').exists()) + status, _, body = self.request('POST', f'/api/upload/commit?id={sid}') + self.assertEqual((status, json.loads(body)['bytes']), (201, len(data))) + self.assertEqual((self.root / 'content/big/game.iso').read_bytes(), data) + self.assertFalse(list((self.root / 'content/big').glob('.upload-*'))) + # The session is gone; a second commit, or a part, finds nothing. + self.assertEqual(self.request('POST', f'/api/upload/commit?id={sid}')[0], 404) + self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=0', b'x')[0], 404) + # The name now exists: a new session for it is refused unless replace. + self.assertEqual(self.request('POST', '/api/upload/session?path=big/game.iso&size=5')[0], 409) + + def test_session_bounds_abort_and_range_download(self): + status, _, body = self.request('POST', '/api/upload/session?path=abort.bin&size=10') + sid = json.loads(body)['id'] + self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=8', b'abc')[0], 416) + self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=x', b'a')[0], 416) + self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=0', b'0123456789')[0], 200) + self.assertEqual(json.loads(self.request('DELETE', f'/api/upload/session?id={sid}')[2])['cancelled'], True) + self.assertFalse((self.root / 'content/abort.bin').exists()) + self.assertFalse(list((self.root / 'content').glob('.upload-*'))) + self.assertEqual(self.request('POST', '/api/upload/session?path=../x&size=1')[0], 400) + # Ranges of a download: a slice, an open end, the last bytes, and one past the end. + (self.root / 'content/range.bin').write_bytes(bytes(range(256)) * 4) + status, headers, body = self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=10-19'}) + self.assertEqual((status, body, headers['Content-Range']), (206, bytes(range(10, 20)), 'bytes 10-19/1024')) + self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=1000-'})[2], + (bytes(range(256)) * 4)[1000:]) + self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=-4'})[2], + bytes(range(252, 256))) + self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=2000-'})[0], 416) + status, headers, body = self.request('GET', '/api/download?path=range.bin') + self.assertEqual((status, len(body), headers['Accept-Ranges']), (200, 1024, 'bytes')) + + def test_parallel_uploads_of_one_name_keep_one(self): + import threading + results = [] + def send(i): + results.append(self.request('PUT', '/api/upload?path=race.bin', str(i).encode() * 100000)[0]) + threads = [threading.Thread(target=send, args=(i,)) for i in range(8)] + [t.start() for t in threads]; [t.join() for t in threads] + self.assertEqual(results.count(201), 1) + self.assertEqual(results.count(409), 7) + content = (self.root / 'content/race.bin').read_bytes() + self.assertEqual(len(set(content)), 1) + self.assertFalse(list((self.root / 'content').glob('.upload-*'))) diff --git a/tests/test_webui_daemon.py b/tests/test_webui_daemon.py index 315353f..f47f2f3 100644 --- a/tests/test_webui_daemon.py +++ b/tests/test_webui_daemon.py @@ -47,7 +47,7 @@ class WebuiDaemon(unittest.TestCase): '-I' + str(ROOT / 'vendor/retroarch/deps/mbedtls'), f'-DPS5_WEBUI_HOMEBREW="{cls.homebrew}"', f'-DPS5_WEBUI_LINK_PORT={cls.link_port}', f'-DPS5_WEBUI_HTTP_PORT={cls.http_port}', f'-DPS5_WEBUI_IDLE_SECONDS={IDLE}', - 'daemon/webui_daemon.cpp', 'src/webui_ps5.cpp', 'src/webui_update.cpp', http, update, + 'daemon/webui_daemon.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp', 'src/webui_update.cpp', http, update, '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True) @classmethod diff --git a/tests/test_webui_transfer_browser.py b/tests/test_webui_transfer_browser.py new file mode 100644 index 0000000..d488643 --- /dev/null +++ b/tests/test_webui_transfer_browser.py @@ -0,0 +1,65 @@ +"""The page's transfer engine in Chromium (tests/webui_transfer_browser.cjs) against +the native server, then the files on disk compared with what the page made.""" +from pathlib import Path +import os +import shutil +import socket +import subprocess +import tempfile +import time +import unittest + +ROOT = Path(__file__).resolve().parent.parent +PLAYWRIGHT = os.environ.get('PLAYWRIGHT_PATH') or next( + (str(p) for p in (Path.home() / '.cache/codex-runtimes/codex-primary-runtime/dependencies/node/node_modules/playwright-core', + Path('/usr/lib/chatgpt/resources/cua_node/lib/node_modules/playwright-core')) if p.exists()), '') + + +def pattern(size, seed): + data = bytearray(size) + for i in range(0, size, 4096): + data[i] = (seed + i) & 255 + return bytes(data) + + +@unittest.skipUnless(PLAYWRIGHT and shutil.which('node') and Path('/usr/bin/chromium').exists(), 'needs node, Playwright and Chromium') +class TransferBrowser(unittest.TestCase): + def test_page_engine(self): + http = subprocess.check_output(['bash', 'tools/build-webui-http.sh', 'host'], cwd=ROOT, text=True).strip() + update = subprocess.check_output(['bash', 'tools/build-webui-update.sh', 'host'], cwd=ROOT, text=True).strip() + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) / 'title' + for name in ('config', 'content', 'webui'): + (root / name).mkdir(parents=True) + shutil.copytree(ROOT / 'webui', root / 'webui', dirs_exist_ok=True) + binary = Path(temp) / 'server' + subprocess.run(['c++', '-std=c++17', '-O1', '-pthread', + '-I' + str(ROOT / '.deps/webui/libmicrohttpd-1.0.10/src/include'), + '-I' + str(ROOT / '.deps/native/zlib/zlib-1.3.2/contrib/minizip'), + '-I' + str(ROOT / 'vendor/retroarch/deps/mbedtls'), + 'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp', + 'src/webui_update.cpp', http, update, '-lz', '-o', str(binary)], cwd=ROOT, check=True) + with socket.socket() as s: + s.bind(('127.0.0.1', 0)); port = s.getsockname()[1] + server = subprocess.Popen([str(binary), str(root), str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + time.sleep(0.5) + run = subprocess.run(['node', 'tests/webui_transfer_browser.cjs'], cwd=ROOT, capture_output=True, text=True, timeout=600, + env={**os.environ, 'PLAYWRIGHT_PATH': PLAYWRIGHT, 'WEBUI_TEST_URL': f'http://127.0.0.1:{port}'}) + self.assertEqual(run.returncode, 0, run.stdout + run.stderr) + print(run.stdout.strip()) + finally: + server.terminate(); server.wait(timeout=5) + folder = root / 'content/engine' + for i in range(600): + self.assertEqual((folder / f'shot-{i:04}.png').read_bytes(), pattern(1000 + (i * 997) % 60000, i)) + for i in range(3): + self.assertEqual((folder / f'medium-{i}.bin').read_bytes(), pattern(9 * 1024 * 1024 + i, 7 + i)) + for i in range(2): + self.assertEqual((folder / f'large-{i}.iso').read_bytes(), pattern(150 * 1024 * 1024 + 4097 * i, 31 + i)) + self.assertFalse((folder / 'cancel.iso').exists()) + self.assertEqual(list(folder.glob('.upload-*')), []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/webui_transfer_browser.cjs b/tests/webui_transfer_browser.cjs new file mode 100644 index 0000000..9185014 --- /dev/null +++ b/tests/webui_transfer_browser.cjs @@ -0,0 +1,59 @@ +/* The page's transfer engine in Chromium against a native webui_server_main fixture + * (tests/test_webui_transfer_browser.py starts it): many small files go as batches, + * large ones as parallel parts, the rest one to a request; all arrive whole, names + * that exist are refused, and a cancelled large file leaves nothing behind. */ +const assert = require('node:assert/strict'); +const { chromium } = require(process.env.PLAYWRIGHT_PATH || 'playwright'); +const base = process.env.WEBUI_TEST_URL; +(async () => { + const browser = await chromium.launch({ headless: true, executablePath: process.env.CHROMIUM_PATH || '/usr/bin/chromium' }); + try { + const page = await browser.newPage(); + const errors = []; page.on('pageerror', e => errors.push(e.message)); + await page.route('https://api.github.com/**', route => route.fulfill({ json: [] })); + const requests = []; page.on('request', r => { if (r.url().includes('/api/upload')) requests.push(new URL(r.url()).pathname); }); + await page.goto(base); + await page.waitForFunction(() => !document.querySelector('#browse-files').disabled); + // 600 small files, 3 medium, 2 large: made in the page, queued as a person drops them. + const summary = await page.evaluate(async () => { + const make = (name, size, seed) => { const b = new Uint8Array(size); for (let i = 0; i < size; i += 4096) b[i] = (seed + i) & 255; return new File([b], name); }; + const files = []; + for (let i = 0; i < 600; i++) files.push(make(`shot-${String(i).padStart(4, '0')}.png`, 1000 + (i * 997) % 60000, i)); + for (let i = 0; i < 3; i++) files.push(make(`medium-${i}.bin`, 9 * 1024 * 1024 + i, 7 + i)); + for (let i = 0; i < 2; i++) files.push(make(`large-${i}.iso`, 150 * 1024 * 1024 + 4097 * i, 31 + i)); + const started = performance.now(); + queueFiles(files, 'engine'); + await new Promise(resolve => { const t = setInterval(() => { if (!transfers.some(x => ['queued', 'uploading'].includes(x.state))) { clearInterval(t); resolve(); } }, 50); }); + const states = {}; for (const x of transfers) states[x.state] = (states[x.state] || 0) + 1; + return { states, seconds: (performance.now() - started) / 1000, failed: transfers.filter(x => x.state === 'failed').map(x => x.name + ': ' + x.message).slice(0, 5) }; + }); + assert.deepEqual(summary.states, { complete: 605 }, JSON.stringify(summary)); + const count = name => requests.filter(p => p === name).length; + assert.ok(count('/api/upload/batch') >= 3 && count('/api/upload/batch') <= 10, `batches: ${count('/api/upload/batch')}`); + assert.equal(count('/api/upload'), 3); + assert.equal(count('/api/upload/session'), 2); + assert.equal(count('/api/upload/commit'), 2); + assert.ok(count('/api/upload/part') >= 12, `parts: ${count('/api/upload/part')}`); + // The same names again: every one is refused, none replaced. + const again = await page.evaluate(async () => { + const files = [new File([new Uint8Array(10)], 'shot-0001.png'), new File([new Uint8Array(9 * 1024 * 1024)], 'medium-0.bin')]; + const before = transfers.length; queueFiles(files, 'engine'); + await new Promise(resolve => { const t = setInterval(() => { if (!transfers.some(x => ['queued', 'uploading'].includes(x.state))) { clearInterval(t); resolve(); } }, 50); }); + return transfers.slice(before).map(x => x.state + ': ' + x.message); + }); + assert.deepEqual(again.map(x => x.split(':')[0]), ['failed', 'failed']); + assert.match(again[0], /already exists/); assert.match(again[1], /already exists/); + // A large file cancelled mid-way: the session is dropped. + const cancelled = await page.evaluate(async () => { + const before = transfers.length; queueFiles([new File([new Uint8Array(200 * 1024 * 1024)], 'cancel.iso')], 'engine'); + const transfer = transfers[before]; + await new Promise(resolve => { const t = setInterval(() => { if ((transfer.sent || 0) > 20 * 1024 * 1024) { clearInterval(t); resolve(); } }, 10); }); + cancelTransfer(transfer); + await new Promise(resolve => setTimeout(resolve, 1500)); + return transfer.state; + }); + assert.equal(cancelled, 'cancelled'); + assert.deepEqual(errors, []); + console.log(`PASS: 605 files (600 small in ${count('/api/upload/batch')} batches, 3 single, 2 large in ${count('/api/upload/part')} parts) in ${summary.seconds.toFixed(1)} s; duplicates refused; cancel drops the session.`); + } finally { await browser.close(); } +})().catch(error => { console.error(error); process.exit(1); }); diff --git a/tests/webui_update_fixture.py b/tests/webui_update_fixture.py index 941b7c9..8cdb5c2 100644 --- a/tests/webui_update_fixture.py +++ b/tests/webui_update_fixture.py @@ -14,7 +14,7 @@ shutil.copytree('webui',r/'webui',dirs_exist_ok=True);shutil.copytree('build/web (r/'config/retroarch.cfg').write_text('audio_volume = "0"\ninput_rumble_gain = "100"\n') for file in Path('dist/PPSA99169/cores').glob('*.so'):(r/'cores'/file.name).touch() shutil.copytree('dist/PPSA99169/info',r/'info',dirs_exist_ok=True) -subprocess.run(['c++','-std=c++17','-pthread','-I.deps/webui/libmicrohttpd-1.0.10/src/include','-I.deps/native/zlib/zlib-1.3.2/contrib/minizip','-Ivendor/retroarch/deps/mbedtls','tests/webui_server_main.cpp','src/webui_ps5.cpp','src/webui_update.cpp','build/webui-mhd-host/src/microhttpd/.libs/libmicrohttpd.a','build/webui-update-host/libupdate.a','-lz','-o',str(r/'server')],check=True) +subprocess.run(['c++','-std=c++17','-pthread','-I.deps/webui/libmicrohttpd-1.0.10/src/include','-I.deps/native/zlib/zlib-1.3.2/contrib/minizip','-Ivendor/retroarch/deps/mbedtls','tests/webui_server_main.cpp','src/webui_ps5.cpp','src/webui_transfer.cpp','src/webui_update.cpp','build/webui-mhd-host/src/microhttpd/.libs/libmicrohttpd.a','build/webui-update-host/libupdate.a','-lz','-o',str(r/'server')],check=True) try: subprocess.run([str(r/'server'),str(r),'6770'],check=True) finally: diff --git a/tools/build-webui-daemon.sh b/tools/build-webui-daemon.sh index ae882b5..3a2ad77 100755 --- a/tools/build-webui-daemon.sh +++ b/tools/build-webui-daemon.sh @@ -15,7 +15,7 @@ includes=(-I"$root/src" -I"$root/.deps/webui/libmicrohttpd-1.0.10/src/include" -I"$root/.deps/native/zlib/zlib-1.3.2" -I"$root/.deps/native/zlib/zlib-1.3.2/contrib/minizip" -I"$root/vendor/retroarch/deps/mbedtls") objects=() -for source in daemon/webui_daemon src/webui_ps5 src/webui_update; do +for source in daemon/webui_daemon src/webui_ps5 src/webui_transfer src/webui_update; do "$sdk/bin/prospero-clang++" -std=c++17 -O2 -Wall -Werror "${includes[@]}" \ -c "$root/$source.cpp" -o "$build/${source##*/}.o" objects+=("$build/${source##*/}.o") diff --git a/webui/app.js b/webui/app.js index c9030f1..495f4f6 100644 --- a/webui/app.js +++ b/webui/app.js @@ -197,7 +197,7 @@ function drawTransfers() { progress = document.createElement('progress'); progress.max = 100; progress.setAttribute('aria-label', `Upload progress for ${transfer.name}`); details.append(progress); const cancel = element('button', 'Cancel', 'secondary'); - cancel.addEventListener('click', () => { transfer.state = 'cancelled'; transfer.message = 'Cancelled'; if (transfer.xhr) transfer.xhr.abort(); transfer.file = null; drawTransfers(); }); + cancel.addEventListener('click', () => cancelTransfer(transfer)); row.append(cancel); } progress.value = transfer.percent || 0; @@ -216,35 +216,175 @@ function queueFiles(files, destination) { } drawTransfers(); sendNext(); } +// The transfer engine (src/webui_transfer.h on the console): six lanes run at once. +// Small files travel many to a request (a batch), large files as parts on every free +// lane (an upload session), the rest one to a request. Every file is renamed into +// place on the console only once whole. +const LANES = 6, SMALL = 4 * 1024 ** 2, LARGE = 64 * 1024 ** 2, PART = 32 * 1024 ** 2; +const BATCH_FILES = 256, BATCH_BYTES = 16 * 1024 ** 2, PART_TRIES = 3; +let jobs = [], opening = 0; // sessions being opened: their parts are about to queue +function cancelTransfer(transfer) { + if (!['queued', 'uploading'].includes(transfer.state)) return; + transfer.state = 'cancelled'; transfer.message = 'Cancelled'; + for (const xhr of transfer.requests || []) xhr.abort(); + if (transfer.session) api('/api/upload/session?id=' + encodeURIComponent(transfer.session), { method: 'DELETE' }).catch(() => {}); + transfer.file = null; drawTransfers(); +} +function showProgress(transfer, sent) { + transfer.sent = Math.min(sent, transfer.file?.size ?? sent); + const size = transfer.file?.size || 0, percent = size ? Math.floor(transfer.sent / size * 100) : 100; + if (percent === transfer.percent && transfer.state === 'uploading') return; + transfer.percent = percent; transfer.state = 'uploading'; + transfer.message = percent >= 100 ? 'Finishing on the console…' : `${percent}% · ${bytes(transfer.sent)} of ${bytes(size)}`; + const now = Date.now(); if (now - (transfer.drawn || 0) > 150 || percent >= 100) { transfer.drawn = now; drawTransfers(); } +} +function finish(transfer, state, message) { + if (transfer.state === 'cancelled') return; + transfer.state = state; transfer.message = message; transfer.file = null; transfer.requests = []; + transfer.session = null; drawTransfers(); +} +// One request on a lane: resolves with {status, body}, or rejects when the connection fails. +function send(method, url, body, transfers, onProgress) { + return new Promise((resolve, reject) => { + const xhr = new XMLHttpRequest(); + for (const transfer of transfers) (transfer.requests ||= []).push(xhr); + const done = () => { for (const transfer of transfers) transfer.requests = (transfer.requests || []).filter(x => x !== xhr); }; + xhr.open(method, url); xhr.setRequestHeader('X-RetroArch-Token', token); + if (onProgress) xhr.upload.onprogress = event => onProgress(event.loaded); + xhr.onload = () => { done(); let parsed = {}; try { parsed = JSON.parse(xhr.responseText); } catch {} resolve({ status: xhr.status, body: parsed }); }; + xhr.onerror = () => { done(); reject(new Error('Connection lost. Reconnect, then choose this file again.')); }; + xhr.onabort = () => { done(); reject(new Error('Cancelled')); }; + xhr.send(body); + }); +} +// The batch stream: per file a u16 name length, the name (UTF-8), a u64 size and the +// file itself, then a zero length. A Blob of the files' own Blobs: nothing is copied. +function batchBody(transfers) { + const parts = [], encoder = new TextEncoder(); + for (const transfer of transfers) { + const name = encoder.encode(transfer.path), head = new Uint8Array(2 + name.length + 8), view = new DataView(head.buffer); + view.setUint16(0, name.length, true); head.set(name, 2); view.setBigUint64(2 + name.length, BigInt(transfer.file.size), true); + parts.push(head, transfer.file); + } + parts.push(new Uint8Array(2)); + return new Blob(parts); +} +async function runBatch(transfers) { + transfers = transfers.filter(transfer => transfer.state === 'queued'); + if (!transfers.length) return; + const encoder = new TextEncoder(), starts = []; let at = 0; + for (const transfer of transfers) { at += 2 + encoder.encode(transfer.path).length + 8; starts.push(at); at += transfer.file.size; } + for (const transfer of transfers) showProgress(transfer, 0); + try { + const { status, body: result } = await send('PUT', '/api/upload/batch', batchBody(transfers), transfers, + loaded => transfers.forEach((transfer, i) => showProgress(transfer, loaded - starts[i]))); + const failed = new Map((result.failed || []).map(entry => [entry.name, entry.error])); + const skipped = new Set(result.skippedNames || []); + for (const transfer of transfers) { + if (transfer.state !== 'uploading') continue; + if (skipped.has(transfer.path)) finish(transfer, 'failed', 'A file with this name already exists. Rename your file first.'); + else if (failed.has(transfer.path)) finish(transfer, 'failed', `Upload failed (${failed.get(transfer.path)}). Choose the file again to retry.`); + else if (status === 200) finish(transfer, 'complete', `Uploaded · ${bytes(transfer.file.size)}`); + else finish(transfer, 'failed', result.error || result.problem || 'Upload failed. Choose the file again to retry.'); + } + } catch (error) { + for (const transfer of transfers) if (transfer.state === 'uploading') finish(transfer, 'failed', error.message); + if (error.message !== 'Cancelled') setConnection(false); + } +} +async function runSingle(transfer) { + if (transfer.state !== 'queued') return; + showProgress(transfer, 0); + try { + const { status, body } = await send('PUT', '/api/upload?path=' + encodeURIComponent(transfer.path), transfer.file, [transfer], loaded => showProgress(transfer, loaded)); + if (status === 201) finish(transfer, 'complete', `Uploaded · ${bytes(transfer.file.size)}`); + else finish(transfer, 'failed', body.error || 'Upload failed. Choose the file again to retry.'); + } catch (error) { finish(transfer, error.message === 'Cancelled' ? 'cancelled' : 'failed', error.message); if (error.message !== 'Cancelled') setConnection(false); } +} +// A large file: a session, then its parts as jobs any lane takes. The part that +// completes the file commits it, on its own lane: no lane ever waits on another. +async function runLarge(transfer) { + if (transfer.state !== 'queued') return; + showProgress(transfer, 0); + let opened; + opening++; + try { opened = await send('POST', `/api/upload/session?path=${encodeURIComponent(transfer.path)}&size=${transfer.file.size}`, null, [transfer]); } + catch (error) { finish(transfer, 'failed', error.message); return; } + finally { opening--; } + if (opened.status !== 201) { finish(transfer, 'failed', opened.body.error || 'Upload failed. Choose the file again to retry.'); return; } + transfer.session = opened.body.id; + const size = transfer.file.size, part = Math.max(4 * 1024 ** 2, Math.min(opened.body.partSize || PART, Math.ceil(size / LANES))); + const sent = new Map(), parts = []; + for (let start = 0; start < size; start += part) parts.push([start, Math.min(start + part, size)]); + let left = parts.length, failure = null; + const abandon = message => { + if (failure) return; + failure = message; + api('/api/upload/session?id=' + encodeURIComponent(transfer.session), { method: 'DELETE' }).catch(() => {}); + if (message !== 'Cancelled') finish(transfer, 'failed', message); + }; + const commit = async () => { + try { + const { status, body } = await send('POST', '/api/upload/commit?id=' + encodeURIComponent(transfer.session), null, [transfer]); + if (status === 201) finish(transfer, 'complete', `Uploaded · ${bytes(size)}`); + else finish(transfer, 'failed', body.error || 'Upload failed. Choose the file again to retry.'); + } catch (error) { finish(transfer, 'failed', error.message); } + }; + const partJob = ([start, end]) => async () => { + for (let attempt = 1; !failure; attempt++) { + if (transfer.state !== 'uploading') { abandon('Cancelled'); return; } + try { + const { status, body } = await send('PUT', `/api/upload/part?id=${encodeURIComponent(transfer.session)}&offset=${start}`, transfer.file.slice(start, end), [transfer], + loaded => { sent.set(start, loaded); showProgress(transfer, [...sent.values()].reduce((a, b) => a + b, 0)); }); + if (status === 200) { sent.set(start, end - start); break; } + if (attempt >= PART_TRIES || status === 404 || status === 416) { abandon(body.error || 'A part could not be written.'); return; } + } catch (error) { + if (error.message === 'Cancelled') { abandon('Cancelled'); return; } + if (attempt >= PART_TRIES) { abandon(error.message); return; } + } + } + if (!failure && --left === 0) await commit(); + }; + // First in the queue, so every free lane joins this file at once. + jobs.unshift(...parts.map(partJob)); +} +// Turns the queued transfers into jobs: batches of small files, sessions, single files. +function planJobs() { + let batch = [], batchBytes = 0; + const flush = () => { if (batch.length) { const group = batch; jobs.push(() => runBatch(group)); } batch = []; batchBytes = 0; }; + for (const transfer of transfers) { + if (transfer.state !== 'queued' || transfer.planned) continue; + transfer.planned = true; + const size = transfer.file.size; + if (size < SMALL) { + if (batch.length >= BATCH_FILES || batchBytes + size > BATCH_BYTES) flush(); + batch.push(transfer); batchBytes += size; + } else if (size >= LARGE) jobs.push(() => runLarge(transfer)); + else jobs.push(() => runSingle(transfer)); + } + flush(); +} async function sendNext() { + planJobs(); if (sending) return; sending = true; try { - for (const transfer of transfers) { - if (transfer.state !== 'queued') continue; - if (!connected && !await reconnect()) { transfer.state = 'failed'; transfer.message = 'Console disconnected. Reconnect, then choose this file again.'; transfer.file = null; drawTransfers(); continue; } - await new Promise(resolve => { - const xhr = new XMLHttpRequest(); transfer.xhr = xhr; transfer.state = 'uploading'; - transfer.message = 'Uploading…'; drawTransfers(); - xhr.open('PUT', '/api/upload?path=' + encodeURIComponent(transfer.path)); - xhr.setRequestHeader('X-RetroArch-Token', token); - let lastProgress = 0; - xhr.upload.onprogress = event => { - if (!event.lengthComputable) return; - transfer.percent = Math.floor(event.loaded / event.total * 100); - transfer.message = transfer.percent === 100 ? 'Finishing on the console…' : `${transfer.percent}% · ${bytes(event.loaded)} of ${bytes(event.total)}`; - if (Date.now() - lastProgress > 150 || transfer.percent === 100) { lastProgress = Date.now(); drawTransfers(); } - }; - xhr.onload = () => { - if (xhr.status === 201) { transfer.state = 'complete'; transfer.message = `Uploaded · ${bytes(transfer.file.size)}`; } - else { transfer.state = 'failed'; try { transfer.message = JSON.parse(xhr.responseText).error; } catch { transfer.message = 'Upload failed. Choose the file again to retry.'; } } - }; - xhr.onerror = () => { transfer.state = 'failed'; transfer.message = 'Connection lost. Reconnect, then choose this file again.'; setConnection(false); }; - xhr.onabort = () => { transfer.state = 'cancelled'; transfer.message = 'Cancelled'; }; - xhr.onloadend = () => { transfer.file = null; transfer.xhr = null; drawTransfers(); resolve(); }; - xhr.send(transfer.file); - }); + if (!connected && !await reconnect()) { + for (const transfer of transfers) if (transfer.state === 'queued') finish(transfer, 'failed', 'Console disconnected. Reconnect, then choose this file again.'); + jobs = []; return; } + // Each lane runs one job at a time until none is left; a job is one request (or a + // part and its file's commit), so no lane waits on another lane's work. + const lane = async () => { + for (;;) { + planJobs(); + const job = jobs.shift(); + if (job) await job(); + else if (opening) await new Promise(resolve => setTimeout(resolve, 25)); + else return; + } + }; + await Promise.all(Array.from({ length: LANES }, lane)); } finally { sending = false; await Promise.all([loadLibrary(), loadContent(currentPath), reconnect()]); } } let pickerDestination = '';