Multithreaded analysis #1069
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,206 @@ | ||
| #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::m_pAnalyzerManager = NULL; | ||
|
|
||
|
|
||
| //static | ||
| AnalyzerManager& AnalyzerManager::getInstance(UserSettingsPointer pConfig) { | ||
| if (!m_pAnalyzerManager) { | ||
| // There exists only one UserSettingsPointer in the app, so it doens't matter | ||
| // if we only assign it once. | ||
|
There was a problem hiding this comment. I do not understand the comment. It is untypical to pass a parameter in the getInstance() call. There was a problem hiding this comment. The implementation needs that there exists only one manager so that it can control the number of threads in execution. Since it is used from two different places (analysisfeature and playermanager), it cannot be an instance on those classes either. Still, I think that I was wrong in using the getInstance() naming/model and should have been a MixxxMainWindow instance, where I have UserSettingsPointer. I was busy enough trying to understand QThread and signals so I took the easiest path. Now I am open to change this if you give me the correct guidelines to do so. There was a problem hiding this comment. I would prefer this not become a singleton and instead pass it throughout to all the places it needs to go. Once #941 is in, there will be a better place for it in the Core class but for now it can go in the main window (src/mixxx.cpp). |
||
| m_pAnalyzerManager = new AnalyzerManager(pConfig); | ||
| } | ||
| int maxThreads = m_pAnalyzerManager->m_pConfig->getValue<int>(ConfigKey("[Library]", "MaxAnalysisThreads")); | ||
|
There was a problem hiding this comment. We have the use case, that some users (incuding me) have shared preferences on different machines. There was a problem hiding this comment. Actually, the setting was supposed to be a safeguard, and still, there are more safeguards that should control your use case: Concretely, giving the user the option to control the number of threads allows a way to reduce the usage of resources in case that the machine is not able to keep with the desired audio latency. Finally, on the AnalyzerManager, there is a safety measure where it checks that the idealthreadcount is sane for the machine in use. There was a problem hiding this comment. I am afraid the average user does not even know what threads are. If we found out during testing that we have an performance issue, it is our job to tweak solve it. Is there a use case to pick single thread analysis? If yes, a boolean option might be a better choice. There was a problem hiding this comment. I haven't had any issue. I've tested it on an i7 (4 cores, 8 threads), a core 2 duo (2 cores, 2 threads), and a virtual machine with 2 cores assigned. None of them gave me issues when analyzing during playback using the idealthreadcount. On the virtual machine i needed to have a latency of 90ms, but that's a bug of virtualbox. There was a problem hiding this comment. FWIW, single core CPUs are rather rare nowadays.
That makes sense. This could be useful, for example, if a user wants to analyze their library but also keep 1 CPU core unused by Mixxx for doing something else with their computer while Mixxx is analyzing. |
||
| if (maxThreads < 0 || maxThreads > 2 * QThread::idealThreadCount()) { | ||
| //Assume the value is incorrect, so fix it. | ||
| maxThreads = QThread::idealThreadCount(); | ||
| m_pAnalyzerManager->m_pConfig->setValue<int>(ConfigKey("[Library]", "MaxAnalysisThreads"), maxThreads); | ||
| } | ||
| if (maxThreads == 0) { | ||
| maxThreads = QThread::idealThreadCount(); | ||
| } | ||
| m_pAnalyzerManager->m_MaxThreads = maxThreads; | ||
| return *m_pAnalyzerManager; | ||
| } | ||
|
|
||
| AnalyzerManager::AnalyzerManager(UserSettingsPointer pConfig) : | ||
| m_pConfig(pConfig), | ||
| m_batchTrackQueue(), | ||
| m_prioTrackQueue(), | ||
| m_backgroundWorkers(), | ||
| m_foregroundWorkers(), | ||
| m_pausedWorkers() { | ||
| } | ||
|
|
||
| AnalyzerManager::~AnalyzerManager() { | ||
| stop(true); | ||
| } | ||
|
|
||
| bool AnalyzerManager::isActive(bool includeForeground) { | ||
|
There was a problem hiding this comment. isActive(true) is hard to read at the caller. |
||
| //I don't count foregroundWorkers because | ||
| int total = (includeForeground ? m_foregroundWorkers.size() : 0) + | ||
| m_backgroundWorkers.size() + m_pausedWorkers.size(); | ||
| return total > 0; | ||
| } | ||
|
|
||
| void AnalyzerManager::stop(bool shutdown) { | ||
| QListIterator<AnalyzerWorker*> it(m_backgroundWorkers); | ||
| while (it.hasNext()) { | ||
| AnalyzerWorker* worker = it.next(); | ||
| worker->endProcess(); | ||
| } | ||
| QListIterator<AnalyzerWorker*> it3(m_pausedWorkers); | ||
| while (it3.hasNext()) { | ||
| AnalyzerWorker* worker = it3.next(); | ||
| worker->endProcess(); | ||
| } | ||
| if (shutdown) { | ||
| QListIterator<AnalyzerWorker*> it2(m_foregroundWorkers); | ||
| while (it2.hasNext()) { | ||
| AnalyzerWorker* worker = it2.next(); | ||
| worker->endProcess(); | ||
| } | ||
| //TODO: ensure that they are all forcibly stopped. | ||
| } | ||
| } | ||
| // Analyze it with a foreground worker. (foreground as in interactive, i.e. not a batch worker). | ||
| void AnalyzerManager::analyseTrackNow(TrackPointer tio) { | ||
| if (m_batchTrackQueue.contains(tio)) { | ||
| m_batchTrackQueue.removeAll(tio); | ||
| } | ||
| if (!m_prioTrackQueue.contains(tio)) { | ||
| m_prioTrackQueue.append(tio); | ||
| if (m_foregroundWorkers.size() < m_MaxThreads) { | ||
| createNewWorker(false); | ||
| if (m_foregroundWorkers.size() + m_backgroundWorkers.size() > m_MaxThreads) { | ||
| AnalyzerWorker * backwork = m_backgroundWorkers.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_backgroundWorkers.removeAll(backwork); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| // This is called from the GUI for batch analysis. | ||
| void AnalyzerManager::queueAnalyseTrack(TrackPointer tio) { | ||
| if (!m_batchTrackQueue.contains(tio)) { | ||
| m_batchTrackQueue.append(tio); | ||
| if (m_pausedWorkers.size() + m_backgroundWorkers.size() < m_MaxThreads) { | ||
| createNewWorker(true); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // 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(struct AnalyzerWorker::progress_info* progressInfo) { | ||
| if (progressInfo->current_track) { | ||
| progressInfo->current_track->setAnalyzerProgress( | ||
| progressInfo->track_progress); | ||
| if (progressInfo->track_progress == 1000) { | ||
| emit(trackDone(progressInfo->current_track)); | ||
| } | ||
| progressInfo->current_track.clear(); | ||
| } | ||
| emit(trackProgress(progressInfo->track_progress / 10)); | ||
| if (progressInfo->track_progress == 1000) { | ||
| emit(trackFinished(m_backgroundWorkers.size() + m_batchTrackQueue.size() - 1)); | ||
| } | ||
| progressInfo->sema.release(); | ||
| } | ||
|
|
||
| void AnalyzerManager::slotNextTrack(AnalyzerWorker* worker) { | ||
| //The while loop is done in the event that the track which was added to the queue is no | ||
| //longer available. | ||
|
|
||
| //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. | ||
| TrackPointer track = TrackPointer(); | ||
|
There was a problem hiding this comment. " = TrackPointer();" is redundant. |
||
| if (worker->isBatch()) { | ||
| while (!track && !m_batchTrackQueue.isEmpty()) { | ||
| track = m_batchTrackQueue.dequeue(); | ||
| } | ||
| } | ||
| else { | ||
| while (!track && !m_prioTrackQueue.isEmpty()) { | ||
| track = m_prioTrackQueue.dequeue(); | ||
| } | ||
| } | ||
| if (track) { | ||
| worker->nextTrack(track); | ||
| } | ||
| else { | ||
| worker->endProcess(); | ||
| if (!m_pausedWorkers.isEmpty()) { | ||
| AnalyzerWorker* otherworker = m_pausedWorkers.first(); | ||
| otherworker->resume(); | ||
| m_pausedWorkers.removeOne(otherworker); | ||
| } | ||
| } | ||
| //Check if all queues are empty. | ||
| if (!isActive(false)) { | ||
| emit(queueEmpty()); | ||
| } | ||
| } | ||
| void AnalyzerManager::slotWorkerFinished(AnalyzerWorker* worker) { | ||
| m_backgroundWorkers.removeAll(worker); | ||
| m_foregroundWorkers.removeAll(worker); | ||
| m_pausedWorkers.removeAll(worker); | ||
| } | ||
| 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; | ||
| } | ||
|
|
||
| AnalyzerWorker* AnalyzerManager::createNewWorker(bool batchJob) { | ||
|
|
||
| QThread* thread = new QThread(); | ||
| AnalyzerWorker* worker = new AnalyzerWorker(m_pConfig, batchJob); | ||
| 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(struct AnalyzerWorker::progress_info*)), this, SLOT(slotUpdateProgress(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(); | ||
| if (batchJob) { | ||
| m_backgroundWorkers.append(worker); | ||
| } | ||
| else { | ||
| m_foregroundWorkers.append(worker); | ||
|
There was a problem hiding this comment. What makes a forground worker a "forground" worker? There was a problem hiding this comment. I think it is "priorityWorkers" There was a problem hiding this comment. It's a bad naming from my side, but I couldn't think of a better one at that time. Maybe I should have used worker and priorityworker. Take in mind that the priority is not the only difference, since there are some Mixxx settings that affect them in a different way. (namely the waveform creation and the bpm analysis) |
||
| } | ||
| return worker; | ||
| } | ||
|
|
||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| #ifndef ANALYZER_ANALYZERMANAGER_H | ||
| #define ANALYZER_ANALYZERMANAGER_H | ||
|
|
||
| #include <QList> | ||
| #include <QQueue> | ||
|
|
||
| #include <vector> | ||
|
|
||
| #include "analyzer/analyzerworker.h" | ||
| #include "preferences/usersettings.h" | ||
| #include "track/track.h" | ||
|
|
||
| class TrackCollection; | ||
|
|
||
| class AnalyzerManager : public QObject { | ||
| Q_OBJECT | ||
|
|
||
| protected: | ||
| AnalyzerManager(UserSettingsPointer pConfig); | ||
|
|
||
| public: | ||
| static AnalyzerManager& getInstance(UserSettingsPointer pConfig); | ||
| virtual ~AnalyzerManager(); | ||
|
|
||
| void stop(bool shutdown); | ||
| //This method might need to be protected an called only via slot. | ||
| void analyseTrackNow(TrackPointer tio); | ||
| void queueAnalyseTrack(TrackPointer tio); | ||
| bool isActive(bool includeForeground); | ||
|
|
||
|
|
||
| public slots: | ||
| // This slot is called from the decks and samplers when the track is loaded. | ||
| void slotAnalyseTrack(TrackPointer tio); | ||
| void slotUpdateProgress(struct AnalyzerWorker::progress_info*); | ||
| void slotNextTrack(AnalyzerWorker*); | ||
| void slotPaused(AnalyzerWorker*); | ||
| void slotWorkerFinished(AnalyzerWorker*); | ||
| void slotErrorString(QString); | ||
|
|
||
| signals: | ||
| void trackProgress(int progress); | ||
| void trackDone(TrackPointer track); | ||
| void trackFinished(int size); | ||
| void queueEmpty(); | ||
| private: | ||
|
|
||
| AnalyzerWorker* createNewWorker(bool batchJob); | ||
|
|
||
| static AnalyzerManager* m_pAnalyzerManager; | ||
| UserSettingsPointer m_pConfig; | ||
| int m_MaxThreads; | ||
| // The processing queue and associated mutex | ||
| QQueue<TrackPointer> m_batchTrackQueue; | ||
| QQueue<TrackPointer> m_prioTrackQueue; | ||
|
|
||
| QList<AnalyzerWorker*> m_backgroundWorkers; | ||
| QList<AnalyzerWorker*> m_foregroundWorkers; | ||
| QList<AnalyzerWorker*> m_pausedWorkers; | ||
| }; | ||
|
|
||
| #endif /* ANALYZER_ANALYZERMANAGER_H */ |
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