Optimized the differential scan step by using threads to check files

2025-07-19 17:10:15 +02:00
parent 97866d6143
commit 37c75f8677
7 changed files with 162 additions and 87 deletions
@@ -23,6 +23,7 @@

 #include "core/ILogger.hpp"
 #include "core/ITraceLogger.hpp"
+
 #include "database/objects/Artist.hpp"
 #include "database/objects/ArtistInfo.hpp"
 #include "database/objects/Artwork.hpp"
@@ -367,17 +368,17 @@ namespace lms::db
        });
    }

-    std::size_t Session::getTotalFilesCount()
+    FileStats Session::getFileStats()
    {
-        std::size_t res{};
+        FileStats stats{};

-        res += db::Track::getCount(*this);
-        res += db::Image::getCount(*this);
-        res += db::TrackLyrics::getExternalLyricsCount(*this);
-        res += db::PlayListFile::getCount(*this);
-        res += db::ArtistInfo::getCount(*this);
+        stats.trackCount = db::Track::getCount(*this);
+        stats.artistInfoCount = db::ArtistInfo::getCount(*this);
+        stats.imageCount = db::Image::getCount(*this);
+        stats.trackLyricsCount = db::TrackLyrics::getExternalLyricsCount(*this);
+        stats.playListCount = db::PlayListFile::getCount(*this);

-        return res;
+        return stats;
    }

    void Session::retrieveEntriesToAnalyze(std::vector<std::string>& entryList)
@@ -27,6 +27,7 @@
 #include <vector>

 #include "database/Transaction.hpp"
+#include "database/Types.hpp"

 namespace lms::db
 {
@@ -52,8 +53,8 @@ namespace lms::db
        void retrieveEntriesToAnalyze(std::vector<std::string>& entryList);
        void analyzeEntry(const std::string& entry);

-        bool areAllTablesEmpty();         // need to acquire a read transaction
-        std::size_t getTotalFilesCount(); // need to acquire a read transaction
+        bool areAllTablesEmpty(); // need to acquire a read transaction
+        FileStats getFileStats(); // need to acquire a read transaction

        void prepareTablesIfNeeded(); // need to run only once at startup
        bool migrateSchemaIfNeeded(); // returns true if migration was performed
@@ -102,6 +102,17 @@ namespace lms::db
        }
    };

+    struct FileStats
+    {
+        std::size_t trackCount;
+        std::size_t imageCount;
+        std::size_t trackLyricsCount;
+        std::size_t playListCount;
+        std::size_t artistInfoCount;
+
+        std::size_t getTotalFileCount() const { return trackCount + imageCount + trackLyricsCount + playListCount + artistInfoCount; }
+    };
+
    struct YearRange
    {
        int begin{};
@@ -166,17 +166,17 @@ namespace lms::scanner
        LMS_LOG(DBUPDATER, INFO, "Using " << _jobScheduler->getThreadCount() << " thread(s) for jobs");
        _jobScheduler->setShouldAbortCallback([this]() { return _abortScan; });

-        std::size_t totalFilesCount{};
+        std::size_t totalFileCount{};
        {
            auto& session{ _db.getTLSSession() };
            auto transaction{ session.createReadTransaction() };
-            totalFilesCount = session.getTotalFilesCount();
+            totalFileCount = session.getFileStats().getTotalFileCount();
        }

        // Force optimize in case scanner aborted during a large import, but do this only if there are enough elements in the database
        // Otherwise, indexes may be not used and queries may be slower and slower while adding more and more elements in the db
-        LMS_LOG(DBUPDATER, INFO, "Scanned file count = " << totalFilesCount);
-        if (totalFilesCount >= 1'000)
+        LMS_LOG(DBUPDATER, INFO, "Scanned file count = " << totalFileCount);
+        if (totalFileCount >= 1'000)
            _db.getTLSSession().fullAnalyze();

        refreshTracingLoggerStats();
@@ -194,7 +194,7 @@ namespace lms::scanner

        {
            auto transaction{ session.createReadTransaction() };
-            context.currentStepStats.totalElems = session.getTotalFilesCount();
+            context.currentStepStats.totalElems = session.getFileStats().getTotalFileCount();
        }
        LMS_LOG(DBUPDATER, DEBUG, context.currentStepStats.totalElems << " files to be checked...");

@@ -19,12 +19,13 @@

 #include "ScanStepScanFiles.hpp"

+#include <deque>
+
 #include "ScannerSettings.hpp"
 #include "core/IJob.hpp"
 #include "core/IJobScheduler.hpp"
 #include "core/ILogger.hpp"
 #include "core/ITraceLogger.hpp"
-#include "core/Path.hpp"
 #include "database/IDb.hpp"
 #include "database/Session.hpp"
 #include "scanners/FileToScan.hpp"
@@ -93,28 +94,60 @@ namespace lms::scanner
        class FileScanJob : public core::IJob
        {
        public:
-            FileScanJob(std::unique_ptr<IFileScanOperation> scanOperation)
-                : _scanOperation{ std::move(scanOperation) }
+            FileScanJob(const FileScanners& fileScanners, const MediaLibraryInfo& mediaLibrary, bool fullScan, std::span<const std::filesystem::directory_entry> files)
+                : _fileScanners{ fileScanners }
+                , _mediaLibrary{ mediaLibrary }
+                , _fullScan{ fullScan }
+                , _files{ std::cbegin(files), std::cend(files) }
            {
            }

-            IFileScanOperation& getScanOperation()
+            std::size_t getFileCount() const { return _files.size(); };
+
+            std::span<std::unique_ptr<IFileScanOperation>> getScanOperations()
            {
-                return *_scanOperation;
+                return _scanOperations;
            }

        private:
            core::LiteralString getName() const override
            {
-                return _scanOperation->getName();
+                return "Scan Files";
            }

            void run() override
            {
-                _scanOperation->scan();
+                for (const auto& file : _files)
+                {
+                    IFileScanner* scanner{ _fileScanners.select(file.path()) };
+                    if (!scanner)
+                        continue;
+
+                    FileToScan fileToScan;
+
+                    fileToScan.filePath = file.path();
+                    fileToScan.mediaLibrary = _mediaLibrary;
+                    fileToScan.lastWriteTime.setTime_t(Wt::WDateTime{ std::chrono::file_clock::to_sys(file.last_write_time()) }.toTime_t()); // sec resolution, as stored in the database
+                    fileToScan.fileSize = file.file_size();
+
+                    if (_fullScan || scanner->needsScan(fileToScan))
+                    {
+                        auto scanOperation{ scanner->createScanOperation(std::move(fileToScan)) };
+
+                        {
+                            LMS_SCOPED_TRACE_DETAILED("Scanner", scanOperation->getName());
+                            scanOperation->scan();
+                        }
+                        _scanOperations.push_back(std::move(scanOperation));
+                    }
+                }
            }

-            std::unique_ptr<IFileScanOperation> _scanOperation;
+            const FileScanners& _fileScanners;
+            const MediaLibraryInfo& _mediaLibrary;
+            const bool _fullScan;
+            std::vector<std::filesystem::directory_entry> _files;
+            std::vector<std::unique_ptr<IFileScanOperation>> _scanOperations;
        };
    } // namespace

@@ -128,96 +161,124 @@ namespace lms::scanner
    {
        for (const MediaLibraryInfo& mediaLibrary : _settings.mediaLibraries)
            process(context, mediaLibrary);
+
+        context.stats.totalFileCount = context.currentStepStats.processedElems;
    }

    void ScanStepScanFiles::process(ScanContext& context, const MediaLibraryInfo& mediaLibrary)
    {
-        const std::size_t scanQueueMaxScanRequestCount{ 50 * getJobScheduler().getThreadCount() };
-        constexpr std::size_t processFileResultsBatchSize{ 10 };
+        constexpr std::size_t filesPerScanJob{ 10 };
+        constexpr std::size_t scanQueueMaxSize{ 50 };
+        constexpr std::size_t processFileResultsBatchSize{ 1 };
        constexpr float drainRatio{ 0.85 };

+        std::deque<std::unique_ptr<IFileScanOperation>> operations;
+
        auto processDoneJobs = [&](std::span<std::unique_ptr<core::IJob>> jobsDone) {
+            for (const auto& jobDone : jobsDone)
+            {
+                auto& fileScanJob{ static_cast<FileScanJob&>(*jobDone) };
+                for (std::unique_ptr<IFileScanOperation>& scanOperation : fileScanJob.getScanOperations())
+                    operations.push_back(std::move(scanOperation));
+
+                context.currentStepStats.processedElems += fileScanJob.getFileCount();
+            }
+
            if (!_abortScan)
-                processFileScanResults(context, jobsDone);
+                processFileScanOperations(context, operations, true /* force batch */);
+
+            _progressCallback(context.currentStepStats);
        };

-        JobQueue queue{ getJobScheduler(), scanQueueMaxScanRequestCount, processDoneJobs, processFileResultsBatchSize, drainRatio };
+        {
+            JobQueue queue{ getJobScheduler(), scanQueueMaxSize, processDoneJobs, processFileResultsBatchSize, drainRatio };

-        std::vector<std::unique_ptr<core::IJob>> jobsDone;
-        std::vector<std::unique_ptr<IFileScanOperation>> scanOperations;
+            std::vector<std::filesystem::directory_entry> filesToScan;

-        exploreFilesRecursive(
-            mediaLibrary.rootDirectory, [&](std::error_code ec, const std::filesystem::path& path, const std::filesystem::directory_entry* fileEntry) {
-                LMS_SCOPED_TRACE_DETAILED("Scanner", "OnExploreFile");
+            exploreFilesRecursive(
+                mediaLibrary.rootDirectory, [&](std::error_code ec, const std::filesystem::path& path, const std::filesystem::directory_entry* fileEntry) {
+                    LMS_SCOPED_TRACE_DETAILED("Scanner", "OnExploreFile");

-                assert((ec && !fileEntry) || (!ec && fileEntry));
+                    assert((ec && !fileEntry) || (!ec && fileEntry));

-                if (_abortScan)
-                    return false; // stop iterating
+                    if (_abortScan)
+                        return false; // stop iterating

-                if (ec)
-                {
-                    addError<IOScanError>(context, path, ec);
-                    context.stats.skips++;
-                }
-                else if (IFileScanner * scanner{ getFileScanners().select(path) })
-                {
-                    FileToScan fileToScan;
-
-                    fileToScan.filePath = path;
-                    fileToScan.mediaLibrary = mediaLibrary;
-                    fileToScan.lastWriteTime.setTime_t(Wt::WDateTime{ std::chrono::file_clock::to_sys(fileEntry->last_write_time()) }.toTime_t()); // sec resolution, as stored in the database
-                    fileToScan.fileSize = fileEntry->file_size();
-
-                    if (context.scanOptions.fullScan || scanner->needsScan(fileToScan))
+                    if (ec)
                    {
-                        auto scanOperation{ scanner->createScanOperation(std::move(fileToScan)) };
-                        queue.push(std::make_unique<FileScanJob>(std::move(scanOperation)));
+                        addError<IOScanError>(context, path, ec);
+                        context.stats.skips++;
+                    }
+                    else
+                    {
+                        filesToScan.push_back(*fileEntry);
+
+                        if (filesToScan.size() >= filesPerScanJob)
+                        {
+                            queue.push(std::make_unique<FileScanJob>(getFileScanners(), mediaLibrary, context.scanOptions.fullScan, filesToScan));
+                            filesToScan.clear();
+                        }
                    }

-                    context.currentStepStats.processedElems++;
-                    _progressCallback(context.currentStepStats);
-                }
+                    return true;
+                },
+                &excludeDirFileName);

-                return true;
-            },
-            &excludeDirFileName);
+            if (!filesToScan.empty())
+                queue.push(std::make_unique<FileScanJob>(getFileScanners(), mediaLibrary, context.scanOptions.fullScan, filesToScan));

-        context.stats.totalFileCount = context.currentStepStats.totalElems;
+            _progressCallback(context.currentStepStats);
+        }
+
+        // Process remaining objects
+        processFileScanOperations(context, operations, false /* force batch */);
    }

-    void ScanStepScanFiles::processFileScanResults(ScanContext& context, std::span<std::unique_ptr<core::IJob>> scanJobs)
+    std::size_t ScanStepScanFiles::processFileScanOperations(ScanContext& context, std::deque<std::unique_ptr<IFileScanOperation>>& scanOperations, bool forceBatch)
    {
+        std::size_t count{};
+        constexpr std::size_t writeBatchSize{ 10 };
+
        LMS_SCOPED_TRACE_OVERVIEW("Scanner", "ProcessScanResults");

-        db::Session& dbSession{ _db.getTLSSession() };
-        auto transaction{ dbSession.createWriteTransaction() };
-
-        for (auto& scanJob : scanJobs)
+        while ((forceBatch && scanOperations.size() >= writeBatchSize) || !scanOperations.empty())
        {
-            IFileScanOperation& scanOperation{ static_cast<FileScanJob&>(*scanJob).getScanOperation() };
+            db::Session& dbSession{ _db.getTLSSession() };
+            auto transaction{ dbSession.createWriteTransaction() };

-            LMS_LOG(DBUPDATER, DEBUG, scanOperation.getName() << ": processing result for " << scanOperation.getFilePath());
-            const IFileScanOperation::OperationResult res{ scanOperation.processResult() };
-            switch (res)
+            for (std::size_t i{}; !scanOperations.empty() && i < writeBatchSize; ++i)
            {
-            case IFileScanOperation::OperationResult::Added:
-                context.stats.additions++;
-                break;
-            case IFileScanOperation::OperationResult::Removed:
-                context.stats.deletions++;
-                break;
-            case IFileScanOperation::OperationResult::Skipped:
-                context.stats.failures++;
-                break;
-            case IFileScanOperation::OperationResult::Updated:
-                context.stats.updates++;
-                break;
+                processFileScanOperation(context, *scanOperations.front());
+                scanOperations.pop_front();
+                count++;
            }
-            context.stats.scans++;
-
-            for (const auto& error : scanOperation.getErrors())
-                addError(context, error);
        }
+
+        return count;
+    }
+
+    void ScanStepScanFiles::processFileScanOperation(ScanContext& context, IFileScanOperation& scanOperation)
+    {
+        LMS_LOG(DBUPDATER, DEBUG, scanOperation.getName() << ": processing result for " << scanOperation.getFilePath());
+        const IFileScanOperation::OperationResult res{ scanOperation.processResult() };
+        switch (res)
+        {
+        case IFileScanOperation::OperationResult::Added:
+            context.stats.additions++;
+            break;
+        case IFileScanOperation::OperationResult::Removed:
+            context.stats.deletions++;
+            break;
+        case IFileScanOperation::OperationResult::Skipped:
+            context.stats.failures++;
+            break;
+        case IFileScanOperation::OperationResult::Updated:
+            context.stats.updates++;
+            break;
+        }
+        context.stats.scans++;
+
+        for (const auto& error : scanOperation.getErrors())
+            addError(context, error);
    }
 } // namespace lms::scanner
@@ -19,7 +19,7 @@

 #pragma once

-#include <span>
+#include <deque>

 #include "ScanStepBase.hpp"

@@ -30,7 +30,7 @@ namespace lms::core

 namespace lms::scanner
 {
-    class IFileScanner;
+    class IFileScanOperation;
    struct MediaLibraryInfo;

    class ScanStepScanFiles : public ScanStepBase
@@ -45,6 +45,7 @@ namespace lms::scanner
        void process(ScanContext& context) override;

        void process(ScanContext& context, const MediaLibraryInfo& mediaLibrary);
-        void processFileScanResults(ScanContext& context, std::span<std::unique_ptr<core::IJob>> scanJobs);
+        std::size_t processFileScanOperations(ScanContext& context, std::deque<std::unique_ptr<IFileScanOperation>>& scanOperations, bool forceBatch);
+        void processFileScanOperation(ScanContext& context, IFileScanOperation& operation);
    };
 } // namespace lms::scanner