path: root/src/analyzer/analyzerthread.cpp

#include "analyzer/analyzerthread.h"

#include <mutex>

#include "analyzer/analyzerbeats.h"
#include "analyzer/analyzerebur128.h"
#include "analyzer/analyzergain.h"
#include "analyzer/analyzerkey.h"
#include "analyzer/analyzersilence.h"
#include "analyzer/analyzerwaveform.h"
#include "analyzer/constants.h"
#include "engine/engine.h"
#include "library/dao/analysisdao.h"
#include "moc_analyzerthread.cpp"
#include "sources/audiosourcestereoproxy.h"
#include "sources/soundsourceproxy.h"
#include "track/track.h"
#include "util/db/dbconnectionpooled.h"
#include "util/db/dbconnectionpooler.h"
#include "util/logger.h"
#include "util/timer.h"

namespace {

mixxx::Logger kLogger("AnalyzerThread");

// NOTE(uklotzde, 2018-11-23): The parameterization for the analyzers
// has not been touched while transforming the code from single- to
// multi-threaded processing! Feel free to adjust this if justified.

/// TODO(XXX): Use the vsync timer for the purpose of sending updates
// to the UI thread with a limited rate??

// Maximum frequency of progress updates while busy. A value of 60 ms
// results in ~17 progress updates per second which is sufficient for
// continuous feedback.
const mixxx::Duration kBusyProgressInhibitDuration = mixxx::Duration::fromMillis(60);

void deleteAnalyzerThread(AnalyzerThread* plainPtr) {
    if (plainPtr) {
        plainPtr->deleteAfterFinished();
    }
}

std::once_flag registerMetaTypesOnceFlag;

void registerMetaTypesOnce() {
    qRegisterMetaType<AnalyzerThreadState>();
    // AnalyzerProgress is just an alias/typedef and must be registered explicitly
    // by name!
    qRegisterMetaType<AnalyzerProgress>("AnalyzerProgress");
}

} // anonymous namespace

AnalyzerThread::NullPointer::NullPointer()
        : Pointer(nullptr, [](AnalyzerThread*) {}) {
}

//static
AnalyzerThread::Pointer AnalyzerThread::createInstance(
        int id,
        mixxx::DbConnectionPoolPtr dbConnectionPool,
        UserSettingsPointer pConfig,
        AnalyzerModeFlags modeFlags) {
    return Pointer(new AnalyzerThread(
                           id,
                           dbConnectionPool,
                           pConfig,
                           modeFlags),
            deleteAnalyzerThread);
}

AnalyzerThread::AnalyzerThread(
        int id,
        mixxx::DbConnectionPoolPtr dbConnectionPool,
        UserSettingsPointer pConfig,
        AnalyzerModeFlags modeFlags)
        : WorkerThread(
            QString("AnalyzerThread %1").arg(id),
            (modeFlags & AnalyzerModeFlags::LowPriority ? QThread::LowPriority : QThread::InheritPriority)),
          m_id(id),
          m_dbConnectionPool(std::move(dbConnectionPool)),
          m_pConfig(pConfig),
          m_modeFlags(modeFlags),
          m_nextTrack(2), // minimum capacity
          m_sampleBuffer(mixxx::kAnalysisSamplesPerChunk),
          m_emittedState(AnalyzerThreadState::Void) {
    std::call_once(registerMetaTypesOnceFlag, registerMetaTypesOnce);
}

void AnalyzerThread::doRun() {
    std::unique_ptr<AnalysisDao> pAnalysisDao;
    // The thread-local database connection  must not be closed
    // before returning from this function.
    mixxx::DbConnectionPooler dbConnectionPooler;

    if (m_modeFlags & AnalyzerModeFlags::WithWaveform) {
        dbConnectionPooler = mixxx::DbConnectionPooler(m_dbConnectionPool); // move assignment
        if (!dbConnectionPooler.isPooling()) {
            kLogger.warning()
                    << "Failed to obtain database connection for analyzer thread";
            return;
        }
        QSqlDatabase dbConnection = mixxx::DbConnectionPooled(m_dbConnectionPool);
        m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerWaveform>(m_pConfig, dbConnection)));
    }
    if (AnalyzerGain::isEnabled(ReplayGainSettings(m_pConfig))) {
        m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerGain>(m_pConfig)));
    }
    if (AnalyzerEbur128::isEnabled(ReplayGainSettings(m_pConfig))) {
        m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerEbur128>(m_pConfig)));
    }
    // BPM detection might be disabled in the config, but can be overridden
    // and enabled by explicitly setting the mode flag.
    const bool enforceBpmDetection = (m_modeFlags & AnalyzerModeFlags::WithBeats) != 0;
    m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerBeats>(m_pConfig, enforceBpmDetection)));
    m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerKey>(m_pConfig)));
    m_analyzers.push_back(AnalyzerWithState(std::make_unique<AnalyzerSilence>(m_pConfig)));
    DEBUG_ASSERT(!m_analyzers.empty());
    kLogger.debug() << "Activated" << m_analyzers.size() << "analyzers";

    m_lastBusyProgressEmittedTimer.start();

    mixxx::AudioSource::OpenParams openParams;
    openParams.setChannelCount(mixxx::kAnalysisChannels);

    while (awaitWorkItemsFetched()) {
        DEBUG_ASSERT(m_currentTrack);
        kLogger.debug() << "Analyzing" << m_currentTrack->getFileInfo();

        // Get the audio
        const auto audioSource =
                SoundSourceProxy(m_currentTrack).openAudioSource(openParams);
        if (!audioSource) {
            kLogger.warning()
                    << "Failed to open file for analyzing:"
                    << m_currentTrack->getFileInfo();
            emitDoneProgress(kAnalyzerProgressUnknown);
            continue;
        }

        bool processTrack = false;
        for (auto&& analyzer : m_analyzers) {
            // Make sure not to short-circuit initialize(...)
            if (analyzer.initialize(
                        m_currentTrack,
                        audioSource->getSignalInfo().getSampleRate(),
                        audioSource->frameLength() * mixxx::kAnalysisChannels)) {
                processTrack = true;
            }
        }

        if (processTrack) {
            const auto analysisResult = analyzeAudioSource(audioSource);
            DEBUG_ASSERT(analysisResult != AnalysisResult::Pending);
            if (analysisResult == AnalysisResult::Finished) {
                // The analysis has been finished, and is either complete without
                // any errors or partial if it has been aborted due to a corrupt
                // audio file. In both cases don't reanalyze tracks during this
                // session. A partial analysis would otherwise be repeated again
                // and again, because it is very unlikely that the error vanishes
                // suddenly.
                emitBusyProgress(kAnalyzerProgressFinalizing);
                // This takes around 3 sec on a Atom Netbook
                for (auto&& analyzer : m_analyzers) {
                    analyzer.finish(m_currentTrack);
                }
                emitDoneProgress(kAnalyzerProgressDone);
            } else {
                for (auto&& analyzer : m_analyzers) {
                    analyzer.cancel();
                }
                emitDoneProgress(kAnalyzerProgressUnknown);
            }
        } else {
            kLogger.debug() << "Skipping track analysis because no analyzer initialized.";
            emitDoneProgress(kAnalyzerProgressDone);
        }
    }
    DEBUG_ASSERT(!m_currentTrack);
    DEBUG_ASSERT(isStopping());

    m_analyzers.clear();

    kLogger.debug() << "Exiting worker thread";
    emitProgress(AnalyzerThreadState::Exit);
}

bool AnalyzerThread::submitNextTrack(TrackPointer nextTrack) {
    DEBUG_ASSERT(nextTrack);
    kLogger.debug()
            << "Enqueueing next track"
            << nextTrack->getId();
    if (m_nextTrack.try_emplace(std::move(nextTrack))) {
        // Ensure that the submitted track gets processed eventually
        // by waking the worker thread up after adding a new task to
        // its back queue! Otherwise the thread might not notice if
        // it is currently idle and has fallen asleep.
        wake();
        return true;
    }
    return false;
}

WorkerThread::TryFetchWorkItemsResult AnalyzerThread::tryFetchWorkItems() {
    DEBUG_ASSERT(!m_currentTrack);
    TrackPointer* pFront = m_nextTrack.front();
    if (pFront) {
        m_currentTrack = *pFront;
        m_nextTrack.pop();
        kLogger.debug()
                << "Dequeued next track"
                << m_currentTrack->getId();
        return TryFetchWorkItemsResult::Ready;
    } else {
        emitProgress(AnalyzerThreadState::Idle);
        return TryFetchWorkItemsResult::Idle;
    }
}

AnalyzerThread::AnalysisResult AnalyzerThread::analyzeAudioSource(
        const mixxx::AudioSourcePointer& audioSource) {
    DEBUG_ASSERT(m_currentTrack);

    mixxx::AudioSourceStereoProxy audioSourceProxy(
            audioSource,
            mixxx::kAnalysisFramesPerChunk);
    DEBUG_ASSERT(
            audioSourceProxy.getSignalInfo().getChannelCount() ==
            mixxx::kAnalysisChannels);

    // Analysis starts now
    emitBusyProgress(kAnalyzerProgressNone);

    mixxx::IndexRange remainingFrameRange = audioSource->frameIndexRange();
    while (!remainingFrameRange.empty()) {
        sleepWhileSuspended();
        if (isStopping()) {
            return AnalysisResult::Cancelled;
        }

        // 1st step: Decode next chunk of audio data

        // Split the range for the next chunk from the remaining (= to-be-analyzed) frames
        auto chunkFrameRange =
                remainingFrameRange.splitAndShrinkFront(
                        math_min(mixxx::kAnalysisFramesPerChunk, remainingFrameRange.length()));
        DEBUG_ASSERT(!chunkFrameRange.empty());

        // Request the next chunk of audio data
        const auto readableSampleFrames =
                audioSourceProxy.readSampleFrames(
                        mixxx::WritableSampleFrames(
                                chunkFrameRange,
                                mixxx::SampleBuffer::WritableSlice(m_sampleBuffer)));
        // The returned range fits into the requested range
        DEBUG_ASSERT(readableSampleFrames.frameIndexRange().isSubrangeOf(chunkFrameRange));

        // Sometimes the duration of the audio source is inaccurate and adjusted
        // while reading. We need to adjust all frame ranges to reflect this new
        // situation by restoring all invariants and consistency requirements!

        // Shrink the original range of the current chunks to the actual available
        // range.
        chunkFrameRange = intersect(chunkFrameRange, audioSourceProxy.frameIndexRange());
        // The audio data that has just been read should still fit into the adjusted
        // chunk range.
        DEBUG_ASSERT