owen/lms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Significantly improved scanner performance (now have a dedicated pool of threads for scanning file metadata)

Browse Source
This commit is contained in:
emeric
2024-02-10 18:39:55 +01:00
parent de4a0cf583
commit 6b1fd9c841
5 changed files with 261 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conf/lms.conf
+4 -1
View File
@@ -33,7 +33,7 @@ approot = "/usr/share/lms/approot";
# Location for deployment (See README if you want to deploy on a non root path) # Location for deployment (See README if you want to deploy on a non root path)
deploy-path = "/"; deploy-path = "/";
# Number of threads to be used to dispatch http requests (0 means auto detect) # Number of threads to be used to dispatch http requests (0 means number of logical CPUs)
http-server-thread-count = 0; http-server-thread-count = 0;
# ListenBrainz root API # ListenBrainz root API
@@ -98,3 +98,6 @@ scanner-skip-duplicate-mbid = false;
# Scanner read style for metadata, maybe be 'fast', 'average' or 'accurate' # Scanner read style for metadata, maybe be 'fast', 'average' or 'accurate'
scanner-parser-read-style = "average"; 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;
src/libs/services/scanner/impl/ScanStepScanFiles.cpp
+173 -57
View File
@@ -215,22 +215,94 @@ namespace Scanner
throw LmsException{ "Invalid value for 'scanner-parser-read-style'" }; throw LmsException{ "Invalid value for 'scanner-parser-read-style'" };
} }
std::size_t getScanMetaDataThreadCount()
{
std::size_t threadCount{ Service<IConfig>::get()->getULong("scanner-metadata-thread-count", 0) };
if (threadCount == 0)
threadCount = std::max<std::size_t>(std::thread::hardware_concurrency() / 2, 1);
return threadCount;
}
} // namespace } // namespace
ScanStepScanFiles::MetadataScanQueue::MetadataScanQueue(MetaData::IParser& parser, std::size_t threadCount)
: _metadataParser{ parser }
, _scanContextRunner{ _scanContext, threadCount }
{}
void ScanStepScanFiles::MetadataScanQueue::pushScanRequest(const std::filesystem::path path)
{
{
std::scoped_lock lock{ _mutex };
_ongoingScanCount += 1;
}
_scanContext.post([=, this]
{
std::optional<MetaData::Track> trackMetaData{ _metadataParser.parse(path) };
{
std::scoped_lock lock{ _mutex };
_scanResults.emplace_back(std::make_unique<MetaDataScanResult>(path, std::move(trackMetaData)));
_ongoingScanCount -= 1;
}
_condVar.notify_all();
});
}
std::size_t ScanStepScanFiles::MetadataScanQueue::getResultsCount() const
{
std::scoped_lock lock{ _mutex };
return _scanResults.size();
}
size_t ScanStepScanFiles::MetadataScanQueue::popResults(std::vector<std::unique_ptr<MetaDataScanResult>>& 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 ScanStepScanFiles::MetadataScanQueue::wait(std::size_t maxScanRequestCount)
{
std::unique_lock lock{ _mutex };
_condVar.wait(lock, [=, this] {return _ongoingScanCount <= maxScanRequestCount;});
}
ScanStepScanFiles::ScanStepScanFiles(InitParams& initParams) ScanStepScanFiles::ScanStepScanFiles(InitParams& initParams)
: ScanStepBase{ initParams } : ScanStepBase{ initParams }
, _metadataParser{ MetaData::createParser(MetaData::ParserType::TagLib, getParserReadStyle()) } // For now, always use TagLib , _metadataParser{ MetaData::createParser(MetaData::ParserType::TagLib, getParserReadStyle()) } // For now, always use TagLib
, _metadataScanQueue{ *_metadataParser, getScanMetaDataThreadCount() }
{ {
LMS_LOG(DBUPDATER, INFO, "Using " << _metadataScanQueue.getThreadCount() << " thread(s) for scanning file metadata");
} }
void ScanStepScanFiles::process(ScanContext& context) void ScanStepScanFiles::process(ScanContext& context)
{ {
const std::size_t scanQueueMaxScanRequestCount{ 20 * _metadataScanQueue.getThreadCount() };
const std::size_t processMetaDataBatchSize{ 10 };
{ {
std::vector<std::string> tagsToParse{ _extraTagsToParse }; std::vector<std::string> tagsToParse{ _extraTagsToParse };
tagsToParse.insert(std::end(tagsToParse), std::cbegin(_settings.extraTags), std::cend(_settings.extraTags)); tagsToParse.insert(std::end(tagsToParse), std::cbegin(_settings.extraTags), std::cend(_settings.extraTags));
_metadataParser->setUserExtraTags(tagsToParse); _metadataParser->setUserExtraTags(tagsToParse);
} }
std::vector<std::unique_ptr<MetaDataScanResult>> scanResults;
context.currentStepStats.totalElems = context.stats.filesScanned; context.currentStepStats.totalElems = context.stats.filesScanned;
for (const ScannerSettings::MediaLibraryInfo& mediaLibrary : _settings.mediaLibraries) for (const ScannerSettings::MediaLibraryInfo& mediaLibrary : _settings.mediaLibraries)
@@ -247,22 +319,32 @@ namespace Scanner
} }
else if (PathUtils::hasFileAnyExtension(path, _settings.supportedExtensions)) else if (PathUtils::hasFileAnyExtension(path, _settings.supportedExtensions))
{ {
scanAudioFile(path, context, mediaLibrary); if (checkFileNeedScan(context, path, mediaLibrary))
_metadataScanQueue.pushScanRequest(path);
context.currentStepStats.processedElems++; context.currentStepStats.processedElems++;
_progressCallback(context.currentStepStats); _progressCallback(context.currentStepStats);
// optimize the database during scan (if we import a very large database, it may be too late to do it once at end)
if ((context.currentStepStats.processedElems % 1'000) == 0)
_db.getTLSSession().optimize();
} }
while (_metadataScanQueue.getResultsCount() > processMetaDataBatchSize)
{
_metadataScanQueue.popResults(scanResults, processMetaDataBatchSize);
processMetaDataScanResults(context, scanResults, mediaLibrary);
}
_metadataScanQueue.wait(scanQueueMaxScanRequestCount);
return true; return true;
}, &excludeDirFileName); }, &excludeDirFileName);
_metadataScanQueue.wait();
while (_metadataScanQueue.popResults(scanResults, processMetaDataBatchSize) > 0)
processMetaDataScanResults(context, scanResults, mediaLibrary);
} }
} }
void ScanStepScanFiles::scanAudioFile(const std::filesystem::path& file, ScanContext& context, const ScannerSettings::MediaLibraryInfo& libraryInfo) bool ScanStepScanFiles::checkFileNeedScan(ScanContext& context, const std::filesystem::path& file, const ScannerSettings::MediaLibraryInfo& libraryInfo)
{ {
ScanStats& stats{ context.stats }; ScanStats& stats{ context.stats };
Wt::WDateTime lastWriteTime; Wt::WDateTime lastWriteTime;
@@ -274,7 +356,7 @@ namespace Scanner
{ {
LMS_LOG(DBUPDATER, ERROR, e.what()); LMS_LOG(DBUPDATER, ERROR, e.what());
stats.skips++; stats.skips++;
return; return false;
} }
bool needUpdateLibrary{}; bool needUpdateLibrary{};
@@ -296,7 +378,7 @@ namespace Scanner
if (trackMediaLibrary && trackMediaLibrary->getId() == libraryInfo.id) if (trackMediaLibrary && trackMediaLibrary->getId() == libraryInfo.id)
{ {
stats.skips++; stats.skips++;
return; return false;
} }
needUpdateLibrary = true; needUpdateLibrary = true;
@@ -312,26 +394,60 @@ namespace Scanner
assert(track); assert(track);
track.modify()->setMediaLibrary(Database::MediaLibrary::find(dbSession, libraryInfo.id)); // may be null, will be handled in the next scan anyway track.modify()->setMediaLibrary(Database::MediaLibrary::find(dbSession, libraryInfo.id)); // may be null, will be handled in the next scan anyway
stats.updates++; stats.updates++;
return; return false;
} }
std::optional<MetaData::Track> trackInfo{ _metadataParser->parse(file) }; return true; // need to scan
if (!trackInfo) }
{
context.stats.errors.emplace_back(file, ScanErrorType::CannotParseFile);
return;
}
stats.scans++;
void ScanStepScanFiles::processMetaDataScanResults(ScanContext& context, std::span<const std::unique_ptr<MetaDataScanResult>> scanResults, const ScannerSettings::MediaLibraryInfo& libraryInfo)
{
Database::Session& dbSession{ _db.getTLSSession() }; Database::Session& dbSession{ _db.getTLSSession() };
auto transaction{ dbSession.createWriteTransaction() }; auto transaction{ dbSession.createWriteTransaction() };
for (const auto& scanResult : scanResults)
{
if (_abortScan)
return;
if (scanResult->trackMetaData)
{
context.stats.scans++;
processFileMetaData(context, scanResult->path, *scanResult->trackMetaData, libraryInfo);
// optimize the database during scan (if we import a very large database, it may be too late to do it once at end)
if ((context.stats.scans % 1'000) == 0)
_db.getTLSSession().optimize();
}
else
{
context.stats.errors.emplace_back(scanResult->path, ScanErrorType::CannotParseFile);
}
}
}
void ScanStepScanFiles::processFileMetaData(ScanContext& context, const std::filesystem::path& file, const MetaData::Track& trackMetadata, const ScannerSettings::MediaLibraryInfo& libraryInfo)
{
ScanStats& stats{ context.stats };
Wt::WDateTime lastWriteTime;
try
{
lastWriteTime = PathUtils::getLastWriteTime(file);
}
catch (LmsException& e)
{
LMS_LOG(DBUPDATER, ERROR, e.what());
stats.skips++;
return;
}
Database::Session& dbSession{ _db.getTLSSession() };
Track::pointer track{ Track::findByPath(dbSession, file) }; Track::pointer track{ Track::findByPath(dbSession, file) };
if (trackInfo->mbid && (!track || _settings.skipDuplicateMBID)) if (trackMetadata.mbid && (!track || _settings.skipDuplicateMBID))
{ {
std::vector<Track::pointer> duplicateTracks{ Track::findByMBID(dbSession, *trackInfo->mbid) }; std::vector<Track::pointer> duplicateTracks{ Track::findByMBID(dbSession, *trackMetadata.mbid) };
// find for existing MBIDs as the file may have just been moved // find for existing MBIDs as the file may have just been moved
if (!track && duplicateTracks.size() == 1) if (!track && duplicateTracks.size() == 1)
@@ -378,7 +494,7 @@ namespace Scanner
} }
// We estimate this is an audio file if the duration is not null // We estimate this is an audio file if the duration is not null
if (trackInfo->duration == std::chrono::milliseconds::zero()) if (trackMetadata.duration == std::chrono::milliseconds::zero())
{ {
LMS_LOG(DBUPDATER, DEBUG, "Skipped '" << file.string() << "' (duration is 0)"); LMS_LOG(DBUPDATER, DEBUG, "Skipped '" << file.string() << "' (duration is 0)");
@@ -394,8 +510,8 @@ namespace Scanner
// ***** Title // ***** Title
std::string title; std::string title;
if (!trackInfo->title.empty()) if (!trackMetadata.title.empty())
title = trackInfo->title; title = trackMetadata.title;
else else
{ {
// TODO parse file name guess track etc. // TODO parse file name guess track etc.
@@ -424,78 +540,78 @@ namespace Scanner
track.modify()->setMediaLibrary(MediaLibrary::find(dbSession, libraryInfo.id)); // may be null if settings are updated in // => next scan will correct this track.modify()->setMediaLibrary(MediaLibrary::find(dbSession, libraryInfo.id)); // may be null if settings are updated in // => next scan will correct this
track.modify()->clearArtistLinks(); track.modify()->clearArtistLinks();
// Do not fallback on artists with the same name but having a MBID for artist and releaseArtists, as it may be corrected by properly tagging files // Do not fallback on artists with the same name but having a MBID for artist and releaseArtists, as it may be corrected by properly tagging files
for (const Artist::pointer& artist : getOrCreateArtists(dbSession, trackInfo->artists, false)) for (const Artist::pointer& artist : getOrCreateArtists(dbSession, trackMetadata.artists, false))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, artist, TrackArtistLinkType::Artist)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, artist, TrackArtistLinkType::Artist));
if (trackInfo->medium && trackInfo->medium->release) if (trackMetadata.medium && trackMetadata.medium->release)
{ {
for (const Artist::pointer& releaseArtist : getOrCreateArtists(dbSession, trackInfo->medium->release->artists, false)) for (const Artist::pointer& releaseArtist : getOrCreateArtists(dbSession, trackMetadata.medium->release->artists, false))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, releaseArtist, TrackArtistLinkType::ReleaseArtist)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, releaseArtist, TrackArtistLinkType::ReleaseArtist));
} }
// Allow fallbacks on artists with the same name even if they have MBID, since there is no tag to indicate the MBID of these artists // Allow fallbacks on artists with the same name even if they have MBID, since there is no tag to indicate the MBID of these artists
// We could ask MusicBrainz to get all the information, but that would heavily slow down the import process // We could ask MusicBrainz to get all the information, but that would heavily slow down the import process
for (const Artist::pointer& conductor : getOrCreateArtists(dbSession, trackInfo->conductorArtists, true)) for (const Artist::pointer& conductor : getOrCreateArtists(dbSession, trackMetadata.conductorArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, conductor, TrackArtistLinkType::Conductor)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, conductor, TrackArtistLinkType::Conductor));
for (const Artist::pointer& composer : getOrCreateArtists(dbSession, trackInfo->composerArtists, true)) for (const Artist::pointer& composer : getOrCreateArtists(dbSession, trackMetadata.composerArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, composer, TrackArtistLinkType::Composer)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, composer, TrackArtistLinkType::Composer));
for (const Artist::pointer& lyricist : getOrCreateArtists(dbSession, trackInfo->lyricistArtists, true)) for (const Artist::pointer& lyricist : getOrCreateArtists(dbSession, trackMetadata.lyricistArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, lyricist, TrackArtistLinkType::Lyricist)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, lyricist, TrackArtistLinkType::Lyricist));
for (const Artist::pointer& mixer : getOrCreateArtists(dbSession, trackInfo->mixerArtists, true)) for (const Artist::pointer& mixer : getOrCreateArtists(dbSession, trackMetadata.mixerArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, mixer, TrackArtistLinkType::Mixer)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, mixer, TrackArtistLinkType::Mixer));
for (const auto& [role, performers] : trackInfo->performerArtists) for (const auto& [role, performers] : trackMetadata.performerArtists)
{ {
for (const Artist::pointer& performer : getOrCreateArtists(dbSession, performers, true)) for (const Artist::pointer& performer : getOrCreateArtists(dbSession, performers, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, performer, TrackArtistLinkType::Performer, role)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, performer, TrackArtistLinkType::Performer, role));
} }
for (const Artist::pointer& producer : getOrCreateArtists(dbSession, trackInfo->producerArtists, true)) for (const Artist::pointer& producer : getOrCreateArtists(dbSession, trackMetadata.producerArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, producer, TrackArtistLinkType::Producer)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, producer, TrackArtistLinkType::Producer));
for (const Artist::pointer& remixer : getOrCreateArtists(dbSession, trackInfo->remixerArtists, true)) for (const Artist::pointer& remixer : getOrCreateArtists(dbSession, trackMetadata.remixerArtists, true))
track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, remixer, TrackArtistLinkType::Remixer)); track.modify()->addArtistLink(TrackArtistLink::create(dbSession, track, remixer, TrackArtistLinkType::Remixer));
track.modify()->setScanVersion(_settings.scanVersion); track.modify()->setScanVersion(_settings.scanVersion);
if (trackInfo->medium && trackInfo->medium->release) if (trackMetadata.medium && trackMetadata.medium->release)
track.modify()->setRelease(getOrCreateRelease(dbSession, *trackInfo->medium->release, file.parent_path())); track.modify()->setRelease(getOrCreateRelease(dbSession, *trackMetadata.medium->release, file.parent_path()));
else else
track.modify()->setRelease({}); track.modify()->setRelease({});
track.modify()->setTotalTrack(trackInfo->medium ? trackInfo->medium->trackCount : std::nullopt); track.modify()->setTotalTrack(trackMetadata.medium ? trackMetadata.medium->trackCount : std::nullopt);
track.modify()->setReleaseReplayGain(trackInfo->medium ? trackInfo->medium->replayGain : std::nullopt); track.modify()->setReleaseReplayGain(trackMetadata.medium ? trackMetadata.medium->replayGain : std::nullopt);
track.modify()->setDiscSubtitle(trackInfo->medium ? trackInfo->medium->name : ""); track.modify()->setDiscSubtitle(trackMetadata.medium ? trackMetadata.medium->name : "");
track.modify()->setClusters(getOrCreateClusters(dbSession, trackInfo->userExtraTags)); track.modify()->setClusters(getOrCreateClusters(dbSession, trackMetadata.userExtraTags));
track.modify()->setLastWriteTime(lastWriteTime); track.modify()->setLastWriteTime(lastWriteTime);
track.modify()->setName(title); track.modify()->setName(title);
track.modify()->setDuration(trackInfo->duration); track.modify()->setDuration(trackMetadata.duration);
track.modify()->setBitrate(trackInfo->bitrate); track.modify()->setBitrate(trackMetadata.bitrate);
track.modify()->setAddedTime(Wt::WDateTime::currentDateTime()); track.modify()->setAddedTime(Wt::WDateTime::currentDateTime());
track.modify()->setTrackNumber(trackInfo->position); track.modify()->setTrackNumber(trackMetadata.position);
track.modify()->setDiscNumber(trackInfo->medium ? trackInfo->medium->position : std::nullopt); track.modify()->setDiscNumber(trackMetadata.medium ? trackMetadata.medium->position : std::nullopt);
track.modify()->setDate(trackInfo->date); track.modify()->setDate(trackMetadata.date);
track.modify()->setYear(trackInfo->year); track.modify()->setYear(trackMetadata.year);
track.modify()->setOriginalDate(trackInfo->originalDate); track.modify()->setOriginalDate(trackMetadata.originalDate);
track.modify()->setOriginalYear(trackInfo->originalYear); track.modify()->setOriginalYear(trackMetadata.originalYear);
// If a file has an OriginalDate but no date, set it to ease filtering // If a file has an OriginalDate but no date, set it to ease filtering
if (!trackInfo->date.isValid() && trackInfo->originalDate.isValid()) if (!trackMetadata.date.isValid() && trackMetadata.originalDate.isValid())
track.modify()->setDate(trackInfo->originalDate); track.modify()->setDate(trackMetadata.originalDate);
// If a file has an OriginalYear but no Year, set it to ease filtering // If a file has an OriginalYear but no Year, set it to ease filtering
if (!trackInfo->year && trackInfo->originalYear) if (!trackMetadata.year && trackMetadata.originalYear)
track.modify()->setYear(trackInfo->originalYear); track.modify()->setYear(trackMetadata.originalYear);
track.modify()->setRecordingMBID(trackInfo->recordingMBID); track.modify()->setRecordingMBID(trackMetadata.recordingMBID);
track.modify()->setTrackMBID(trackInfo->mbid); track.modify()->setTrackMBID(trackMetadata.mbid);
if (auto trackFeatures{ TrackFeatures::find(dbSession, track->getId()) }) if (auto trackFeatures{ TrackFeatures::find(dbSession, track->getId()) })
trackFeatures.remove(); // TODO: only if MBID changed? trackFeatures.remove(); // TODO: only if MBID changed?
track.modify()->setHasCover(trackInfo->hasCover); track.modify()->setHasCover(trackMetadata.hasCover);
track.modify()->setCopyright(trackInfo->copyright); track.modify()->setCopyright(trackMetadata.copyright);
track.modify()->setCopyrightURL(trackInfo->copyrightURL); track.modify()->setCopyrightURL(trackMetadata.copyrightURL);
track.modify()->setTrackReplayGain(trackInfo->replayGain); track.modify()->setTrackReplayGain(trackMetadata.replayGain);
track.modify()->setArtistDisplayName(trackInfo->artistDisplayName); track.modify()->setArtistDisplayName(trackMetadata.artistDisplayName);
} }
} }
src/libs/services/scanner/impl/ScanStepScanFiles.hpp
+41 -1
View File
@@ -19,11 +19,16 @@
#pragma once #pragma once
#include <condition_variable>
#include <deque>
#include <filesystem> #include <filesystem>
#include <mutex>
#include <span>
#include <string> #include <string>
#include <vector> #include <vector>
#include "metadata/IParser.hpp" #include "metadata/IParser.hpp"
#include "utils/IOContextRunner.hpp"
#include "ScanStepBase.hpp" #include "ScanStepBase.hpp"
namespace Scanner namespace Scanner
@@ -38,9 +43,44 @@ namespace Scanner
std::string_view getStepName() const override { return "Scanning files"; } std::string_view getStepName() const override { return "Scanning files"; }
void process(ScanContext& context) override; void process(ScanContext& context) override;
void scanAudioFile(const std::filesystem::path& file, ScanContext& context, const ScannerSettings::MediaLibraryInfo& libraryInfo); bool checkFileNeedScan(ScanContext& context, const std::filesystem::path& file, const ScannerSettings::MediaLibraryInfo& libraryInfo);
struct MetaDataScanResult
{
std::filesystem::path path;
std::optional<MetaData::Track> trackMetaData;
};
void processMetaDataScanResults(ScanContext& context, std::span<const std::unique_ptr<MetaDataScanResult>> scanResults, const ScannerSettings::MediaLibraryInfo& libraryInfo);
void processFileMetaData(ScanContext& context, const std::filesystem::path& file, const MetaData::Track& trackMetadata, const ScannerSettings::MediaLibraryInfo& libraryInfo);
std::unique_ptr<MetaData::IParser> _metadataParser; std::unique_ptr<MetaData::IParser> _metadataParser;
const std::vector<std::string> _extraTagsToParse{ "GENRE", "MOOD", "LANGUAGE", "ALBUMGROUPING" }; const std::vector<std::string> _extraTagsToParse{ "GENRE", "MOOD", "LANGUAGE", "ALBUMGROUPING" };
class MetadataScanQueue
{
public:
MetadataScanQueue(MetaData::IParser& parser, std::size_t threadCount);
std::size_t getThreadCount() const { return _scanContextRunner.getThreadCount(); }
void pushScanRequest(const std::filesystem::path path);
std::size_t getResultsCount() const;
size_t popResults(std::vector<std::unique_ptr<MetaDataScanResult>>& results, std::size_t maxCount);
void wait(std::size_t maxScanRequestCount = 0); // wait until ongoing scan request count <= maxScanRequestCount
private:
MetaData::IParser& _metadataParser;
boost::asio::io_context _scanContext;
IOContextRunner _scanContextRunner;
mutable std::mutex _mutex ;
std::size_t _ongoingScanCount{};
std::deque<std::unique_ptr<MetaDataScanResult>> _scanResults;
std::condition_variable _condVar;
};
MetadataScanQueue _metadataScanQueue;
std::deque<std::unique_ptr<MetaDataScanResult>> _metaDataScanResults;
}; };
} }
src/libs/utils/impl/IOContextRunner.cpp
+30 -25
View File
@@ -24,40 +24,45 @@
#include "utils/ILogger.hpp" #include "utils/ILogger.hpp"
IOContextRunner::IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount) IOContextRunner::IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount)
: _ioService {ioService} : _ioService{ ioService }
, _work {ioService} , _work{ ioService }
{ {
LMS_LOG(UTILS, INFO, "Starting IO context with " << threadCount << " threads..."); LMS_LOG(UTILS, INFO, "Starting IO context with " << threadCount << " threads...");
for (std::size_t i {}; i < threadCount; ++i) for (std::size_t i{}; i < threadCount; ++i)
{ {
_threads.emplace_back([&] _threads.emplace_back([&]
{ {
try try
{ {
_ioService.run(); _ioService.run();
} }
catch (const std::exception& e) catch (const std::exception& e)
{ {
LMS_LOG(UTILS, FATAL, "Exception caught in IO context: " << e.what()); LMS_LOG(UTILS, FATAL, "Exception caught in IO context: " << e.what());
std::abort(); std::abort();
} }
}); });
} }
} }
void void
IOContextRunner::stop() IOContextRunner::stop()
{ {
LMS_LOG(UTILS, DEBUG, "Stopping IO context..."); LMS_LOG(UTILS, DEBUG, "Stopping IO context...");
_work.reset(); _work.reset();
_ioService.stop(); _ioService.stop();
LMS_LOG(UTILS, DEBUG, "IO context stopped!"); LMS_LOG(UTILS, DEBUG, "IO context stopped!");
}
std::size_t IOContextRunner::getThreadCount() const
{
return _threads.size();
} }
IOContextRunner::~IOContextRunner() IOContextRunner::~IOContextRunner()
{ {
stop(); stop();
for (std::thread& t : _threads) for (std::thread& t : _threads)
t.join(); t.join();
} }
src/libs/utils/include/utils/IOContextRunner.hpp
+11 -12
View File
@@ -25,19 +25,18 @@
class IOContextRunner class IOContextRunner
{ {
public: public:
IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount); IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount);
~IOContextRunner(); ~IOContextRunner();
IOContextRunner(const IOContextRunner&) = delete; void stop();
IOContextRunner(IOContextRunner&&) = delete; std::size_t getThreadCount() const;
IOContextRunner& operator=(const IOContextRunner&) = delete;
IOContextRunner& operator=(IOContextRunner&&) = delete;
void stop(); private:
IOContextRunner(const IOContextRunner&) = delete;
IOContextRunner& operator=(const IOContextRunner&) = delete;
private: boost::asio::io_service& _ioService;
boost::asio::io_service& _ioService; std::optional<boost::asio::io_service::work> _work;
std::optional<boost::asio::io_service::work> _work; std::vector<std::thread> _threads;
std::vector<std::thread> _threads;
}; };