diff --git a/conf/lms.conf b/conf/lms.conf index 1c2dfa12..0ae9ebea 100644 --- a/conf/lms.conf +++ b/conf/lms.conf @@ -118,5 +118,5 @@ scanner-skip-duplicate-mbid = false; # Scanner read style for metadata, may be 'fast', 'average' or 'accurate' scanner-parser-read-style = "average"; -# Number of threads to use for scanning file metadata (0 means number of logical CPUs / 2) -scanner-metadata-thread-count = 0; \ No newline at end of file +# Number of threads to use for parallelized tasks (e.g., scanning file metadata). 0 means half the number of logical CPUs. +scanner-thread-count = 0; \ No newline at end of file diff --git a/src/libs/core/CMakeLists.txt b/src/libs/core/CMakeLists.txt index 8ce177b5..0eb0675e 100644 --- a/src/libs/core/CMakeLists.txt +++ b/src/libs/core/CMakeLists.txt @@ -10,6 +10,7 @@ add_library(lmscore STATIC impl/ChildProcessManager.cpp impl/Config.cpp impl/FileResourceHandler.cpp + impl/JobScheduler.cpp impl/IOContextRunner.cpp impl/Logger.cpp impl/MimeTypes.cpp diff --git a/src/libs/core/impl/JobScheduler.cpp b/src/libs/core/impl/JobScheduler.cpp new file mode 100644 index 00000000..49e97b46 --- /dev/null +++ b/src/libs/core/impl/JobScheduler.cpp @@ -0,0 +1,129 @@ +/* + * Copyright (C) 2025 Emeric Poupon + * + * This file is part of LMS. + * + * LMS is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * LMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with LMS. If not, see . + */ + +#include "JobScheduler.hpp" + +#include + +#include "core/ITraceLogger.hpp" + +#include "core/IJob.hpp" + +namespace lms::core +{ + std::unique_ptr createJobScheduler(core::LiteralString name, std::size_t threadCount) + { + return std::make_unique(name, threadCount); + } + + JobScheduler::JobScheduler(core::LiteralString name, std::size_t threadCount) + : _name{ name } + , _ioContextRunner{ _ioContext, threadCount, name.str() } + { + } + + JobScheduler::~JobScheduler() = default; + + void JobScheduler::setShouldAbortCallback(ShouldAbortCallback callback) + { + _abortCallback = callback; + } + + std::size_t JobScheduler::getThreadCount() const + { + return _ioContextRunner.getThreadCount(); + } + + void JobScheduler::scheduleJob(std::unique_ptr job) + { + { + std::scoped_lock lock{ _mutex }; + _ongoingJobCount += 1; + } + + auto jobHandler{ [job = std::move(job), this]() mutable { + if (_abortCallback && _abortCallback()) + { + std::scoped_lock lock{ _mutex }; + _ongoingJobCount -= 1; + } + else + { + { + LMS_SCOPED_TRACE_OVERVIEW(_name, job->getName()); + job->run(); + } + + { + std::scoped_lock lock{ _mutex }; + + _doneJobs.emplace_back(std::move(job)); + _ongoingJobCount -= 1; + } + } + + _condVar.notify_all(); + } }; + + boost::asio::post(_ioContext, std::move(jobHandler)); + } + + std::size_t JobScheduler::getJobsDoneCount() const + { + std::scoped_lock lock{ _mutex }; + return _doneJobs.size(); + } + + size_t JobScheduler::popJobsDone(std::vector>& doneJobs, std::size_t maxCount) + { + doneJobs.clear(); + doneJobs.reserve(maxCount); + + { + std::scoped_lock lock{ _mutex }; + + while (doneJobs.size() < maxCount && !_doneJobs.empty()) + { + doneJobs.push_back(std::move(_doneJobs.front())); + _doneJobs.pop_front(); + } + } + + return doneJobs.size(); + } + + void JobScheduler::waitUntilJobCountAtMost(std::size_t maxOngoingJobs) + { + if (_ongoingJobCount <= maxOngoingJobs) + return; + + { + LMS_SCOPED_TRACE_OVERVIEW(_name, "WaitJobs"); + + std::unique_lock lock{ _mutex }; + _condVar.wait(lock, [=, this] { return _ongoingJobCount <= maxOngoingJobs; }); + } + } + + void JobScheduler::wait() + { + waitUntilJobCountAtMost(0); + } + +} // namespace lms::core diff --git a/src/libs/core/impl/JobScheduler.hpp b/src/libs/core/impl/JobScheduler.hpp new file mode 100644 index 00000000..2511fef6 --- /dev/null +++ b/src/libs/core/impl/JobScheduler.hpp @@ -0,0 +1,61 @@ +/* + * Copyright (C) 2025 Emeric Poupon + * + * This file is part of LMS. + * + * LMS is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * LMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with LMS. If not, see . + */ + +#include +#include +#include +#include + +#include "core/IJobScheduler.hpp" +#include "core/IOContextRunner.hpp" + +namespace lms::core +{ + class JobScheduler : public IJobScheduler + { + public: + JobScheduler(core::LiteralString name, std::size_t threadCount); + ~JobScheduler() override; + JobScheduler(const JobScheduler&) = delete; + JobScheduler& operator=(const JobScheduler&) = delete; + + private: + void setShouldAbortCallback(ShouldAbortCallback callback) override; + + std::size_t getThreadCount() const override; + void scheduleJob(std::unique_ptr job) override; + + std::size_t getJobsDoneCount() const override; + size_t popJobsDone(std::vector>& jobs, std::size_t maxCount) override; + + void waitUntilJobCountAtMost(std::size_t maxOngoingJobs) override; + void wait() override; + + core::LiteralString _name; + boost::asio::io_context _ioContext; + core::IOContextRunner _ioContextRunner; + + ShouldAbortCallback _abortCallback; + + mutable std::mutex _mutex; + std::atomic _ongoingJobCount; + std::deque> _doneJobs; + std::condition_variable _condVar; + }; +} // namespace lms::core \ No newline at end of file diff --git a/src/libs/core/include/core/IJob.hpp b/src/libs/core/include/core/IJob.hpp new file mode 100644 index 00000000..eba2ea44 --- /dev/null +++ b/src/libs/core/include/core/IJob.hpp @@ -0,0 +1,32 @@ +/* + * Copyright (C) 2025 Emeric Poupon + * + * This file is part of LMS. + * + * LMS is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * LMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with LMS. If not, see . + */ + +#include "core/LiteralString.hpp" + +namespace lms::core +{ + class IJob + { + public: + virtual ~IJob() = default; + + virtual LiteralString getName() const = 0; + virtual void run() = 0; + }; +} // namespace lms::core \ No newline at end of file diff --git a/src/libs/core/include/core/IJobScheduler.hpp b/src/libs/core/include/core/IJobScheduler.hpp new file mode 100644 index 00000000..f8eed21e --- /dev/null +++ b/src/libs/core/include/core/IJobScheduler.hpp @@ -0,0 +1,48 @@ +/* + * Copyright (C) 2025 Emeric Poupon + * + * This file is part of LMS. + * + * LMS is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * LMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with LMS. If not, see . + */ + +#include +#include +#include + +#include "core/LiteralString.hpp" + +namespace lms::core +{ + class IJob; + class IJobScheduler + { + public: + virtual ~IJobScheduler() = default; + + using ShouldAbortCallback = std::function; + virtual void setShouldAbortCallback(ShouldAbortCallback callback) = 0; + + virtual std::size_t getThreadCount() const = 0; + virtual void scheduleJob(std::unique_ptr job) = 0; + + virtual std::size_t getJobsDoneCount() const = 0; + virtual size_t popJobsDone(std::vector>& jobs, std::size_t maxCount) = 0; + + virtual void waitUntilJobCountAtMost(std::size_t maxOngoingJobs) = 0; + virtual void wait() = 0; + }; + + std::unique_ptr createJobScheduler(core::LiteralString name, std::size_t threadCount); +} // namespace lms::core \ No newline at end of file diff --git a/src/libs/core/test/CMakeLists.txt b/src/libs/core/test/CMakeLists.txt index 3e89617a..82ee61a3 100644 --- a/src/libs/core/test/CMakeLists.txt +++ b/src/libs/core/test/CMakeLists.txt @@ -2,6 +2,7 @@ include(GoogleTest) add_executable(test-core EnumSet.cpp + JobScheduler.cpp LiteralString.cpp PartialDateTime.cpp Path.cpp diff --git a/src/libs/core/test/JobScheduler.cpp b/src/libs/core/test/JobScheduler.cpp new file mode 100644 index 00000000..d1cedaa5 --- /dev/null +++ b/src/libs/core/test/JobScheduler.cpp @@ -0,0 +1,90 @@ +/* + * Copyright (C) 2025 Emeric Poupon + * + * This file is part of LMS. + * + * LMS is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * LMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with LMS. If not, see . + */ + +#include + +#include + +#include "core/IJob.hpp" +#include "core/IJobScheduler.hpp" + +namespace lms::core +{ + namespace + { + class TestJob : public IJob + { + public: + TestJob(std::atomic& count) + : workCount(count) {} + + private: + LiteralString getName() const override { return "test"; }; + void run() override + { + // Simulate some work + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + ++workCount; + } + + std::atomic& workCount; + }; + + } // namespace + TEST(JobScheduler, basic) + { + std::atomic workCount{ 0 }; + + auto scheduler{ createJobScheduler("TestScheduler", 2) }; + ASSERT_NE(scheduler, nullptr); + + for (int i = 0; i < 10; ++i) + scheduler->scheduleJob(std::make_unique(workCount)); + + // Wait for all jobs to complete + scheduler->wait(); + EXPECT_EQ(workCount.load(), 10); + + std::vector> doneJobs; + scheduler->popJobsDone(doneJobs, 10); + EXPECT_EQ(doneJobs.size(), 10); + } + + TEST(JobScheduler, abort) + { + std::atomic workCount{ 0 }; + + auto scheduler{ createJobScheduler("TestScheduler", 2) }; + ASSERT_NE(scheduler, nullptr); + + scheduler->setShouldAbortCallback([] { return true; }); + + for (int i = 0; i < 10; ++i) + scheduler->scheduleJob(std::make_unique(workCount)); + + // Wait for all jobs to complete + scheduler->wait(); + EXPECT_EQ(workCount.load(), 0); // nothing was done + + std::vector> doneJobs; + scheduler->popJobsDone(doneJobs, 10); + EXPECT_EQ(doneJobs.size(), 0); + } + +} // namespace lms::core \ No newline at end of file diff --git a/src/libs/database/include/database/Session.hpp b/src/libs/database/include/database/Session.hpp index 663231be..88b0c7d9 100644 --- a/src/libs/database/include/database/Session.hpp +++ b/src/libs/database/include/database/Session.hpp @@ -107,8 +107,14 @@ namespace lms::db void refreshTracingLoggerStats(); // returning a ptr here to ease further wrapping using operator-> - Wt::Dbo::Session* getDboSession() { return &_session; } - Db& getDb() { return _db; } + Wt::Dbo::Session* getDboSession() + { + return &_session; + } + Db& getDb() + { + return _db; + } template typename Object::pointer create(Args&&... args) diff --git a/src/libs/services/scanner/CMakeLists.txt b/src/libs/services/scanner/CMakeLists.txt index 62d6f0f2..26976852 100644 --- a/src/libs/services/scanner/CMakeLists.txt +++ b/src/libs/services/scanner/CMakeLists.txt @@ -9,7 +9,6 @@ add_library(lmsscanner STATIC impl/scanners/PlayListFileScanner.cpp impl/scanners/Utils.cpp impl/steps/ArtworkUtils.cpp - impl/steps/FileScanQueue.cpp impl/steps/ScanErrorLogger.cpp impl/steps/ScanStepArtistReconciliation.cpp impl/steps/ScanStepAssociateArtistImages.cpp diff --git a/src/libs/services/scanner/impl/ScannerService.cpp b/src/libs/services/scanner/impl/ScannerService.cpp index f5bc18b8..db70e71c 100644 --- a/src/libs/services/scanner/impl/ScannerService.cpp +++ b/src/libs/services/scanner/impl/ScannerService.cpp @@ -24,6 +24,7 @@ #include #include "core/IConfig.hpp" +#include "core/IJobScheduler.hpp" #include "core/ILogger.hpp" #include "core/ITraceLogger.hpp" #include "database/MediaLibrary.hpp" @@ -139,6 +140,17 @@ namespace lms::scanner scanSettings.modify()->setSkipSingleReleasePlayLists(settings.skipSingleReleasePlayLists); // TODO add more fields } + + std::size_t getScannerThreadCount() + { + std::size_t threadCount{ core::Service::get()->getULong("scanner-thread-count", 0) }; + + if (threadCount == 0) + threadCount = std::max(std::thread::hardware_concurrency() / 2, 1); + + return threadCount; + } + } // namespace std::unique_ptr createScannerService(Db& db) @@ -148,9 +160,13 @@ namespace lms::scanner ScannerService::ScannerService(Db& db) : _db{ db } + , _jobScheduler{ core::createJobScheduler("Scanner", getScannerThreadCount()) } { _ioService.setThreadCount(1); + LMS_LOG(DBUPDATER, INFO, "Using " << _jobScheduler->getThreadCount() << " thread(s) for jobs"); + _jobScheduler->setShouldAbortCallback([this]() { return _abortScan; }); + refreshScanSettings(); start(); @@ -447,6 +463,7 @@ namespace lms::scanner std::transform(std::cbegin(_fileScanners), std::cend(_fileScanners), std::back_inserter(fileScanners), [](const std::unique_ptr& scanner) { return scanner.get(); }); ScanStepBase::InitParams params{ + .jobScheduler = *_jobScheduler, .settings = _settings, .lastScanSettings = _lastScanSettings.has_value() ? &(_lastScanSettings.value()) : nullptr, .progressCallback = cbFunc, diff --git a/src/libs/services/scanner/impl/ScannerService.hpp b/src/libs/services/scanner/impl/ScannerService.hpp index fd615dc5..d3598a36 100644 --- a/src/libs/services/scanner/impl/ScannerService.hpp +++ b/src/libs/services/scanner/impl/ScannerService.hpp @@ -20,6 +20,7 @@ #pragma once #include +#include #include #include #include @@ -36,6 +37,11 @@ #include "services/scanner/IScannerService.hpp" #include "steps/IScanStep.hpp" +namespace lms::core +{ + class IJobScheduler; +} + namespace lms::scanner { class IFileScanner; @@ -79,6 +85,9 @@ namespace lms::scanner void notifyInProgressIfNeeded(const ScanStepStats& stats); void notifyInProgress(const ScanStepStats& stats); + db::Db& _db; + std::unique_ptr _jobScheduler; + std::vector> _fileScanners; std::vector> _scanSteps; @@ -88,7 +97,6 @@ namespace lms::scanner boost::asio::system_timer _scheduleTimer{ _ioService }; Events _events; std::chrono::system_clock::time_point _lastScanInProgressEmit; - db::Db& _db; mutable std::shared_mutex _statusMutex; State _curState{ State::NotScheduled }; diff --git a/src/libs/services/scanner/impl/steps/FileScanQueue.cpp b/src/libs/services/scanner/impl/steps/FileScanQueue.cpp deleted file mode 100644 index 9b61a808..00000000 --- a/src/libs/services/scanner/impl/steps/FileScanQueue.cpp +++ /dev/null @@ -1,111 +0,0 @@ -/* - * Copyright (C) 2024 Emeric Poupon - * - * This file is part of LMS. - * - * LMS is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * LMS is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with LMS. If not, see . - */ - -#include "FileScanQueue.hpp" - -#include - -#include "core/ILogger.hpp" -#include "core/ITraceLogger.hpp" - -#include "scanners/IFileScanOperation.hpp" - -namespace lms::scanner -{ - FileScanQueue::FileScanQueue(std::size_t threadCount, bool& abort) - : _scanContextRunner{ _scanIoContext, threadCount, "FileScan" } - , _abort{ abort } - { - } - - FileScanQueue::~FileScanQueue() = default; - - void FileScanQueue::pushScanRequest(std::unique_ptr operation) - { - { - std::scoped_lock lock{ _mutex }; - _ongoingScanCount += 1; - } - - auto operationHandler{ [operation = std::move(operation), this]() mutable { - if (_abort) - { - std::scoped_lock lock{ _mutex }; - _ongoingScanCount -= 1; - } - else - { - { - LMS_SCOPED_TRACE_OVERVIEW("Scanner", operation->getName()); - LMS_LOG(DBUPDATER, DEBUG, operation->getName() << ": scanning file " << operation->getFilePath()); - operation->scan(); - } - - { - std::scoped_lock lock{ _mutex }; - - _scanResults.emplace_back(std::move(operation)); - _ongoingScanCount -= 1; - } - } - - _condVar.notify_all(); - } }; - - boost::asio::post(_scanIoContext, std::move(operationHandler)); - } - - std::size_t FileScanQueue::getResultsCount() const - { - std::scoped_lock lock{ _mutex }; - return _scanResults.size(); - } - - size_t FileScanQueue::popResults(std::vector>& results, std::size_t maxCount) - { - results.clear(); - results.reserve(maxCount); - - { - std::scoped_lock lock{ _mutex }; - - while (results.size() < maxCount && !_scanResults.empty()) - { - results.push_back(std::move(_scanResults.front())); - _scanResults.pop_front(); - } - } - - return results.size(); - } - - void FileScanQueue::wait(std::size_t maxScanRequestCount) - { - if (_ongoingScanCount <= maxScanRequestCount) - return; - - { - LMS_SCOPED_TRACE_OVERVIEW("Scanner", "WaitParseResults"); - - std::unique_lock lock{ _mutex }; - _condVar.wait(lock, [=, this] { return _ongoingScanCount <= maxScanRequestCount; }); - } - } - -} // namespace lms::scanner diff --git a/src/libs/services/scanner/impl/steps/FileScanQueue.hpp b/src/libs/services/scanner/impl/steps/FileScanQueue.hpp deleted file mode 100644 index 0cf4259e..00000000 --- a/src/libs/services/scanner/impl/steps/FileScanQueue.hpp +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright (C) 2024 Emeric Poupon - * - * This file is part of LMS. - * - * LMS is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * LMS is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with LMS. If not, see . - */ - -#pragma once - -#include -#include -#include -#include -#include -#include - -#include "core/IOContextRunner.hpp" - -namespace lms::scanner -{ - class IFileScanOperation; - - class FileScanQueue - { - public: - FileScanQueue(std::size_t threadCount, bool& abort); - ~FileScanQueue(); - FileScanQueue(const FileScanQueue&) = delete; - FileScanQueue& operator=(const FileScanQueue&) = delete; - - std::size_t getThreadCount() const { return _scanContextRunner.getThreadCount(); } - - void pushScanRequest(std::unique_ptr operation); - - std::size_t getResultsCount() const; - size_t popResults(std::vector>& results, std::size_t maxCount); - - void wait(std::size_t maxScanRequestCount = 0); // wait until ongoing scan request count <= maxScanRequestCount - - private: - boost::asio::io_context _scanIoContext; - core::IOContextRunner _scanContextRunner; - - mutable std::mutex _mutex; - std::atomic _ongoingScanCount{}; - std::deque> _scanResults; - std::condition_variable _condVar; - bool& _abort; - }; -} // namespace lms::scanner \ No newline at end of file diff --git a/src/libs/services/scanner/impl/steps/ScanStepBase.cpp b/src/libs/services/scanner/impl/steps/ScanStepBase.cpp index a82d7fb7..bde21dde 100644 --- a/src/libs/services/scanner/impl/steps/ScanStepBase.cpp +++ b/src/libs/services/scanner/impl/steps/ScanStepBase.cpp @@ -31,6 +31,7 @@ namespace lms::scanner , _progressCallback{ initParams.progressCallback } , _abortScan{ initParams.abortScan } , _db{ initParams.db } + , _jobScheduler{ initParams.jobScheduler } , _fileScanners(std::cbegin(initParams.fileScanners), std::cend(initParams.fileScanners)) , _lastScanSettings{ initParams.lastScanSettings } { diff --git a/src/libs/services/scanner/impl/steps/ScanStepBase.hpp b/src/libs/services/scanner/impl/steps/ScanStepBase.hpp index a467ddab..2677726d 100644 --- a/src/libs/services/scanner/impl/steps/ScanStepBase.hpp +++ b/src/libs/services/scanner/impl/steps/ScanStepBase.hpp @@ -27,6 +27,11 @@ #include "IScanStep.hpp" #include "ScanErrorLogger.hpp" +namespace lms::core +{ + class IJobScheduler; +} + namespace lms::db { class Db; @@ -46,6 +51,7 @@ namespace lms::scanner struct InitParams { + core::IJobScheduler& jobScheduler; const ScannerSettings& settings; const ScannerSettings* lastScanSettings{}; ProgressCallback progressCallback; @@ -59,6 +65,7 @@ namespace lms::scanner ScanStepBase& operator=(const ScanStepBase&) = delete; protected: + core::IJobScheduler& getJobScheduler() { return _jobScheduler; }; const ScannerSettings* getLastScanSettings() const { return _lastScanSettings; } IFileScanner* selectFileScanner(const std::filesystem::path& filePath) const; void visitFileScanners(const std::function& visitor) const; @@ -78,6 +85,7 @@ namespace lms::scanner db::Db& _db; private: + core::IJobScheduler& _jobScheduler; std::unordered_map _scannerByFile; std::unordered_map _scannerByExtension; std::vector _fileScanners; diff --git a/src/libs/services/scanner/impl/steps/ScanStepScanFiles.cpp b/src/libs/services/scanner/impl/steps/ScanStepScanFiles.cpp index 1ed23520..5ed9873e 100644 --- a/src/libs/services/scanner/impl/steps/ScanStepScanFiles.cpp +++ b/src/libs/services/scanner/impl/steps/ScanStepScanFiles.cpp @@ -19,9 +19,9 @@ #include "ScanStepScanFiles.hpp" -#include "FileScanQueue.hpp" #include "ScannerSettings.hpp" -#include "core/IConfig.hpp" +#include "core/IJob.hpp" +#include "core/IJobScheduler.hpp" #include "core/ILogger.hpp" #include "core/ITraceLogger.hpp" #include "core/Path.hpp" @@ -39,15 +39,32 @@ namespace lms::scanner namespace { - std::size_t getScanMetaDataThreadCount() + class FileScanJob : public core::IJob { - std::size_t threadCount{ core::Service::get()->getULong("scanner-metadata-thread-count", 0) }; + public: + FileScanJob(std::unique_ptr scanOperation) + : _scanOperation{ std::move(scanOperation) } + { + } - if (threadCount == 0) - threadCount = std::max(std::thread::hardware_concurrency() / 2, 1); + IFileScanOperation& getScanOperation() + { + return *_scanOperation; + } - return threadCount; - } + private: + core::LiteralString getName() const override + { + return _scanOperation->getName(); + } + + void run() override + { + _scanOperation->scan(); + } + + std::unique_ptr _scanOperation; + }; FileToScan retrieveFileInfo(const std::filesystem::path& file, const MediaLibraryInfo& mediaLibrary, std::error_code& ec) { @@ -71,7 +88,6 @@ namespace lms::scanner ScanStepScanFiles::ScanStepScanFiles(InitParams& initParams) : ScanStepBase{ initParams } - , _fileScanQueue{ getScanMetaDataThreadCount(), _abortScan } { visitFileScanners([](IFileScanner* scanner) { for (const std::filesystem::path& file : scanner->getSupportedFiles()) @@ -80,8 +96,6 @@ namespace lms::scanner for (const std::filesystem::path& extension : scanner->getSupportedExtensions()) LMS_LOG(DBUPDATER, INFO, scanner->getName() << ": supporting file extension " << extension); }); - - LMS_LOG(DBUPDATER, INFO, "Using " << _fileScanQueue.getThreadCount() << " thread(s) for scanning file metadata"); } bool ScanStepScanFiles::needProcess([[maybe_unused]] const ScanContext& context) const @@ -100,9 +114,13 @@ namespace lms::scanner void ScanStepScanFiles::process(ScanContext& context, const MediaLibraryInfo& mediaLibrary) { - const std::size_t scanQueueMaxScanRequestCount{ 100 * _fileScanQueue.getThreadCount() }; - const std::size_t processFileResultsBatchSize{ 10 }; + core::IJobScheduler& jobScheduler{ getJobScheduler() }; + assert(jobScheduler.getJobsDoneCount() == 0); + const std::size_t scanQueueMaxScanRequestCount{ 100 * jobScheduler.getThreadCount() }; + constexpr std::size_t processFileResultsBatchSize{ 10 }; + + std::vector> jobsDone; std::vector> scanOperations; core::pathUtils::exploreFilesRecursive( @@ -130,7 +148,7 @@ namespace lms::scanner if (context.scanOptions.fullScan || scanner->needsScan(fileToScan)) { auto scanOperation{ scanner->createScanOperation(std::move(fileToScan)) }; - _fileScanQueue.pushScanRequest(std::move(scanOperation)); + jobScheduler.scheduleJob(std::make_unique(std::move(scanOperation))); } } @@ -138,38 +156,48 @@ namespace lms::scanner _progressCallback(context.currentStepStats); } - while (_fileScanQueue.getResultsCount() > (scanQueueMaxScanRequestCount / 2)) + while (jobScheduler.getJobsDoneCount() > (scanQueueMaxScanRequestCount / 2)) { - _fileScanQueue.popResults(scanOperations, processFileResultsBatchSize); - processFileScanResults(context, scanOperations); + jobScheduler.popJobsDone(jobsDone, processFileResultsBatchSize); + processFileScanResults(context, jobsDone); + jobsDone.clear(); } - _fileScanQueue.wait(scanQueueMaxScanRequestCount); + jobScheduler.waitUntilJobCountAtMost(scanQueueMaxScanRequestCount); return true; }, &excludeDirFileName); - _fileScanQueue.wait(); + jobScheduler.wait(); - while (!_abortScan && _fileScanQueue.popResults(scanOperations, processFileResultsBatchSize) > 0) - processFileScanResults(context, scanOperations); + while (jobScheduler.popJobsDone(jobsDone, processFileResultsBatchSize) > 0) + { + if (!_abortScan) + processFileScanResults(context, jobsDone); + + jobsDone.clear(); + } + + assert(jobScheduler.getJobsDoneCount() == 0); } - void ScanStepScanFiles::processFileScanResults(ScanContext& context, std::span> scanOperations) + void ScanStepScanFiles::processFileScanResults(ScanContext& context, std::span> scanJobs) { LMS_SCOPED_TRACE_OVERVIEW("Scanner", "ProcessScanResults"); db::Session& dbSession{ _db.getTLSSession() }; auto transaction{ dbSession.createWriteTransaction() }; - for (auto& scanOperation : scanOperations) + for (auto& scanJob : scanJobs) { if (_abortScan) return; - LMS_LOG(DBUPDATER, DEBUG, scanOperation->getName() << ": processing result for " << scanOperation->getFilePath()); - const IFileScanOperation::OperationResult res{ scanOperation->processResult() }; + IFileScanOperation& scanOperation{ static_cast(*scanJob).getScanOperation() }; + + LMS_LOG(DBUPDATER, DEBUG, scanOperation.getName() << ": processing result for " << scanOperation.getFilePath()); + const IFileScanOperation::OperationResult res{ scanOperation.processResult() }; switch (res) { case IFileScanOperation::OperationResult::Added: @@ -187,7 +215,7 @@ namespace lms::scanner } context.stats.scans++; - for (const auto& error : scanOperation->getErrors()) + for (const auto& error : scanOperation.getErrors()) addError(context, error); } } diff --git a/src/libs/services/scanner/impl/steps/ScanStepScanFiles.hpp b/src/libs/services/scanner/impl/steps/ScanStepScanFiles.hpp index 90e9ce87..1dfc315b 100644 --- a/src/libs/services/scanner/impl/steps/ScanStepScanFiles.hpp +++ b/src/libs/services/scanner/impl/steps/ScanStepScanFiles.hpp @@ -19,12 +19,15 @@ #pragma once -#include #include -#include "FileScanQueue.hpp" #include "ScanStepBase.hpp" +namespace lms::core +{ + class IJob; +} + namespace lms::scanner { class IFileScanner; @@ -42,8 +45,6 @@ namespace lms::scanner void process(ScanContext& context) override; void process(ScanContext& context, const MediaLibraryInfo& mediaLibrary); - void processFileScanResults(ScanContext& context, std::span> scanOperations); - - FileScanQueue _fileScanQueue; + void processFileScanResults(ScanContext& context, std::span> scanJobs); }; } // namespace lms::scanner