feat: multithreaded rubberband for stem

src/engine/bufferscalers/enginebufferscalerubberband.cpp

@@ -8,6 +8,7 @@
 #include "util/counter.h"
 #include "util/defs.h"
 #include "util/math.h"
+#include "util/mutex.h"
 #include "util/sample.h"
 #include "util/timer.h"
 
@@ -57,7 +58,7 @@ void EngineBufferScaleRubberBand::setScaleParameters(double base_rate,
 
     if (pitchScale > 0) {
         //qDebug() << "EngineBufferScaleRubberBand setPitchScale" << *pitch << pitchScale;
-        m_pRubberBand->setPitchScale(pitchScale);
+        m_pRubberBand.setPitchScale(pitchScale);
     }
 
     // RubberBand handles checking for whether the change in timeRatio is a
@@ -67,20 +68,20 @@ void EngineBufferScaleRubberBand::setScaleParameters(double base_rate,
     double timeRatioInverse = base_rate * speed_abs;
     if (timeRatioInverse > 0) {
         //qDebug() << "EngineBufferScaleRubberBand setTimeRatio" << 1 / timeRatioInverse;
-        m_pRubberBand->setTimeRatio(1.0 / timeRatioInverse);
+        m_pRubberBand.setTimeRatio(1.0 / timeRatioInverse);
     }
 
     if (runningEngineVersion() == 2) {
-        if (m_pRubberBand->getInputIncrement() == 0) {
+        if (m_pRubberBand.getInputIncrement() == 0) {
             qWarning() << "EngineBufferScaleRubberBand inputIncrement is 0."
                        << "On RubberBand <=1.8.1 a SIGFPE is imminent despite"
                        << "our workaround. Taking evasive action."
                        << "Please file an issue on https://github.com/mixxxdj/mixxx/issues";
 
             // This is much slower than the minimum seek speed workaround above.
-            while (m_pRubberBand->getInputIncrement() == 0) {
+            while (m_pRubberBand.getInputIncrement() == 0) {
                 timeRatioInverse += 0.001;
-                m_pRubberBand->setTimeRatio(1.0 / timeRatioInverse);
+                m_pRubberBand.setTimeRatio(1.0 / timeRatioInverse);
             }
             speed_abs = timeRatioInverse / base_rate;
             *pTempoRatio = m_bBackwards ? -speed_abs : speed_abs;
@@ -97,7 +98,7 @@ void EngineBufferScaleRubberBand::onOutputSignalChanged() {
     // memory allocations that may block the real-time thread.
     // When is this function actually invoked??
     if (!getOutputSignal().isValid()) {
-        m_pRubberBand.reset();
+        // m_pRubberBand.reset();
         return;
     }
 
@@ -110,7 +111,7 @@ void EngineBufferScaleRubberBand::onOutputSignalChanged() {
         m_bufferPtrs.resize(channelCount);
     }
 
-    m_pRubberBand.reset();
+    // m_pRubberBand.reset();
 
     for (int c = 0; c < channelCount; c++) {
         if (m_buffers[c].size() == MAX_BUFFER_LEN) {
@@ -132,19 +133,19 @@ void EngineBufferScaleRubberBand::onOutputSignalChanged() {
     }
 #endif
 
-    m_pRubberBand = std::make_unique<RubberBandStretcher>(
+    m_pRubberBand.reset(
             getOutputSignal().getSampleRate(),
             getOutputSignal().getChannelCount(),
             rubberbandOptions);
     // Setting the time ratio to a very high value will cause RubberBand
     // to preallocate buffers large enough to (almost certainly)
     // avoid memory reallocations during playback.
-    m_pRubberBand->setTimeRatio(2.0);
-    m_pRubberBand->setTimeRatio(1.0);
+    m_pRubberBand.setTimeRatio(2.0);
+    m_pRubberBand.setTimeRatio(1.0);
 }
 
 void EngineBufferScaleRubberBand::clear() {
-    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand) {
+    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand.isValid()) {
         return;
     }
     reset();
@@ -153,17 +154,21 @@ void EngineBufferScaleRubberBand::clear() {
 SINT EngineBufferScaleRubberBand::retrieveAndDeinterleave(
         CSAMPLE* pBuffer,
         SINT frames) {
-    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand) {
+    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand.isValid()) {
         return 0;
     }
-    const SINT frames_available = m_pRubberBand->available();
+    const SINT frames_available = m_pRubberBand.available();
     // NOTE: If we still need to throw away padding, then we can also
     //       immediately read those frames in addition to the frames we actually
     //       need for the output
     const SINT frames_to_read = math_min(frames_available, frames + m_remainingPaddingInOutput);
-    DEBUG_ASSERT(frames_to_read <= m_buffers[0].size());
-    SINT received_frames = static_cast<SINT>(m_pRubberBand->retrieve(
-            m_bufferPtrs.data(), frames_to_read));
+    DEBUG_ASSERT(frames_to_read <= MAX_BUFFER_LEN);
+    SINT received_frames;
+    {
+        ScopedTimer t(u"RubberBand::retrieve");
+        received_frames = static_cast<SINT>(m_pRubberBand.retrieve(
+                m_bufferPtrs.data(), frames_to_read));
+    }
     SINT frame_offset = 0;
 
     // As explained below in `reset()`, the first time this is called we need to
@@ -215,7 +220,7 @@ SINT EngineBufferScaleRubberBand::retrieveAndDeinterleave(
 void EngineBufferScaleRubberBand::deinterleaveAndProcess(
         const CSAMPLE* pBuffer,
         SINT frames) {
-    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand) {
+    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand.isValid()) {
         return;
     }
     DEBUG_ASSERT(frames <= static_cast<SINT>(m_buffers[0].size()));
@@ -252,17 +257,21 @@ void EngineBufferScaleRubberBand::deinterleaveAndProcess(
     } break;
     }
 
-    m_pRubberBand->process(m_bufferPtrs.data(),
-            frames,
-            false);
+    {
+        ScopedTimer t(u"RubberBand::process");
+        m_pRubberBand.process(m_bufferPtrs.data(),
+                frames,
+                false);
+    }
 }
 
 double EngineBufferScaleRubberBand::scaleBuffer(
         CSAMPLE* pOutputBuffer,
         SINT iOutputBufferSize) {
-    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand) {
+    VERIFY_OR_DEBUG_ASSERT(m_pRubberBand.isValid()) {
         return 0.0;
     }
+    ScopedTimer t(u"EngineBufferScaleRubberBand::scaleBuffer");
     if (m_dBaseRate == 0.0 || m_dTempoRatio == 0.0) {
         SampleUtil::clear(pOutputBuffer, iOutputBufferSize);
         // No actual samples/frames have been read from the
@@ -289,7 +298,7 @@ double EngineBufferScaleRubberBand::scaleBuffer(
         read += getOutputSignal().frames2samples(received_frames);
 
         const SINT next_block_frames_required =
-                static_cast<SINT>(m_pRubberBand->getSamplesRequired());
+                static_cast<SINT>(m_pRubberBand.getSamplesRequired());
         if (remaining_frames > 0 && next_block_frames_required > 0) {
             // The requested setting becomes effective after all previous frames have been processed
             m_effectiveRate = m_dBaseRate * m_dTempoRatio;
@@ -355,39 +364,39 @@ void EngineBufferScaleRubberBand::useEngineFiner(bool enable) {
 // for how these two functions were implemented within librubberband itself
 size_t EngineBufferScaleRubberBand::getPreferredStartPad() const {
 #if RUBBERBANDV3
-    return m_pRubberBand->getPreferredStartPad();
+    return m_pRubberBand.getPreferredStartPad();
 #else
     // `getPreferredStartPad()` returns `window_size / 2`, while with
     // `getLatency()` both time stretching engines return `window_size / 2 /
     // pitch_scale`
     return static_cast<size_t>(std::ceil(
-            m_pRubberBand->getLatency() * m_pRubberBand->getPitchScale()));
+            m_pRubberBand.getLatency() * m_pRubberBand.getPitchScale()));
 #endif
 }
 
 size_t EngineBufferScaleRubberBand::getStartDelay() const {
 #if RUBBERBANDV3
-    return m_pRubberBand->getStartDelay();
+    return m_pRubberBand.getStartDelay();
 #else
     // In newer Rubber Band versions `getLatency()` is a deprecated alias for
     // `getStartDelay()`, so they should behave the same. In the commit linked
     // above the behavior was different for the R3 stretcher, but that was only
     // during the initial betas of Rubberband 3.0 so we shouldn't have to worry
     // about that.
-    return m_pRubberBand->getLatency();
+    return m_pRubberBand.getLatency();
 #endif
 }
 
 int EngineBufferScaleRubberBand::runningEngineVersion() {
 #if RUBBERBANDV3
-    return m_pRubberBand->getEngineVersion();
+    return m_pRubberBand.getEngineVersion();
 #else
     return 2;
 #endif
 }
 
 void EngineBufferScaleRubberBand::reset() {
-    m_pRubberBand->reset();
+    m_pRubberBand.reset();
 
     // As mentioned in the docs (https://breakfastquay.com/rubberband/code-doc/)
     // and FAQ (https://breakfastquay.com/rubberband/integration.html#faqs), you
@@ -404,7 +413,10 @@ void EngineBufferScaleRubberBand::reset() {
     }
     while (remaining_padding > 0) {
         const size_t pad_samples = std::min<size_t>(remaining_padding, block_size);
-        m_pRubberBand->process(m_bufferPtrs.data(), pad_samples, false);
+        {
+            ScopedTimer t(u"RubberBand::process");
+            m_pRubberBand.process(m_bufferPtrs.data(), pad_samples, false);
+        }
 
         remaining_padding -= pad_samples;
     }
@@ -415,3 +427,165 @@ void EngineBufferScaleRubberBand::reset() {
     // `retrieveAndDeinterleave()` first starts producing audio.
     m_remainingPaddingInOutput = static_cast<SINT>(getStartDelay());
 }
+
+RubberBandWorker::RubberBandWorker(mixxx::audio::SampleRate sampleRate,
+        const RubberBandStretcher::Options& opt)
+        : QThread(),
+          m_rubberBand(std::make_unique<RubberBandStretcher>(
+                  sampleRate, mixxx::audio::ChannelCount::stereo(), opt)) {
+}
+
+void RubberBandWorker::process(const float* const* input, size_t samples, bool final) {
+    auto locker = lockMutex(&m_waitLock);
+    m_job.input = input;
+    m_job.samples = samples;
+    m_job.final = final;
+    m_ready = false;
+    m_waitCondition.wakeOne();
+}
+
+void RubberBandWorker::waitReady() {
+    auto locker = lockMutex(&m_waitLock);
+    while (!m_ready) {
+        m_waitCondition.wait(&m_waitLock);
+    }
+}
+
+void RubberBandWorker::stop() {
+    requestInterruption();
+    m_waitCondition.wakeOne();
+    wait();
+}
+void RubberBandWorker::run() {
+    auto locker = lockMutex(&m_waitLock);
+    while (!isInterruptionRequested()) {
+        if (!m_ready) {
+            m_rubberBand->process(m_job.input, m_job.samples, m_job.final);
+            m_ready = true;
+            m_waitCondition.wakeOne();
+        }
+        m_waitCondition.wait(&m_waitLock);
+    }
+    quit();
+}
+
+int RubberBandManager::getEngineVersion() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getEngineVersion();
+}
+void RubberBandManager::setTimeRatio(double ratio) {
+    for (auto& worker : m_workers) {
+        worker->m_rubberBand->setTimeRatio(ratio);
+    }
+}
+size_t RubberBandManager::getSamplesRequired() const {
+    size_t require = 0;
+    for (auto& worker : m_workers) {
+        require = qMax(require, worker->m_rubberBand->getSamplesRequired());
+    }
+    return require;
+}
+int RubberBandManager::available() const {
+    int available = std::numeric_limits<int>::max();
+    for (auto& worker : m_workers) {
+        available = qMin(available, worker->m_rubberBand->available());
+    }
+    return available;
+}
+size_t RubberBandManager::retrieve(float* const* output, size_t samples) const {
+    if (m_workers.size() == 1) {
+        return m_workers[0]->m_rubberBand->retrieve(output, samples);
+    } else {
+        size_t ret = 0;
+        for (const auto& worker : m_workers) {
+            size_t thisRet = worker->m_rubberBand->retrieve(output, samples);
+            DEBUG_ASSERT(!ret || thisRet == ret);
+            ret = qMax(thisRet, ret);
+            output += mixxx::audio::ChannelCount::stereo();
+        }
+        return ret;
+    }
+}
+size_t RubberBandManager::getInputIncrement() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getInputIncrement();
+}
+size_t RubberBandManager::getLatency() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getLatency();
+}
+double RubberBandManager::getPitchScale() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getPitchScale();
+}
+size_t RubberBandManager::getPreferredStartPad() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getPreferredStartPad();
+}
+size_t RubberBandManager::getStartDelay() const {
+    VERIFY_OR_DEBUG_ASSERT(isValid()) {
+        return -1;
+    }
+    return m_workers[0]->m_rubberBand->getStartDelay();
+}
+void RubberBandManager::process(const float* const* input, size_t samples, bool final) {
+    if (m_workers.size() == 1) {
+        return m_workers[0]->m_rubberBand->process(input, samples, final);
+    } else {
+        for (auto& worker : m_workers) {
+            worker->process(input, samples, final);
+            input += mixxx::audio::ChannelCount::stereo();
+        }
+        for (auto& worker : m_workers) {
+            worker->waitReady();
+        }
+    }
+}
+void RubberBandManager::reset() {
+    for (auto& worker : m_workers) {
+        worker->m_rubberBand->reset();
+    }
+}
+void RubberBandManager::reset(mixxx::audio::SampleRate sampleRate,
+        mixxx::audio::ChannelCount chCount,
+        const RubberBandStretcher::Options& opt) {
+    DEBUG_ASSERT(0 == chCount % 2);
+    for (auto& worker : m_workers) {
+        worker->stop();
+    }
+    m_workers.clear();
+
+    for (int c = 0; c < chCount / 2; c++) {
+        m_workers.emplace_back(std::make_unique<RubberBandWorker>(sampleRate, opt));
+    }
+
+    if (m_workers.size() > 1) {
+        for (auto& worker : m_workers) {
+            worker->start(QThread::HighPriority);
+        }
+    }
+}
+void RubberBandManager::setPitchScale(double scale) {
+    for (auto& worker : m_workers) {
+        worker->m_rubberBand->setPitchScale(scale);
+    }
+}
+
+RubberBandManager::~RubberBandManager() {
+    for (auto& worker : m_workers) {
+        worker->stop();
+    }
+}
+bool RubberBandManager::isValid() const {
+    return m_workers.size();
+}