Multithreaded analysis #1069
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,296 @@ | ||
| #include "analyzer/analyzermanager.h" | ||
|
|
||
| #include <typeinfo> | ||
| #include <QThread> | ||
|
|
||
| #include <QtDebug> | ||
| #include <QMutexLocker> | ||
| #include <QListIterator> | ||
|
|
||
| #include "library/trackcollection.h" | ||
| #include "mixer/playerinfo.h" | ||
| #include "track/track.h" | ||
| #include "util/compatibility.h" | ||
| #include "util/event.h" | ||
| #include "util/timer.h" | ||
| #include "util/trace.h" | ||
|
|
||
| AnalyzerManager::AnalyzerManager(UserSettingsPointer pConfig) : | ||
| m_pConfig(pConfig), | ||
| m_nextWorkerId(0), | ||
| m_defaultTrackQueue(), | ||
| m_prioTrackQueue(), | ||
| m_defaultWorkers(), | ||
| m_priorityWorkers(), | ||
| m_pausedWorkers() { | ||
|
|
||
| int maxThreads = m_pConfig->getValue<int>(ConfigKey("[Library]", "MaxAnalysisThreads")); | ||
| int ideal = QThread::idealThreadCount(); | ||
|
There was a problem hiding this comment. Ideal can be moved into the if branch below. There was a problem hiding this comment. Oh. yup... Initially I was using the ideal in the if, instead of the hardcoded 32, but then i had the use case where i get -1 for ideal and i wasn't sure how to handle that. |
||
| if (ideal < 1) { | ||
| if (maxThreads > 0 && maxThreads <= 32) { | ||
| qDebug() << "Cannot detect idealThreadCount. maxThreads is: " << maxThreads; | ||
| ideal = maxThreads; | ||
| } | ||
| else { | ||
| qWarning() << "Cannot detect idealThreadCount. Using the sane value of 1"; | ||
| ideal = 1; | ||
| } | ||
| } | ||
| if (maxThreads <= 0 || maxThreads > ideal) { | ||
| qWarning() << "maxThreads value is incorrect. Changing it to " << ideal; | ||
|
There was a problem hiding this comment. this should also print the invalid value. There was a problem hiding this comment. It should not add issue a warning if the Mixxx starts the first time with this new preference option. There was a problem hiding this comment. Ok, will print that too. There was a problem hiding this comment. This should work: There was a problem hiding this comment. Oh, Yup, that would work too, although I meant where is the configuration initizalized for the first time, so that I could assign the ideal value to it at that time. There was a problem hiding this comment. With exact this command. If "[Library]", "MaxAnalysisThreads" is not set ideal is returned. |
||
| //Assume the value is incorrect, so fix it. | ||
| maxThreads = ideal; | ||
| } | ||
| m_MaxThreads = maxThreads; | ||
| } | ||
|
|
||
| AnalyzerManager::~AnalyzerManager() { | ||
| stop(true); | ||
| } | ||
|
|
||
| bool AnalyzerManager::isActive() { | ||
| int total = m_priorityWorkers.size() + | ||
| m_defaultWorkers.size() + m_pausedWorkers.size(); | ||
| return total > 0; | ||
| } | ||
| bool AnalyzerManager::isDefaultQueueActive() { | ||
| int total = m_defaultWorkers.size() + m_pausedWorkers.size(); | ||
|
|
||
| return total > 0; | ||
| } | ||
|
|
||
| void AnalyzerManager::stop(bool shutdown) { | ||
| m_defaultTrackQueue.clear(); | ||
| QListIterator<AnalyzerWorker*> it(m_defaultWorkers); | ||
|
There was a problem hiding this comment. Range based for loops more compact and easier to grasp than while loops with subsequently numbered local variables. |
||
| while (it.hasNext()) { | ||
| AnalyzerWorker* worker = it.next(); | ||
| worker->endProcess(); | ||
| m_endingWorkers.append(worker); | ||
| } | ||
| QListIterator<AnalyzerWorker*> it3(m_pausedWorkers); | ||
| while (it3.hasNext()) { | ||
| AnalyzerWorker* worker = it3.next(); | ||
| worker->endProcess(); | ||
| m_endingWorkers.append(worker); | ||
| } | ||
| if (shutdown) { | ||
| m_prioTrackQueue.clear(); | ||
| QListIterator<AnalyzerWorker*> it2(m_priorityWorkers); | ||
| while (it2.hasNext()) { | ||
| AnalyzerWorker* worker = it2.next(); | ||
| worker->endProcess(); | ||
| m_endingWorkers.append(worker); | ||
| } | ||
| //TODO: ensure that they are all forcibly stopped. | ||
| } | ||
| } | ||
| //Add a track to be analyzed with a priority worker. (Like those required by loading a track into a player). | ||
| void AnalyzerManager::analyseTrackNow(TrackPointer tio) { | ||
| if (m_defaultTrackQueue.contains(tio)) { | ||
| m_defaultTrackQueue.removeAll(tio); | ||
| } | ||
| //TODO: There's one scenario that we still miss: load on a deck a track that is currently | ||
| //being analyzed by the background worker. We cannot reuse the background worker, but we should discard its work. | ||
| if (!m_prioTrackQueue.contains(tio)) { | ||
| m_prioTrackQueue.append(tio); | ||
| if (m_priorityWorkers.size() < m_MaxThreads) { | ||
| createNewWorker(WorkerType::priorityWorker); | ||
| if (m_priorityWorkers.size() + m_defaultWorkers.size() > m_MaxThreads) { | ||
| AnalyzerWorker * backwork = m_defaultWorkers.first(); | ||
| backwork->pause(); | ||
| //Ideally i would have done this on the slotPaused slot, but then i cannot | ||
| //ensure i won't call pause twice for the same worker. | ||
| m_pausedWorkers.append(backwork); | ||
| m_defaultWorkers.removeAll(backwork); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| // This is called from the GUI for the analysis feature of the library. | ||
| void AnalyzerManager::queueAnalyseTrack(TrackPointer tio) { | ||
| if (!m_defaultTrackQueue.contains(tio)) { | ||
| m_defaultTrackQueue.append(tio); | ||
| if (m_pausedWorkers.size() + m_defaultWorkers.size() < m_MaxThreads) { | ||
| createNewWorker(WorkerType::defaultWorker); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // This slot is called from the decks and samplers when the track is loaded. | ||
| void AnalyzerManager::slotAnalyseTrack(TrackPointer tio) { | ||
| analyseTrackNow(tio); | ||
| } | ||
|
|
||
|
|
||
| //slot | ||
| void AnalyzerManager::slotUpdateProgress(int workerIdx, struct AnalyzerWorker::progress_info* progressInfo) { | ||
| //Updates to wave overview and player status text comes from a signal emited from the track by calling setAnalyzerProgress. | ||
| progressInfo->current_track->setAnalyzerProgress(progressInfo->track_progress); | ||
| //These update the Analysis feature and analysis view. | ||
| emit(trackProgress(workerIdx, progressInfo->track_progress / 10)); | ||
| if (progressInfo->track_progress == 1000) { | ||
| //Right now no one is listening to trackDone, but it's here just in case. | ||
| emit(trackDone(progressInfo->current_track)); | ||
| //Report that a track analysis has finished, and how many are still remaining. | ||
| emit(trackFinished(m_defaultWorkers.size() + m_defaultTrackQueue.size() - 1)); | ||
| } | ||
| //TODO: Which is the consequence of not calling reset? | ||
| progressInfo->current_track.reset(); | ||
| progressInfo->sema.release(); | ||
| } | ||
|
|
||
| void AnalyzerManager::slotNextTrack(AnalyzerWorker* worker) { | ||
| //TODO: The old scan checked in isLoadedTrackWaiting for pTrack->getAnalyzerProgress() | ||
| // and either tried to load a previuos scan, or discarded the track if it had already been | ||
| // analyzed. I don't fully understand the scenario and I am not doing that right now. | ||
|
|
||
| //This is used when the maxThreads change. Extra workers are paused until active workers end. | ||
| //Then, those are terminated and the paused workers are resumed. | ||
| TrackPointer track; | ||
| AnalyzerWorker* forepaused=nullptr; | ||
| foreach(AnalyzerWorker* worker, m_pausedWorkers) { | ||
|
There was a problem hiding this comment. worker is also user outside the loop. Please pick a new name to avoid confusion. There was a problem hiding this comment. Ouch! 👍 Indeed, i added that when adding the refresh of the threads from the preferences, but I overlooked the names. |
||
| if (worker->isPriorized()) { forepaused=worker; break; } | ||
| } | ||
| if (!forepaused) { | ||
| if (worker->isPriorized()) { | ||
| while (!track && !m_prioTrackQueue.isEmpty()) { | ||
| track = m_prioTrackQueue.dequeue(); | ||
| } | ||
| } | ||
| else { | ||
| if (m_defaultWorkers.size() + m_pausedWorkers.size() <= m_MaxThreads) { | ||
| //The while loop is done in the event that the track which was added to the queue is no | ||
| //longer available. | ||
| while (!track && !m_defaultTrackQueue.isEmpty()) { | ||
| track = m_defaultTrackQueue.dequeue(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| if (track) { | ||
| worker->nextTrack(track); | ||
| } | ||
| else { | ||
| worker->endProcess(); | ||
| //Removing from active lists, so that "isActive" can return the correct value. | ||
| m_defaultWorkers.removeAll(worker); | ||
| m_priorityWorkers.removeAll(worker); | ||
| m_endingWorkers.append(worker); | ||
|
|
||
| if (forepaused) { | ||
| forepaused->resume(); | ||
| m_pausedWorkers.removeOne(forepaused); | ||
| m_priorityWorkers.append(forepaused); | ||
| } | ||
| else if (!m_pausedWorkers.isEmpty()) { | ||
| AnalyzerWorker* otherworker = m_pausedWorkers.first(); | ||
| otherworker->resume(); | ||
| m_pausedWorkers.removeOne(otherworker); | ||
| if (otherworker->isPriorized()) { | ||
| m_priorityWorkers.append(otherworker); | ||
| } | ||
| else { | ||
| m_defaultWorkers.append(otherworker); | ||
| } | ||
| } | ||
| } | ||
| //Check if background workers are empty. | ||
| if (!isDefaultQueueActive()) { | ||
| emit(queueEmpty()); | ||
| } | ||
| } | ||
| void AnalyzerManager::slotWorkerFinished(AnalyzerWorker* worker) { | ||
| m_endingWorkers.removeAll(worker); | ||
| m_defaultWorkers.removeAll(worker); | ||
| m_priorityWorkers.removeAll(worker); | ||
| m_pausedWorkers.removeAll(worker); | ||
| if (!isDefaultQueueActive()) { | ||
| emit(queueEmpty()); | ||
| } | ||
| } | ||
| void AnalyzerManager::slotPaused(AnalyzerWorker* worker) { | ||
| //No useful code to execute right now. | ||
|
There was a problem hiding this comment. Please add Q_UNUSED(worker); to eliminate a warning |
||
| } | ||
| void AnalyzerManager::slotErrorString(QString errMsg) { | ||
| //TODO: This is currently unused. | ||
| qWarning() << "Testing with :" << errMsg; | ||
| } | ||
|
|
||
|
|
||
| void AnalyzerManager::slotMaxThreadsChanged(int threads) { | ||
| // If it is Active, adapt the amount of workers. If it is not active, it will just update the variable. | ||
| if (threads < m_MaxThreads) { | ||
| //Pause workers | ||
| while (!m_defaultWorkers.isEmpty() | ||
| && m_priorityWorkers.size() + m_defaultWorkers.size() > threads) { | ||
| AnalyzerWorker * backwork = m_defaultWorkers.first(); | ||
| backwork->pause(); | ||
| //Ideally i would have done this on the slotPaused slot, but then i cannot | ||
| //ensure i won't call pause twice for the same worker. | ||
| m_pausedWorkers.append(backwork); | ||
| m_defaultWorkers.removeAll(backwork); | ||
| } | ||
| while (m_priorityWorkers.size() > threads) { | ||
| AnalyzerWorker * backwork = m_priorityWorkers.first(); | ||
| backwork->pause(); | ||
| //Ideally i would have done this on the slotPaused slot, but then i cannot | ||
| //ensure i won't call pause twice for the same worker. | ||
| m_pausedWorkers.append(backwork); | ||
| m_priorityWorkers.removeAll(backwork); | ||
| } | ||
| } | ||
| else { | ||
| //resume workers | ||
| int pendingworkers=threads-m_MaxThreads; | ||
| foreach(AnalyzerWorker* worker, m_pausedWorkers) { | ||
| if (worker->isPriorized() && pendingworkers > 0) { | ||
| worker->resume(); | ||
| m_pausedWorkers.removeOne(worker); | ||
| m_priorityWorkers.append(worker); | ||
| --pendingworkers; | ||
| } | ||
| } | ||
| foreach(AnalyzerWorker* worker, m_pausedWorkers) { | ||
| if (!worker->isPriorized() && pendingworkers > 0) { | ||
| worker->resume(); | ||
| m_pausedWorkers.removeOne(worker); | ||
| m_defaultWorkers.append(worker); | ||
| --pendingworkers; | ||
| } | ||
| } | ||
| //Create new workers, if tracks in queue. | ||
| pendingworkers = math_min(pendingworkers,m_defaultTrackQueue.size()); | ||
| for ( ;pendingworkers > 0; --pendingworkers) { | ||
| createNewWorker(WorkerType::defaultWorker); | ||
| } | ||
| } | ||
| m_MaxThreads=threads; | ||
| } | ||
|
|
||
| AnalyzerWorker* AnalyzerManager::createNewWorker(WorkerType wtype) { | ||
| bool priorized = (wtype == WorkerType::priorityWorker); | ||
| QThread* thread = new QThread(); | ||
| AnalyzerWorker* worker = new AnalyzerWorker(m_pConfig, ++m_nextWorkerId, priorized); | ||
| worker->moveToThread(thread); | ||
| //Auto startup and auto cleanup of worker and thread. | ||
| connect(thread, SIGNAL(started()), worker, SLOT(slotProcess())); | ||
| connect(worker, SIGNAL(finished()), thread, SLOT(quit())); | ||
| connect(worker, SIGNAL(finished()), worker, SLOT(deleteLater())); | ||
| connect(thread, SIGNAL(finished()), thread, SLOT(deleteLater())); | ||
| //Connect with manager. | ||
| connect(worker, SIGNAL(updateProgress(int, struct AnalyzerWorker::progress_info*)), this, SLOT(slotUpdateProgress(int, struct AnalyzerWorker::progress_info*))); | ||
| connect(worker, SIGNAL(waitingForNextTrack(AnalyzerWorker*)), this, SLOT(slotNextTrack(AnalyzerWorker*))); | ||
| connect(worker, SIGNAL(paused(AnalyzerWorker*)), this, SLOT(slotPaused(AnalyzerWorker*))); | ||
| connect(worker, SIGNAL(workerFinished(AnalyzerWorker*)), this, SLOT(slotWorkerFinished(AnalyzerWorker*))); | ||
| connect(worker, SIGNAL(error(QString)), this, SLOT(slotErrorString(QString))); | ||
| thread->start(QThread::LowPriority); | ||
| if (priorized) { | ||
| m_priorityWorkers.append(worker); | ||
| } | ||
| else { | ||
| m_defaultWorkers.append(worker); | ||
| } | ||
| return worker; | ||
| } | ||
|
|
||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Note: These two changes were done for compatibility when building with QT5. If needed, they can be restored