From 39d32063c169d567407622528252f486946045e3 Mon Sep 17 00:00:00 2001 From: "Vladimir Golovnev (Glassez)" Date: Sat, 12 Sep 2020 11:07:12 +0300 Subject: [PATCH] Implement disk IO thread wrapper --- src/base/bittorrent/customstorage.cpp | 185 ++++++++++++++++++++++++++ src/base/bittorrent/customstorage.h | 69 +++++++++- src/base/bittorrent/session.cpp | 10 +- 3 files changed, 262 insertions(+), 2 deletions(-) diff --git a/src/base/bittorrent/customstorage.cpp b/src/base/bittorrent/customstorage.cpp index 864707821..93c458d50 100644 --- a/src/base/bittorrent/customstorage.cpp +++ b/src/base/bittorrent/customstorage.cpp @@ -35,6 +35,190 @@ #include "base/utils/fs.h" #include "common.h" +#if (LIBTORRENT_VERSION_NUM >= 20000) +#include + +std::unique_ptr customDiskIOConstructor( + lt::io_context &ioContext, const lt::settings_interface &settings, lt::counters &counters) +{ + return std::make_unique(lt::default_disk_io_constructor(ioContext, settings, counters)); +} + +customDiskIOThread::customDiskIOThread(std::unique_ptr nativeDiskIOThread) + : m_nativeDiskIO {std::move(nativeDiskIOThread)} +{ +} + +lt::storage_holder customDiskIOThread::new_torrent(const lt::storage_params &storageParams, const std::shared_ptr &torrent) +{ + lt::storage_holder storageHolder = m_nativeDiskIO->new_torrent(storageParams, torrent); + + const QString savePath = Utils::Fs::expandPathAbs(QString::fromStdString(storageParams.path)); + m_storageData[storageHolder] = { + savePath + , storageParams.mapped_files ? *storageParams.mapped_files : storageParams.files + , storageParams.priorities}; + + return storageHolder; +} + +void customDiskIOThread::remove_torrent(lt::storage_index_t storage) +{ + m_nativeDiskIO->remove_torrent(storage); +} + +void customDiskIOThread::async_read(lt::storage_index_t storage, const lt::peer_request &peerRequest + , std::function handler + , lt::disk_job_flags_t flags) +{ + m_nativeDiskIO->async_read(storage, peerRequest, std::move(handler), flags); +} + +bool customDiskIOThread::async_write(lt::storage_index_t storage, const lt::peer_request &peerRequest + , const char *buf, std::shared_ptr diskObserver + , std::function handler, lt::disk_job_flags_t flags) +{ + return m_nativeDiskIO->async_write(storage, peerRequest, buf, diskObserver, std::move(handler), flags); +} + +void customDiskIOThread::async_hash(lt::storage_index_t storage, lt::piece_index_t piece + , lt::span hash, lt::disk_job_flags_t flags + , std::function handler) +{ + m_nativeDiskIO->async_hash(storage, piece, hash, flags, std::move(handler)); +} + +void customDiskIOThread::async_hash2(lt::storage_index_t storage, lt::piece_index_t piece + , int offset, lt::disk_job_flags_t flags + , std::function handler) +{ + m_nativeDiskIO->async_hash2(storage, piece, offset, flags, std::move(handler)); +} + +void customDiskIOThread::async_move_storage(lt::storage_index_t storage, std::string path, lt::move_flags_t flags + , std::function handler) +{ + const QString newSavePath {Utils::Fs::expandPathAbs(QString::fromStdString(path))}; + + if (flags == lt::move_flags_t::dont_replace) + handleCompleteFiles(storage, newSavePath); + + m_nativeDiskIO->async_move_storage(storage, path, flags + , [=, handler = std::move(handler)](lt::status_t status, const std::string &path, const lt::storage_error &error) + { + if (status != lt::status_t::fatal_disk_error) + m_storageData[storage].savePath = newSavePath; + + handler(status, path, error); + }); +} + +void customDiskIOThread::async_release_files(lt::storage_index_t storage, std::function handler) +{ + m_nativeDiskIO->async_release_files(storage, std::move(handler)); +} + +void customDiskIOThread::async_check_files(lt::storage_index_t storage, const lt::add_torrent_params *resume_data + , lt::aux::vector links + , std::function handler) +{ + handleCompleteFiles(storage, m_storageData[storage].savePath); + m_nativeDiskIO->async_check_files(storage, resume_data, links, std::move(handler)); +} + +void customDiskIOThread::async_stop_torrent(lt::storage_index_t storage, std::function handler) +{ + m_nativeDiskIO->async_stop_torrent(storage, std::move(handler)); +} + +void customDiskIOThread::async_rename_file(lt::storage_index_t storage, lt::file_index_t index, std::string name + , std::function handler) +{ + m_nativeDiskIO->async_rename_file(storage, index, name + , [=, handler = std::move(handler)](const std::string &name, lt::file_index_t index, const lt::storage_error &error) + { + if (!error) + m_storageData[storage].files.rename_file(index, name); + handler(name, index, error); + }); +} + +void customDiskIOThread::async_delete_files(lt::storage_index_t storage, lt::remove_flags_t options + , std::function handler) +{ + m_nativeDiskIO->async_delete_files(storage, options, std::move(handler)); +} + +void customDiskIOThread::async_set_file_priority(lt::storage_index_t storage, lt::aux::vector priorities + , std::function)> handler) +{ + m_nativeDiskIO->async_set_file_priority(storage, priorities + , [=, handler = std::move(handler)](const lt::storage_error &error, lt::aux::vector priorities) + { + m_storageData[storage].filePriorities = priorities; + handler(error, priorities); + }); +} + +void customDiskIOThread::async_clear_piece(lt::storage_index_t storage, lt::piece_index_t index + , std::function handler) +{ + m_nativeDiskIO->async_clear_piece(storage, index, std::move(handler)); +} + +void customDiskIOThread::update_stats_counters(lt::counters &counters) const +{ + m_nativeDiskIO->update_stats_counters(counters); +} + +std::vector customDiskIOThread::get_status(lt::storage_index_t index) const +{ + return m_nativeDiskIO->get_status(index); +} + +void customDiskIOThread::abort(bool wait) +{ + m_nativeDiskIO->abort(wait); +} + +void customDiskIOThread::submit_jobs() +{ + m_nativeDiskIO->submit_jobs(); +} + +void customDiskIOThread::settings_updated() +{ + m_nativeDiskIO->settings_updated(); +} + +void customDiskIOThread::handleCompleteFiles(lt::storage_index_t storage, const QString &savePath) +{ + const QDir saveDir {savePath}; + const StorageData storageData = m_storageData[storage]; + const lt::file_storage &fileStorage = storageData.files; + for (const lt::file_index_t fileIndex : fileStorage.file_range()) { + // ignore files that have priority 0 + if ((storageData.filePriorities.end_index() > fileIndex) && (storageData.filePriorities[fileIndex] == lt::dont_download)) + continue; + + // ignore pad files + if (fileStorage.pad_file_at(fileIndex)) continue; + + const QString filePath = QString::fromStdString(fileStorage.file_path(fileIndex)); + if (filePath.endsWith(QB_EXT)) { + const QString completeFilePath = filePath.left(filePath.size() - QB_EXT.size()); + QFile completeFile {saveDir.absoluteFilePath(completeFilePath)}; + if (completeFile.exists()) { + QFile incompleteFile {saveDir.absoluteFilePath(filePath)}; + incompleteFile.remove(); + completeFile.rename(incompleteFile.fileName()); + } + } + } +} + +#else + lt::storage_interface *customStorageConstructor(const lt::storage_params ¶ms, lt::file_pool &pool) { return new CustomStorage {params, pool}; @@ -98,3 +282,4 @@ void CustomStorage::handleCompleteFiles(const QString &savePath) } } } +#endif diff --git a/src/base/bittorrent/customstorage.h b/src/base/bittorrent/customstorage.h index a55640eed..a23d1679b 100644 --- a/src/base/bittorrent/customstorage.h +++ b/src/base/bittorrent/customstorage.h @@ -30,10 +30,76 @@ #include #include -#include +#include #include +#if (LIBTORRENT_VERSION_NUM >= 20000) +#include +#include +#include + +#include +#else +#include +#endif + +#if (LIBTORRENT_VERSION_NUM >= 20000) +std::unique_ptr customDiskIOConstructor( + lt::io_context &ioContext, lt::settings_interface const &settings, lt::counters &counters); + +class customDiskIOThread final : public lt::disk_interface +{ +public: + explicit customDiskIOThread(std::unique_ptr nativeDiskIOThread); + + lt::storage_holder new_torrent(const lt::storage_params &storageParams, const std::shared_ptr &torrent) override; + void remove_torrent(lt::storage_index_t storageIndex) override; + void async_read(lt::storage_index_t storageIndex, const lt::peer_request &peerRequest + , std::function handler + , lt::disk_job_flags_t flags) override; + bool async_write(lt::storage_index_t storageIndex, const lt::peer_request &peerRequest + , const char *buf, std::shared_ptr diskObserver + , std::function handler, lt::disk_job_flags_t flags) override; + void async_hash(lt::storage_index_t storageIndex, lt::piece_index_t piece, lt::span hash, lt::disk_job_flags_t flags + , std::function handler) override; + void async_hash2(lt::storage_index_t storage, lt::piece_index_t piece, int offset, lt::disk_job_flags_t flags + , std::function handler) override; + void async_move_storage(lt::storage_index_t storage, std::string path, lt::move_flags_t flags + , std::function handler) override; + void async_release_files(lt::storage_index_t storage, std::function handler) override; + void async_check_files(lt::storage_index_t storage, const lt::add_torrent_params *resume_data + , lt::aux::vector links + , std::function handler) override; + void async_stop_torrent(lt::storage_index_t storage, std::function handler) override; + void async_rename_file(lt::storage_index_t storage, lt::file_index_t index, std::string name + , std::function handler) override; + void async_delete_files(lt::storage_index_t storage, lt::remove_flags_t options, std::function handler) override; + void async_set_file_priority(lt::storage_index_t storage, lt::aux::vector priorities + , std::function)> handler) override; + void async_clear_piece(lt::storage_index_t storage, lt::piece_index_t index, std::function handler) override; + void update_stats_counters(lt::counters &counters) const override; + std::vector get_status(lt::storage_index_t index) const override; + void abort(bool wait) override; + void submit_jobs() override; + void settings_updated() override; + +private: + void handleCompleteFiles(libtorrent::storage_index_t storage, const QString &savePath); + + std::unique_ptr m_nativeDiskIO; + + struct StorageData + { + QString savePath; + lt::file_storage files; + lt::aux::vector filePriorities; + }; + QHash m_storageData; +}; + +#else + lt::storage_interface *customStorageConstructor(const lt::storage_params ¶ms, lt::file_pool &pool); class CustomStorage final : public lt::default_storage @@ -51,3 +117,4 @@ private: lt::aux::vector m_filePriorities; QString m_savePath; }; +#endif diff --git a/src/base/bittorrent/session.cpp b/src/base/bittorrent/session.cpp index afbd0b102..484926f33 100644 --- a/src/base/bittorrent/session.cpp +++ b/src/base/bittorrent/session.cpp @@ -1059,7 +1059,11 @@ void Session::initializeNativeSession() #endif loadLTSettings(pack); - m_nativeSession = new lt::session {lt::session_params {pack, {}}}; + lt::session_params sessionParams {pack, {}}; +#if (LIBTORRENT_VERSION_NUM >= 20000) + sessionParams.disk_io_constructor = customDiskIOConstructor; +#endif + m_nativeSession = new lt::session {sessionParams}; LogMsg(tr("Peer ID: ") + QString::fromStdString(peerId)); LogMsg(tr("HTTP User-Agent is '%1'").arg(USER_AGENT)); @@ -2103,7 +2107,9 @@ bool Session::loadTorrent(LoadTorrentParams params) { lt::add_torrent_params &p = params.ltAddTorrentParams; +#if (LIBTORRENT_VERSION_NUM < 20000) p.storage = customStorageConstructor; +#endif // Limits p.max_connections = maxConnectionsPerTorrent(); p.max_uploads = maxUploadsPerTorrent(); @@ -2189,7 +2195,9 @@ bool Session::loadMetadata(const MagnetUri &magnetUri) // Solution to avoid accidental file writes p.flags |= lt::torrent_flags::upload_mode; +#if (LIBTORRENT_VERSION_NUM < 20000) p.storage = customStorageConstructor; +#endif // Adding torrent to BitTorrent session lt::error_code ec;