diff --git a/include/curl/DownloadStruct.hpp b/include/curl/DownloadStruct.hpp index ac0e459..19ab607 100644 --- a/include/curl/DownloadStruct.hpp +++ b/include/curl/DownloadStruct.hpp @@ -4,13 +4,21 @@ #include #include -#include #include +#include +#include namespace curl { inline constexpr size_t SHARED_BUFFER_SIZE = 0x500000; + // This is for synchronizing the buffer. + enum class BufferState + { + Empty, + Full + }; + // clang-format off struct DownloadStruct : sys::threadpool::DataStruct { @@ -21,13 +29,10 @@ namespace curl std::condition_variable condition{}; /// @brief Shared buffer that is read into. - std::array sharedBuffer{}; - - /// @brief Current offset in the shared buffer. - size_t sharedOffset{}; + std::vector sharedBuffer{}; /// @brief Bool to signal when the buffer is ready/empty. - bool bufferReady{}; + BufferState bufferState{}; /// @brief Destination file to write to. fslib::File *dest{}; @@ -40,6 +45,9 @@ namespace curl /// @brief Size of the file being downloaded. int64_t fileSize{}; + + /// @brief Signals when the write thread is complete. + std::binary_semaphore writeComplete{0}; }; // clang-format on @@ -47,10 +55,11 @@ namespace curl sys::ProgressTask *task, int64_t fileSize) { - auto downloadStruct = std::make_shared(); - downloadStruct->dest = &dest; - downloadStruct->task = task; - downloadStruct->offset = fileSize; + auto downloadStruct = std::make_shared(); + downloadStruct->dest = &dest; + downloadStruct->task = task; + downloadStruct->fileSize = fileSize; + downloadStruct->sharedBuffer.reserve(SHARED_BUFFER_SIZE); return downloadStruct; } } diff --git a/include/sys/defines.hpp b/include/sys/defines.hpp index 501b6d6..250963a 100644 --- a/include/sys/defines.hpp +++ b/include/sys/defines.hpp @@ -2,5 +2,5 @@ namespace sys { - using byte = unsigned char; + using Byte = unsigned char; } diff --git a/source/curl/curl.cpp b/source/curl/curl.cpp index ad75103..105addb 100644 --- a/source/curl/curl.cpp +++ b/source/curl/curl.cpp @@ -70,29 +70,25 @@ size_t curl::download_file_threaded(const char *buffer, size_t size, size_t coun std::mutex &lock = download->lock; std::condition_variable &condition = download->condition; auto &sharedBuffer = download->sharedBuffer; - size_t &sharedOffset = download->sharedOffset; - bool &bufferReady = download->bufferReady; + curl::BufferState &bufferState = download->bufferState; sys::ProgressTask *task = download->task; size_t &offset = download->offset; - int64_t &fileSize = download->fileSize; + int64_t fileSize = download->fileSize; const size_t downloadSize = size * count; - const std::span bufferSpan{reinterpret_cast(buffer), downloadSize}; { - std::unique_lock bufferLock(lock); - condition.wait(bufferLock, [&]() { return bufferReady == false; }); + std::unique_lock bufferLock{lock}; + condition.wait(bufferLock, [&]() { return bufferState == curl::BufferState::Empty; }); - std::copy(bufferSpan.begin(), bufferSpan.end(), &sharedBuffer[sharedOffset]); - sharedOffset += downloadSize; + sharedBuffer.insert(sharedBuffer.end(), buffer, buffer + downloadSize); const int64_t nextOffset = offset + downloadSize; - if (sharedOffset >= SIZE_DOWNLOAD_THRESHOLD || nextOffset >= fileSize) + if (sharedBuffer.size() >= SIZE_DOWNLOAD_THRESHOLD || nextOffset >= fileSize) { - bufferReady = true; + bufferState = curl::BufferState::Full; condition.notify_one(); } - offset += downloadSize; } @@ -108,31 +104,33 @@ void curl::download_write_thread_function(sys::threadpool::JobData jobData) std::mutex &lock = castData->lock; std::condition_variable &condition = castData->condition; auto &sharedBuffer = castData->sharedBuffer; - size_t &sharedOffset = castData->sharedOffset; - bool &bufferReady = castData->bufferReady; + curl::BufferState &bufferState = castData->bufferState; fslib::File &dest = *castData->dest; - size_t fileSize = castData->fileSize; + int64_t fileSize = castData->fileSize; + auto &writeComplete = castData->writeComplete; - auto localBuffer = std::make_unique(SIZE_DOWNLOAD_THRESHOLD + 0x100000); // Gonna give this some room. + std::vector localBuffer{}; + localBuffer.reserve(curl::SHARED_BUFFER_SIZE); - for (size_t i = 0; i < fileSize;) + for (int64_t i = 0; i < fileSize;) { - size_t bufferSize{}; { - std::unique_lock bufferLock(lock); - condition.wait(bufferLock, [&]() { return bufferReady == true; }); + std::unique_lock bufferLock{lock}; + condition.wait(bufferLock, [&]() { return bufferState == curl::BufferState::Full; }); // Copy and reset the offset. - bufferSize = sharedOffset; - std::copy(sharedBuffer.begin(), sharedBuffer.begin() + sharedOffset, localBuffer.get()); - sharedOffset = 0; + localBuffer.assign_range(sharedBuffer); + sharedBuffer.clear(); - bufferReady = false; + bufferState = curl::BufferState::Empty; condition.notify_one(); } - dest.write(localBuffer.get(), bufferSize); - i += bufferSize; + + dest.write(localBuffer.data(), localBuffer.size()); + i += localBuffer.size(); } + + writeComplete.release(); } bool curl::get_header_value(const curl::HeaderArray &array, std::string_view header, std::string &valueOut) diff --git a/source/fs/io.cpp b/source/fs/io.cpp index b79b495..ba267b6 100644 --- a/source/fs/io.cpp +++ b/source/fs/io.cpp @@ -13,6 +13,7 @@ #include #include #include +#include namespace { @@ -29,8 +30,9 @@ struct FileThreadStruct : sys::threadpool::DataStruct std::condition_variable condition{}; bool bufferReady{}; ssize_t readSize{}; - std::unique_ptr sharedBuffer{}; + std::unique_ptr sharedBuffer{}; fslib::File *source{}; + std::binary_semaphore writeComplete{0}; }; // clang-format on @@ -42,8 +44,9 @@ static void read_thread_function(sys::threadpool::JobData jobData) std::condition_variable &condition = castData->condition; bool &bufferReady = castData->bufferReady; ssize_t &readSize = castData->readSize; - std::unique_ptr &sharedBuffer = castData->sharedBuffer; + std::unique_ptr &sharedBuffer = castData->sharedBuffer; fslib::File &source = *castData->source; + auto &writeComplete = castData->writeComplete; const int64_t fileSize = source.get_size(); for (int64_t i = 0; i < fileSize;) @@ -62,6 +65,8 @@ static void read_thread_function(sys::threadpool::JobData jobData) if (localRead == -1) { break; } i += localRead; } + + writeComplete.release(); } void fs::copy_file(const fslib::Path &source, const fslib::Path &destination, sys::ProgressTask *task) @@ -82,16 +87,17 @@ void fs::copy_file(const fslib::Path &source, const fslib::Path &destination, sy } auto sharedData = std::make_shared(); - sharedData->sharedBuffer = std::make_unique(SIZE_FILE_BUFFER); + sharedData->sharedBuffer = std::make_unique(SIZE_FILE_BUFFER); sharedData->source = &sourceFile; - auto localBuffer = std::make_unique(SIZE_FILE_BUFFER); + auto localBuffer = std::make_unique(SIZE_FILE_BUFFER); std::mutex &lock = sharedData->lock; std::condition_variable &condition = sharedData->condition; bool &bufferReady = sharedData->bufferReady; ssize_t &readSize = sharedData->readSize; auto &sharedBuffer = sharedData->sharedBuffer; + auto &writeComplete = sharedData->writeComplete; sys::threadpool::push_job(read_thread_function, sharedData); for (int64_t i = 0; i < sourceSize; i++) @@ -115,6 +121,8 @@ void fs::copy_file(const fslib::Path &source, const fslib::Path &destination, sy i += localRead; if (task) { task->update_current(static_cast(i)); } } + + writeComplete.acquire(); } void fs::copy_file_commit(const fslib::Path &source, @@ -140,16 +148,17 @@ void fs::copy_file_commit(const fslib::Path &source, } auto sharedData = std::make_shared(); - sharedData->sharedBuffer = std::make_unique(SIZE_FILE_BUFFER); + sharedData->sharedBuffer = std::make_unique(SIZE_FILE_BUFFER); sharedData->source = &sourceFile; - auto localBuffer = std::make_unique(SIZE_FILE_BUFFER); + auto localBuffer = std::make_unique(SIZE_FILE_BUFFER); std::mutex &lock = sharedData->lock; std::condition_variable &condition = sharedData->condition; bool &bufferReady = sharedData->bufferReady; ssize_t &readSize = sharedData->readSize; auto &sharedBuffer = sharedData->sharedBuffer; + auto &writeComplete = sharedData->writeComplete; int64_t journalCount{}; sys::threadpool::push_job(read_thread_function, sharedData); @@ -187,7 +196,9 @@ void fs::copy_file_commit(const fslib::Path &source, journalCount += localRead; if (task) { task->update_current(static_cast(i)); } } + destFile.close(); + writeComplete.acquire(); const bool commitError = error::fslib(fslib::commit_data_to_file_system(destination.get_device_name())); if (commitError) { ui::PopMessageManager::push_message(popTicks, popCommitFailed); } diff --git a/source/fs/zip.cpp b/source/fs/zip.cpp index 9ae30e5..771cd5b 100644 --- a/source/fs/zip.cpp +++ b/source/fs/zip.cpp @@ -32,7 +32,7 @@ struct ZipIOBase : sys::threadpool::DataStruct std::condition_variable condition{}; ssize_t readSize{}; bool bufferReady{}; - std::unique_ptr sharedBuffer{}; + std::unique_ptr sharedBuffer{}; }; struct ZipReadStruct : ZipIOBase @@ -55,7 +55,7 @@ static void zip_read_thread_function(sys::threadpool::JobData jobData) std::condition_variable &condition = castData->condition; ssize_t &readSize = castData->readSize; bool &bufferReady = castData->bufferReady; - std::unique_ptr &sharedBuffer = castData->sharedBuffer; + std::unique_ptr &sharedBuffer = castData->sharedBuffer; fslib::File &source = *castData->source; const int64_t fileSize = source.get_size(); @@ -86,7 +86,7 @@ static void unzip_read_thread_function(sys::threadpool::JobData jobData) std::condition_variable &condition = castData->condition; ssize_t &readSize = castData->readSize; bool &bufferReady = castData->bufferReady; - std::unique_ptr &sharedBuffer = castData->sharedBuffer; + std::unique_ptr &sharedBuffer = castData->sharedBuffer; fs::MiniUnzip &unzip = *castData->unzip; const int64_t fileSize = unzip.get_uncompressed_size(); @@ -133,9 +133,9 @@ void fs::copy_directory_to_zip(const fslib::Path &source, fs::MiniZip &dest, sys const int64_t fileSize = sourceFile.get_size(); auto sharedData = std::make_shared(); sharedData->source = &sourceFile; - sharedData->sharedBuffer = std::make_unique(SIZE_ZIP_BUFFER); + sharedData->sharedBuffer = std::make_unique(SIZE_ZIP_BUFFER); - auto localBuffer = std::make_unique(SIZE_ZIP_BUFFER); + auto localBuffer = std::make_unique(SIZE_ZIP_BUFFER); if (task) { @@ -222,16 +222,16 @@ void fs::copy_zip_to_directory(fs::MiniUnzip &unzip, const fslib::Path &dest, in } auto sharedData = std::make_shared(); - sharedData->sharedBuffer = std::make_unique(SIZE_UNZIP_BUFFER); + sharedData->sharedBuffer = std::make_unique(SIZE_UNZIP_BUFFER); sharedData->unzip = &unzip; - auto localBuffer = std::make_unique(SIZE_UNZIP_BUFFER); + auto localBuffer = std::make_unique(SIZE_UNZIP_BUFFER); std::mutex &lock = sharedData->lock; std::condition_variable &condition = sharedData->condition; ssize_t &readSize = sharedData->readSize; bool &bufferReady = sharedData->bufferReady; - std::unique_ptr &sharedBuffer = sharedData->sharedBuffer; + std::unique_ptr &sharedBuffer = sharedData->sharedBuffer; int64_t journalCount{}; sys::threadpool::push_job(unzip_read_thread_function, sharedData); diff --git a/source/remote/GoogleDrive.cpp b/source/remote/GoogleDrive.cpp index 1771afc..cf6fa08 100644 --- a/source/remote/GoogleDrive.cpp +++ b/source/remote/GoogleDrive.cpp @@ -307,10 +307,7 @@ bool remote::GoogleDrive::download_file(const remote::Item *file, const fslib::P remote::URL url{URL_DRIVE_FILE_API}; url.append_path(file->get_id()).append_parameter("alt", "media"); - auto download = std::make_shared(); - download->dest = &destFile; - download->task = task; - download->fileSize = itemSize; + auto download = curl::create_download_struct(destFile, task, itemSize); curl::prepare_get(m_curl); curl::set_option(m_curl, CURLOPT_HTTPHEADER, header.get()); @@ -321,6 +318,8 @@ bool remote::GoogleDrive::download_file(const remote::Item *file, const fslib::P sys::threadpool::push_job(curl::download_write_thread_function, download); if (!curl::perform(m_curl)) { return false; } + download->writeComplete.acquire(); + return true; } diff --git a/source/remote/WebDav.cpp b/source/remote/WebDav.cpp index 2420d73..f1c317a 100644 --- a/source/remote/WebDav.cpp +++ b/source/remote/WebDav.cpp @@ -202,11 +202,7 @@ bool remote::WebDav::download_file(const remote::Item *item, const fslib::Path & remote::URL url{m_origin}; url.append_path(item->get_id()); - auto download = std::make_shared(); - download->dest = &destFile; - download->task = task; - download->fileSize = itemSize; - + auto download = curl::create_download_struct(destFile, task, itemSize); curl::reset_handle(m_curl); WebDav::append_credentials(); curl::set_option(m_curl, CURLOPT_HTTPGET, 1L); @@ -220,6 +216,8 @@ bool remote::WebDav::download_file(const remote::Item *item, const fslib::Path & sys::threadpool::push_job(curl::download_write_thread_function, download); if (!curl::perform(m_curl)) { return false; } + download->writeComplete.acquire(); + return true; } diff --git a/source/sys/threadpool.cpp b/source/sys/threadpool.cpp index a0b6d41..3ca1ac0 100644 --- a/source/sys/threadpool.cpp +++ b/source/sys/threadpool.cpp @@ -44,7 +44,7 @@ void sys::threadpool::initialize() for (size_t i = 0; i < COUNT_THREADS; i++) { // NOTE: If pool size increases, i + 1 isn't going to work anymore. - error::libnx(threadCreate(&s_threads[i], thread_pool_function, nullptr, nullptr, SIZE_THREAD_STACK, 0x2C, i + 1)); + error::libnx(threadCreate(&s_threads[i], thread_pool_function, nullptr, nullptr, SIZE_THREAD_STACK, 0x2B, i + 1)); error::libnx(threadStart(&s_threads[i])); } } @@ -64,7 +64,7 @@ void sys::threadpool::push_job(sys::threadpool::JobFunction function, sys::threa { std::lock_guard jobGuard{s_jobMutex}; s_jobQueue.push(std::make_pair(function, data)); - s_jobCondition.notify_one(); + s_jobCondition.notify_all(); } static void thread_pool_function(void *) diff --git a/source/tasks/backup.cpp b/source/tasks/backup.cpp index 0107983..b565de3 100644 --- a/source/tasks/backup.cpp +++ b/source/tasks/backup.cpp @@ -28,27 +28,14 @@ static fs::ScopedSaveMount create_scoped_mount(const FsSaveDataInfo *saveInfo); void tasks::backup::create_new_backup_local(sys::threadpool::JobData taskData) { - logger::log("create_new_backup_local"); auto castData = std::static_pointer_cast(taskData); - logger::log("static_pointer_cast"); - - sys::ProgressTask *task = static_cast(castData->task); - logger::log("task"); - - data::User *user = castData->user; - logger::log("user"); - - data::TitleInfo *titleInfo = castData->titleInfo; - logger::log("titleInfo"); - - const fslib::Path &target = castData->path; - logger::log("path"); + sys::ProgressTask *task = static_cast(castData->task); + data::User *user = castData->user; + data::TitleInfo *titleInfo = castData->titleInfo; + const fslib::Path &target = castData->path; BackupMenuState *spawningState = castData->spawningState; - logger::log("spawningState"); - - const bool killTask = castData->killTask; - logger::log("casts & references"); + const bool killTask = castData->killTask; if (error::is_null(task)) { return; } if (error::is_null(user) || error::is_null(titleInfo)) { TASK_FINISH_RETURN(task); }