From ef25c3edc7b06c512079debd6dac6a1a7e5f7b48 Mon Sep 17 00:00:00 2001 From: ITotalJustice <47043333+ITotalJustice@users.noreply.github.com> Date: Tue, 20 May 2025 22:50:05 +0100 Subject: [PATCH] multi thread game dumps and file uploads. previous uploads were all single threaded, which meant that it only uploaded as fast as the slowest source. usb transfer is still single threaded due it being random access for both files and data, making it hard for the read thread to run freely. --- sphaira/CMakeLists.txt | 1 + sphaira/include/threaded_file_transfer.hpp | 17 + sphaira/include/ui/menus/filebrowser.hpp | 2 +- sphaira/include/ui/progress_box.hpp | 5 + sphaira/source/threaded_file_transfer.cpp | 372 +++++++++++++++++++++ sphaira/source/ui/menus/filebrowser.cpp | 59 +++- sphaira/source/ui/menus/game_menu.cpp | 126 ++++--- sphaira/source/ui/menus/gc_menu.cpp | 118 +++---- sphaira/source/ui/progress_box.cpp | 1 + 9 files changed, 555 insertions(+), 146 deletions(-) create mode 100644 sphaira/include/threaded_file_transfer.hpp create mode 100644 sphaira/source/threaded_file_transfer.cpp diff --git a/sphaira/CMakeLists.txt b/sphaira/CMakeLists.txt index 2f43ef2..4321525 100644 --- a/sphaira/CMakeLists.txt +++ b/sphaira/CMakeLists.txt @@ -79,6 +79,7 @@ add_executable(sphaira source/swkbd.cpp source/i18n.cpp source/ftpsrv_helper.cpp + source/threaded_file_transfer.cpp source/usb/base.cpp source/usb/usbds.cpp diff --git a/sphaira/include/threaded_file_transfer.hpp b/sphaira/include/threaded_file_transfer.hpp new file mode 100644 index 0000000..a85ab5d --- /dev/null +++ b/sphaira/include/threaded_file_transfer.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include "ui/progress_box.hpp" +#include +#include + +namespace sphaira::thread { + +using ReadFunctionCallback = std::function; +using WriteFunctionCallback = std::function; +using PullFunctionCallback = std::function; +using StartFunctionCallback = std::function; + +Result Transfer(ui::ProgressBox* pbox, s64 size, ReadFunctionCallback rfunc, WriteFunctionCallback wfunc); +Result TransferPull(ui::ProgressBox* pbox, s64 size, ReadFunctionCallback rfunc, StartFunctionCallback sfunc); + +} // namespace sphaira::thread diff --git a/sphaira/include/ui/menus/filebrowser.hpp b/sphaira/include/ui/menus/filebrowser.hpp index 5c2c683..2c92588 100644 --- a/sphaira/include/ui/menus/filebrowser.hpp +++ b/sphaira/include/ui/menus/filebrowser.hpp @@ -236,7 +236,7 @@ private: auto CheckIfUpdateFolder() -> Result; auto get_collection(const fs::FsPath& path, const fs::FsPath& parent_name, FsDirCollection& out, bool inc_file, bool inc_dir, bool inc_size) -> Result; - auto get_collections(const fs::FsPath& path, const fs::FsPath& parent_name, FsDirCollections& out) -> Result; + auto get_collections(const fs::FsPath& path, const fs::FsPath& parent_name, FsDirCollections& out, bool inc_size = false) -> Result; void SetFs(const fs::FsPath& new_path, u32 new_type); diff --git a/sphaira/include/ui/progress_box.hpp b/sphaira/include/ui/progress_box.hpp index b329cce..15009ea 100644 --- a/sphaira/include/ui/progress_box.hpp +++ b/sphaira/include/ui/progress_box.hpp @@ -37,6 +37,10 @@ struct ProgressBox final : Widget { auto CopyFile(const fs::FsPath& src, const fs::FsPath& dst) -> Result; void Yield(); + auto GetCpuId() const { + return m_cpuid; + } + auto OnDownloadProgressCallback() { return [this](s64 dltotal, s64 dlnow, s64 ultotal, s64 ulnow){ if (this->ShouldExit()) { @@ -81,6 +85,7 @@ private: std::vector m_image_data{}; // shared data end. + int m_cpuid{}; int m_image{}; bool m_own_image{}; }; diff --git a/sphaira/source/threaded_file_transfer.cpp b/sphaira/source/threaded_file_transfer.cpp new file mode 100644 index 0000000..370f4d9 --- /dev/null +++ b/sphaira/source/threaded_file_transfer.cpp @@ -0,0 +1,372 @@ +#include "threaded_file_transfer.hpp" +#include "log.hpp" +#include "defines.hpp" + +#include +#include +#include + +namespace sphaira::thread { +namespace { + +constexpr u64 READ_BUFFER_MAX = 1024*1024*4; + +struct ThreadBuffer { + ThreadBuffer() { + buf.reserve(READ_BUFFER_MAX); + } + + std::vector buf; + s64 off; +}; + +template +struct RingBuf { +private: + ThreadBuffer buf[Size]{}; + unsigned r_index{}; + unsigned w_index{}; + + static_assert((sizeof(RingBuf::buf) & (sizeof(RingBuf::buf) - 1)) == 0, "Must be power of 2!"); + +public: + void ringbuf_reset() { + this->r_index = this->w_index; + } + + unsigned ringbuf_capacity() const { + return sizeof(this->buf) / sizeof(this->buf[0]); + } + + unsigned ringbuf_size() const { + return (this->w_index - this->r_index) % (ringbuf_capacity() * 2U); + } + + unsigned ringbuf_free() const { + return ringbuf_capacity() - ringbuf_size(); + } + + void ringbuf_push(std::vector& buf_in, s64 off_in) { + auto& value = this->buf[this->w_index % ringbuf_capacity()]; + value.off = off_in; + std::swap(value.buf, buf_in); + + this->w_index = (this->w_index + 1U) % (ringbuf_capacity() * 2U); + } + + void ringbuf_pop(std::vector& buf_out, s64& off_out) { + auto& value = this->buf[this->r_index % ringbuf_capacity()]; + off_out = value.off; + std::swap(value.buf, buf_out); + + this->r_index = (this->r_index + 1U) % (ringbuf_capacity() * 2U); + } +}; + +struct ThreadData { + ThreadData(ui::ProgressBox* _pbox, s64 size, ReadFunctionCallback _rfunc, WriteFunctionCallback _wfunc); + + auto GetResults() -> Result; + void WakeAllThreads(); + + void SetReadResult(Result result) { + read_result = result; + } + + void SetWriteResult(Result result) { + write_result = result; + } + + void SetPullResult(Result result) { + pull_result = result; + } + + auto GetWriteOffset() const { + return write_offset; + } + + auto GetWriteSize() const { + return write_size; + } + + Result Pull(void* data, s64 size, u64* bytes_read); + Result readFuncInternal(); + Result writeFuncInternal(); + +private: + Result SetWriteBuf(std::vector& buf, s64 size); + Result GetWriteBuf(std::vector& buf_out, s64& off_out); + Result SetPullBuf(std::vector& buf, s64 size); + Result GetPullBuf(void* data, s64 size, u64* bytes_read); + + Result Read(void* buf, s64 size, u64* bytes_read); + +private: + // these need to be copied + ui::ProgressBox* pbox{}; + ReadFunctionCallback rfunc{}; + WriteFunctionCallback wfunc{}; + + // these need to be created + Mutex mutex{}; + Mutex pull_mutex{}; + + CondVar can_read{}; + CondVar can_write{}; + CondVar can_pull{}; + CondVar can_pull_write{}; + + RingBuf<2> write_buffers{}; + std::vector pull_buffer{}; + s64 pull_buffer_offset{}; + + u64 read_buffer_size{}; + u64 max_buffer_size{}; + + // these are shared between threads + volatile s64 read_offset{}; + volatile s64 write_offset{}; + volatile s64 write_size{}; + + volatile Result read_result{}; + volatile Result write_result{}; + volatile Result pull_result{}; +}; + +ThreadData::ThreadData(ui::ProgressBox* _pbox, s64 size, ReadFunctionCallback _rfunc, WriteFunctionCallback _wfunc) +: pbox{_pbox}, rfunc{_rfunc}, wfunc{_wfunc} { + mutexInit(std::addressof(mutex)); + mutexInit(std::addressof(pull_mutex)); + + condvarInit(std::addressof(can_read)); + condvarInit(std::addressof(can_write)); + condvarInit(std::addressof(can_pull)); + condvarInit(std::addressof(can_pull_write)); + + write_size = size; + read_buffer_size = READ_BUFFER_MAX; + max_buffer_size = READ_BUFFER_MAX; +} + +auto ThreadData::GetResults() -> Result { + R_UNLESS(!pbox->ShouldExit(), 0x1); + R_TRY(read_result); + R_TRY(write_result); + R_TRY(pull_result); + R_SUCCEED(); +} + +void ThreadData::WakeAllThreads() { + condvarWakeAll(std::addressof(can_read)); + condvarWakeAll(std::addressof(can_write)); + condvarWakeAll(std::addressof(can_pull)); + condvarWakeAll(std::addressof(can_pull_write)); + + mutexUnlock(std::addressof(mutex)); + mutexUnlock(std::addressof(pull_mutex)); +} + +Result ThreadData::SetWriteBuf(std::vector& buf, s64 size) { + buf.resize(size); + + mutexLock(std::addressof(mutex)); + if (!write_buffers.ringbuf_free()) { + R_TRY(condvarWait(std::addressof(can_read), std::addressof(mutex))); + } + + ON_SCOPE_EXIT(mutexUnlock(std::addressof(mutex))); + R_TRY(GetResults()); + write_buffers.ringbuf_push(buf, 0); + return condvarWakeOne(std::addressof(can_write)); +} + +Result ThreadData::GetWriteBuf(std::vector& buf_out, s64& off_out) { + mutexLock(std::addressof(mutex)); + if (!write_buffers.ringbuf_size()) { + R_TRY(condvarWait(std::addressof(can_write), std::addressof(mutex))); + } + + ON_SCOPE_EXIT(mutexUnlock(std::addressof(mutex))); + R_TRY(GetResults()); + write_buffers.ringbuf_pop(buf_out, off_out); + return condvarWakeOne(std::addressof(can_read)); +} + +Result ThreadData::SetPullBuf(std::vector& buf, s64 size) { + buf.resize(size); + + mutexLock(std::addressof(pull_mutex)); + if (!pull_buffer.empty()) { + R_TRY(condvarWait(std::addressof(can_pull_write), std::addressof(pull_mutex))); + } + + ON_SCOPE_EXIT(mutexUnlock(std::addressof(pull_mutex))); + R_TRY(GetResults()); + + pull_buffer.swap(buf); + return condvarWakeOne(std::addressof(can_pull)); +} + +Result ThreadData::GetPullBuf(void* data, s64 size, u64* bytes_read) { + mutexLock(std::addressof(pull_mutex)); + if (pull_buffer.empty()) { + R_TRY(condvarWait(std::addressof(can_pull), std::addressof(pull_mutex))); + } + + ON_SCOPE_EXIT(mutexUnlock(std::addressof(pull_mutex))); + R_TRY(GetResults()); + + *bytes_read = size = std::min(size, pull_buffer.size() - pull_buffer_offset); + std::memcpy(data, pull_buffer.data() + pull_buffer_offset, size); + pull_buffer_offset += size; + + if (pull_buffer_offset == pull_buffer.size()) { + pull_buffer_offset = 0; + pull_buffer.clear(); + return condvarWakeOne(std::addressof(can_pull_write)); + } else { + R_SUCCEED(); + } +} + +Result ThreadData::Read(void* buf, s64 size, u64* bytes_read) { + size = std::min(size, write_size - read_offset); + const auto rc = rfunc(buf, read_offset, size, bytes_read); + read_offset += *bytes_read; + return rc; +} + +Result ThreadData::Pull(void* data, s64 size, u64* bytes_read) { + return GetPullBuf(data, size, bytes_read); +} + +// read thread reads all data from the source +Result ThreadData::readFuncInternal() { + // the main buffer which data is read into. + std::vector buf; + buf.reserve(this->max_buffer_size); + + while (this->read_offset < this->write_size && R_SUCCEEDED(this->GetResults())) { + // read more data + s64 read_size = this->read_buffer_size; + + u64 bytes_read{}; + buf.resize(read_size); + R_TRY(this->Read(buf.data(), read_size, std::addressof(bytes_read))); + auto buf_size = bytes_read; + + R_TRY(this->SetWriteBuf(buf, buf_size)); + } + + log_write("read success\n"); + R_SUCCEED(); +} + +// write thread writes data to the nca placeholder. +Result ThreadData::writeFuncInternal() { + std::vector buf; + buf.reserve(this->max_buffer_size); + + while (this->write_offset < this->write_size && R_SUCCEEDED(this->GetResults())) { + s64 dummy_off; + R_TRY(this->GetWriteBuf(buf, dummy_off)); + const auto size = buf.size(); + + if (!this->wfunc) { + R_TRY(this->SetPullBuf(buf, buf.size())); + } else { + R_TRY(this->wfunc(buf.data(), this->write_offset, buf.size())); + } + + this->write_offset += size; + } + + log_write("finished write thread!\n"); + R_SUCCEED(); +} + +void readFunc(void* d) { + auto t = static_cast(d); + t->SetReadResult(t->readFuncInternal()); + log_write("read thread returned now\n"); +} + +void writeFunc(void* d) { + auto t = static_cast(d); + t->SetWriteResult(t->writeFuncInternal()); + log_write("write thread returned now\n"); +} + +auto GetAlternateCore(int id) { + return id == 1 ? 2 : 1; +} + +Result TransferInternal(ui::ProgressBox* pbox, s64 size, ReadFunctionCallback rfunc, WriteFunctionCallback wfunc, StartFunctionCallback sfunc) { + const auto WRITE_THREAD_CORE = sfunc ? pbox->GetCpuId() : GetAlternateCore(pbox->GetCpuId()); + const auto READ_THREAD_CORE = GetAlternateCore(WRITE_THREAD_CORE); + + ThreadData t_data{pbox, size, rfunc, wfunc}; + + Thread t_read{}; + R_TRY(threadCreate(&t_read, readFunc, std::addressof(t_data), nullptr, 1024*64, 0x20, READ_THREAD_CORE)); + ON_SCOPE_EXIT(threadClose(&t_read)); + + Thread t_write{}; + R_TRY(threadCreate(&t_write, writeFunc, std::addressof(t_data), nullptr, 1024*64, 0x20, WRITE_THREAD_CORE)); + ON_SCOPE_EXIT(threadClose(&t_write)); + + log_write("starting threads\n"); + R_TRY(threadStart(std::addressof(t_read))); + ON_SCOPE_EXIT(threadWaitForExit(std::addressof(t_read))); + + R_TRY(threadStart(std::addressof(t_write))); + ON_SCOPE_EXIT(threadWaitForExit(std::addressof(t_write))); + + if (sfunc) { + t_data.SetPullResult(sfunc([&](void* data, s64 size, u64* bytes_read) -> Result { + R_TRY(t_data.GetResults()); + return t_data.Pull(data, size, bytes_read); + })); + } else { + while (t_data.GetWriteOffset() != t_data.GetWriteSize() && R_SUCCEEDED(t_data.GetResults())) { + pbox->UpdateTransfer(t_data.GetWriteOffset(), t_data.GetWriteSize()); + svcSleepThread(1e+6); + } + } + + // wait for all threads to close. + log_write("waiting for threads to close\n"); + for (;;) { + t_data.WakeAllThreads(); + pbox->Yield(); + + if (R_FAILED(waitSingleHandle(t_read.handle, 1000))) { + continue; + } else if (R_FAILED(waitSingleHandle(t_write.handle, 1000))) { + continue; + } + break; + } + log_write("threads closed\n"); + + // if any of the threads failed, wake up all threads so they can exit. + if (R_FAILED(t_data.GetResults())) { + log_write("some reads failed, waking threads\n"); + log_write("returning due to fail\n"); + return t_data.GetResults(); + } + + return t_data.GetResults(); +} + +} // namespace + +Result Transfer(ui::ProgressBox* pbox, s64 size, ReadFunctionCallback rfunc, WriteFunctionCallback wfunc) { + return TransferInternal(pbox, size, rfunc, wfunc, nullptr); +} + +Result TransferPull(ui::ProgressBox* pbox, s64 size, ReadFunctionCallback rfunc, StartFunctionCallback sfunc) { + return TransferInternal(pbox, size, rfunc, nullptr, sfunc); +} + +} // namespace::thread diff --git a/sphaira/source/ui/menus/filebrowser.cpp b/sphaira/source/ui/menus/filebrowser.cpp index a10e9ba..41f8115 100644 --- a/sphaira/source/ui/menus/filebrowser.cpp +++ b/sphaira/source/ui/menus/filebrowser.cpp @@ -19,6 +19,7 @@ #include "swkbd.hpp" #include "i18n.hpp" #include "location.hpp" +#include "threaded_file_transfer.hpp" #include "yati/yati.hpp" #include "yati/source/file.hpp" @@ -1200,36 +1201,66 @@ void Menu::UploadFiles() { App::Push(std::make_shared(0, "Uploading"_i18n, "", [this, loc](auto pbox) -> bool { auto targets = GetSelectedEntries(); - const auto file_add = [&](const fs::FsPath& file_path, const char* name){ + const auto file_add = [&](s64 file_size, const fs::FsPath& file_path, const char* name) -> Result { // the file name needs to be relative to the current directory. const auto relative_file_name = file_path.s + std::strlen(m_path); pbox->SetTitle(name); pbox->NewTransfer(relative_file_name); - const auto result = curl::Api().FromFile( - CURL_LOCATION_TO_API(loc), - curl::Path{file_path}, - curl::OnProgress{pbox->OnDownloadProgressCallback()}, - curl::UploadInfo{relative_file_name} - ); + FsFile file; + R_TRY(m_fs->OpenFile(file_path, FsOpenMode_Read, &file)); + ON_SCOPE_EXIT(fsFileClose(&file)); - return result.success; + return thread::TransferPull(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return fsFileRead(&file, off, data, size, FsReadOption_None, bytes_read); + }, + [&](thread::PullFunctionCallback pull) -> Result { + s64 offset{}; + const auto result = curl::Api().FromMemory( + CURL_LOCATION_TO_API(loc), + curl::OnProgress{pbox->OnDownloadProgressCallback()}, + curl::UploadInfo{ + relative_file_name, file_size, + [&](void *ptr, size_t size) -> size_t { + // curl will request past the size of the file, causing an error. + if (offset >= file_size) { + log_write("finished file upload\n"); + return 0; + } + + u64 bytes_read{}; + if (R_FAILED(pull(ptr, size, &bytes_read))) { + log_write("failed to read in custom callback: %zd size: %zd\n", offset, size); + return 0; + } + + offset += bytes_read; + return bytes_read; + } + } + ); + + R_UNLESS(result.success, 0x1); + R_SUCCEED(); + } + ); }; for (auto& e : targets) { if (e.IsFile()) { const auto file_path = GetNewPath(e); - if (!file_add(file_path, e.GetName().c_str())) { + if (R_FAILED(file_add(e.file_size, file_path, e.GetName().c_str()))) { return false; } } else { FsDirCollections collections; - get_collections(GetNewPath(e), e.name, collections); + get_collections(GetNewPath(e), e.name, collections, true); for (const auto& collection : collections) { for (const auto& file : collection.files) { const auto file_path = fs::AppendPath(collection.path, file.name); - if (!file_add(file_path, file.name)) { + if (R_FAILED(file_add(file.file_size, file_path, file.name))) { return false; } } @@ -1887,10 +1918,10 @@ auto Menu::get_collection(const fs::FsPath& path, const fs::FsPath& parent_name, R_SUCCEED(); } -auto Menu::get_collections(const fs::FsPath& path, const fs::FsPath& parent_name, FsDirCollections& out) -> Result { +auto Menu::get_collections(const fs::FsPath& path, const fs::FsPath& parent_name, FsDirCollections& out, bool inc_size) -> Result { // get a list of all the files / dirs FsDirCollection collection; - R_TRY(get_collection(path, parent_name, collection, true, true, false)); + R_TRY(get_collection(path, parent_name, collection, true, true, inc_size)); log_write("got collection: %s parent_name: %s files: %zu dirs: %zu\n", path.s, parent_name.s, collection.files.size(), collection.dirs.size()); out.emplace_back(collection); @@ -1900,7 +1931,7 @@ auto Menu::get_collections(const fs::FsPath& path, const fs::FsPath& parent_name const auto new_path = std::make_unique(Menu::GetNewPath(path, p.name)); const auto new_parent_name = std::make_unique(Menu::GetNewPath(parent_name, p.name)); log_write("trying to get nested collection: %s parent_name: %s\n", new_path->s, new_parent_name->s); - R_TRY(get_collections(*new_path, *new_parent_name, out)); + R_TRY(get_collections(*new_path, *new_parent_name, out, inc_size)); } R_SUCCEED(); diff --git a/sphaira/source/ui/menus/game_menu.cpp b/sphaira/source/ui/menus/game_menu.cpp index 3df4504..b7a2cba 100644 --- a/sphaira/source/ui/menus/game_menu.cpp +++ b/sphaira/source/ui/menus/game_menu.cpp @@ -5,6 +5,7 @@ #include "defines.hpp" #include "i18n.hpp" #include "location.hpp" +#include "threaded_file_transfer.hpp" #include "ui/menus/game_menu.hpp" #include "ui/sidebar.hpp" @@ -291,15 +292,17 @@ Result DumpNspToFile(ProgressBox* pbox, std::span entries) { auto source = std::make_unique(entries); for (const auto& e : entries) { + auto path = e.path; + const auto file_size = e.nsp_size; pbox->SetTitle(e.application_name); - pbox->NewTransfer(e.path); + pbox->NewTransfer(path); - const auto temp_path = fs::AppendPath(DUMP_PATH, e.path + ".temp"); + const auto temp_path = fs::AppendPath(DUMP_PATH, path + ".temp"); fs.CreateDirectoryRecursivelyWithPath(temp_path); fs.DeleteFile(temp_path); - const auto flags = e.nsp_size >= BIG_FILE_SIZE ? FsCreateOption_BigFile : 0; - R_TRY(fs.CreateFile(temp_path, e.nsp_size, flags)); + const auto flags = file_size >= BIG_FILE_SIZE ? FsCreateOption_BigFile : 0; + R_TRY(fs.CreateFile(temp_path, file_size, flags)); ON_SCOPE_EXIT(fs.DeleteFile(temp_path)); { @@ -307,27 +310,17 @@ Result DumpNspToFile(ProgressBox* pbox, std::span entries) { R_TRY(fs.OpenFile(temp_path, FsOpenMode_Write, &file)); ON_SCOPE_EXIT(fsFileClose(&file)); - s64 offset{}; - std::vector buf(1024*1024*4); // 4MiB - - while (offset < e.nsp_size) { - if (pbox->ShouldExit()) { - R_THROW(0xFFFF); + R_TRY(thread::Transfer(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return source->Read(path, data, off, size, bytes_read); + }, + [&](const void* data, s64 off, s64 size) -> Result { + return fsFileWrite(&file, off, data, size, FsWriteOption_None); } - - u64 bytes_read; - R_TRY(source->Read(e.path, buf.data(), offset, buf.size(), &bytes_read)); - pbox->Yield(); - - R_TRY(fsFileWrite(&file, offset, buf.data(), bytes_read, FsWriteOption_None)); - pbox->Yield(); - - pbox->UpdateTransfer(offset, e.nsp_size); - offset += bytes_read; - } + )); } - const auto path = fs::AppendPath(DUMP_PATH, e.path); + path = fs::AppendPath(DUMP_PATH, path); fs.DeleteFile(path); R_TRY(fs.RenameFile(temp_path, path)); } @@ -373,24 +366,19 @@ Result DumpNspToUsbS2S(ProgressBox* pbox, std::span entries) { Result DumpNspToDevNull(ProgressBox* pbox, std::span entries) { auto source = std::make_unique(entries); for (const auto& e : entries) { + const auto path = e.path; + const auto file_size = e.nsp_size; pbox->SetTitle(e.application_name); - pbox->NewTransfer(e.path); + pbox->NewTransfer(path); - s64 offset{}; - std::vector buf(1024*1024*4); // 4MiB - - while (offset < e.nsp_size) { - if (pbox->ShouldExit()) { - R_THROW(0xFFFF); + R_TRY(thread::Transfer(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return source->Read(path, data, off, size, bytes_read); + }, + [&](const void* data, s64 off, s64 size) -> Result { + R_SUCCEED(); } - - u64 bytes_read; - R_TRY(source->Read(e.path, buf.data(), offset, buf.size(), &bytes_read)); - pbox->Yield(); - - pbox->UpdateTransfer(offset, e.nsp_size); - offset += bytes_read; - } + )); } R_SUCCEED(); @@ -403,39 +391,45 @@ Result DumpNspToNetwork(ProgressBox* pbox, const location::Entry& loc, std::span R_THROW(0xFFFF); } + const auto path = e.path; + const auto file_size = e.nsp_size; pbox->SetTitle(e.application_name); - pbox->NewTransfer(e.path); + pbox->NewTransfer(path); - s64 offset{}; - const auto result = curl::Api().FromMemory( - CURL_LOCATION_TO_API(loc), - curl::OnProgress{pbox->OnDownloadProgressCallback()}, - curl::UploadInfo{ - e.path, e.nsp_size, - [&pbox, &e, &source, &offset](void *ptr, size_t size) -> size_t { - u64 bytes_read{}; - if (R_FAILED(source->Read(e.path, ptr, offset, size, &bytes_read))) { - // curl will request past the size of the file, causing an error. - // only log the error if it failed in the middle of a transfer. - if (offset != e.nsp_size) { - log_write("failed to read in custom callback: %zd size: %zd\n", offset, e.nsp_size); - } - return 0; - } - - offset += bytes_read; - return bytes_read; - } + R_TRY(thread::TransferPull(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return source->Read(path, data, off, size, bytes_read); }, - curl::OnUploadSeek{ - [&e, &offset](s64 new_offset){ - offset = new_offset; - return true; - } - } - ); + [&](thread::PullFunctionCallback pull) -> Result { + s64 offset{}; + const auto result = curl::Api().FromMemory( + CURL_LOCATION_TO_API(loc), + curl::OnProgress{pbox->OnDownloadProgressCallback()}, + curl::UploadInfo{ + path, file_size, + [&](void *ptr, size_t size) -> size_t { + // curl will request past the size of the file, causing an error. + if (offset >= file_size) { + log_write("finished file upload\n"); + return 0; + } - R_UNLESS(result.success, 0x1); + u64 bytes_read{}; + if (R_FAILED(pull(ptr, size, &bytes_read))) { + log_write("failed to read in custom callback: %zd size: %zd\n", offset, size); + return 0; + } + + offset += bytes_read; + return bytes_read; + } + } + ); + + R_UNLESS(result.success, 0x1); + R_SUCCEED(); + } + )); } R_SUCCEED(); diff --git a/sphaira/source/ui/menus/gc_menu.cpp b/sphaira/source/ui/menus/gc_menu.cpp index 42d2f4f..66a6478 100644 --- a/sphaira/source/ui/menus/gc_menu.cpp +++ b/sphaira/source/ui/menus/gc_menu.cpp @@ -13,6 +13,7 @@ #include "i18n.hpp" #include "download.hpp" #include "location.hpp" +#include "threaded_file_transfer.hpp" #include #include @@ -285,6 +286,7 @@ Result DumpNspToFile(ProgressBox* pbox, std::span paths, XciEn R_TRY(fs.GetFsOpenResult()); for (auto path : paths) { + const auto file_size = e.GetSize(path); pbox->SetTitle(e.application_name); pbox->NewTransfer(path); @@ -292,9 +294,8 @@ Result DumpNspToFile(ProgressBox* pbox, std::span paths, XciEn fs.CreateDirectoryRecursivelyWithPath(temp_path); fs.DeleteFile(temp_path); - const auto size = e.GetSize(path); - const auto flags = size >= BIG_FILE_SIZE ? FsCreateOption_BigFile : 0; - R_TRY(fs.CreateFile(temp_path, size, flags)); + const auto flags = file_size >= BIG_FILE_SIZE ? FsCreateOption_BigFile : 0; + R_TRY(fs.CreateFile(temp_path, file_size, flags)); ON_SCOPE_EXIT(fs.DeleteFile(temp_path)); { @@ -302,24 +303,14 @@ Result DumpNspToFile(ProgressBox* pbox, std::span paths, XciEn R_TRY(fs.OpenFile(temp_path, FsOpenMode_Write, &file)); ON_SCOPE_EXIT(fsFileClose(&file)); - s64 offset{}; - std::vector buf(1024*1024*4); // 4MiB - - while (offset < size) { - if (pbox->ShouldExit()) { - R_THROW(0xFFFF); + R_TRY(thread::Transfer(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return e.Read(path, data, off, size, bytes_read); + }, + [&](const void* data, s64 off, s64 size) -> Result { + return fsFileWrite(&file, off, data, size, FsWriteOption_None); } - - u64 bytes_read; - R_TRY(e.Read(path, buf.data(), offset, buf.size(), &bytes_read)); - pbox->Yield(); - - R_TRY(fsFileWrite(&file, offset, buf.data(), bytes_read, FsWriteOption_None)); - pbox->Yield(); - - pbox->UpdateTransfer(offset, size); - offset += bytes_read; - } + )); } path = fs::AppendPath(DUMP_PATH, path); @@ -367,25 +358,18 @@ Result DumpNspToUsbS2S(ProgressBox* pbox, std::span paths, Xci Result DumpNspToDevNull(ProgressBox* pbox, std::span paths, XciEntry& e) { for (const auto& path : paths) { + const auto file_size = e.GetSize(path); pbox->SetTitle(e.application_name); pbox->NewTransfer(path); - s64 offset{}; - const auto size = e.GetSize(path); - std::vector buf(1024*1024*4); // 4MiB - - while (offset < size) { - if (pbox->ShouldExit()) { - R_THROW(0xFFFF); + R_TRY(thread::Transfer(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return e.Read(path, data, off, size, bytes_read); + }, + [&](const void* data, s64 off, s64 size) -> Result { + R_SUCCEED(); } - - u64 bytes_read; - R_TRY(e.Read(path, buf.data(), offset, buf.size(), &bytes_read)); - pbox->Yield(); - - pbox->UpdateTransfer(offset, size); - offset += bytes_read; - } + )); } R_SUCCEED(); @@ -397,41 +381,45 @@ Result DumpNspToNetwork(ProgressBox* pbox, const location::Entry& loc, std::span R_THROW(0xFFFF); } + const auto file_size = e.GetSize(path); pbox->SetTitle(e.application_name); pbox->NewTransfer(path); - s64 offset{}; - const auto size = e.GetSize(path); - - const auto result = curl::Api().FromMemory( - CURL_LOCATION_TO_API(loc), - curl::OnProgress{pbox->OnDownloadProgressCallback()}, - curl::UploadInfo{ - path, size, - [&pbox, &e, &offset, &path](void *ptr, size_t size) -> size_t { - u64 bytes_read{}; - if (R_FAILED(e.Read(path, ptr, offset, size, &bytes_read))) { - // curl will request past the size of the file, causing an error. - // only log the error if it failed in the middle of a transfer. - if (offset != size) { - log_write("failed to read in custom callback: %zd size: %zd\n", offset, size); - } - return 0; - } - - offset += bytes_read; - return bytes_read; - } + R_TRY(thread::TransferPull(pbox, file_size, + [&](void* data, s64 off, s64 size, u64* bytes_read) -> Result { + return e.Read(path, data, off, size, bytes_read); }, - curl::OnUploadSeek{ - [&offset](s64 new_offset){ - offset = new_offset; - return true; - } - } - ); + [&](thread::PullFunctionCallback pull) -> Result { + s64 offset{}; + const auto result = curl::Api().FromMemory( + CURL_LOCATION_TO_API(loc), + curl::OnProgress{pbox->OnDownloadProgressCallback()}, + curl::UploadInfo{ + path, file_size, + [&](void *ptr, size_t size) -> size_t { + // curl will request past the size of the file, causing an error. + if (offset >= file_size) { + log_write("finished file upload\n"); + return 0; + } - R_UNLESS(result.success, 0x1); + u64 bytes_read{}; + if (R_FAILED(pull(ptr, size, &bytes_read))) { + log_write("failed to read in custom callback: %zd size: %zd\n", offset, size); + return 0; + } + + offset += bytes_read; + return bytes_read; + } + } + ); + + + R_UNLESS(result.success, 0x1); + R_SUCCEED(); + } + )); } R_SUCCEED(); diff --git a/sphaira/source/ui/progress_box.cpp b/sphaira/source/ui/progress_box.cpp index 0a0662d..409dd8c 100644 --- a/sphaira/source/ui/progress_box.cpp +++ b/sphaira/source/ui/progress_box.cpp @@ -38,6 +38,7 @@ ProgressBox::ProgressBox(int image, const std::string& action, const std::string m_action = action; m_image = image; + m_cpuid = cpuid; m_thread_data.pbox = this; m_thread_data.callback = callback; if (R_FAILED(threadCreate(&m_thread, threadFunc, &m_thread_data, nullptr, stack_size, prio, cpuid))) {