Simplified helper, recycle buffers across tracks

src/libs/audio/impl/utils/PcmDecodeStreamer.cpp

@@ -21,26 +21,25 @@
 
 #include <boost/asio/post.hpp>
 
-#include "audio/Exception.hpp"
 #include "core/ILogger.hpp"
 
+#include "audio/Exception.hpp"
 #include "audio/IAudioOutput.hpp"
 #include "audio/IPcmDecoder.hpp"
 
 namespace lms::audio::utils
 {
-    std::shared_ptr<IPcmDecodeStreamer> createPcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& parameters)
+    std::unique_ptr<IPcmDecodeStreamer> createPcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& parameters)
     {
-        return std::make_shared<PcmDecodeStreamer>(ioContext, parameters);
+        return std::make_unique<PcmDecodeStreamer>(ioContext, parameters);
     }
 
     PcmDecodeStreamer::PcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& parameters)
         : _ioContext{ ioContext }
         , _strand{ _ioContext }
         , _outputStream{ parameters.outputStream }
-        , _pcmDecoder{ audio::createPcmDecoder(parameters.file, parameters.offset, parameters.pcmParameters) }
     {
-        prepareBuffers();
+        prepareBuffers(parameters.bufferCount, parameters.bufferDuration);
     }
 
     PcmDecodeStreamer::~PcmDecodeStreamer()
@@ -48,42 +47,51 @@ namespace lms::audio::utils
         assert(!isWritePending());
     }
 
-    void PcmDecodeStreamer::start(DecodeCompleteCallback cb)
+    void PcmDecodeStreamer::start(const std::filesystem::path& path, std::chrono::microseconds offset, DecodeCompleteCallback cb)
     {
         assert(cb);
-        assert(!_decodeCompleteCallback);
+        assert(isComplete()); // previous job must be finished or cancelled
+
+        _pcmDecoder = audio::createPcmDecoder(path, offset, getPcmParameters()); // may throw
+        _aborted = false;
+        _eofReached = false;
 
         _ioContext.get_executor().on_work_started();
         _decodeCompleteCallback = std::move(cb);
 
-        boost::asio::post(_strand, [self = shared_from_this()] {
+        boost::asio::post(_strand, [this] {
-            self->decodeSome();
+            decodeSome();
         });
     }
 
     void PcmDecodeStreamer::abort()
     {
-        boost::asio::post(_strand, [self = shared_from_this()] {
+        if (isComplete())
-            LMS_LOG(AUDIO, DEBUG, "Processing abort");
+            return;
-            self->_aborted = true;
-            self->_outputStream.flush();
 
-            if (!self->isWritePending())
+        boost::asio::post(_strand, [this] {
-                self->notifyDecodeComplete();
+            LMS_LOG(AUDIO, DEBUG, "Processing abort");
+            _aborted = true;
+            _outputStream.flush();
+
+            if (!isWritePending())
+                notifyDecodeComplete();
         });
     }
 
+    bool PcmDecodeStreamer::isComplete() const
+    {
+        return !_pcmDecoder;
+    }
+
     const audio::PcmParameters& PcmDecodeStreamer::getPcmParameters() const
     {
-        return _pcmDecoder->getParameters();
+        return _outputStream.getParameters();
     }
 
-    void PcmDecodeStreamer::prepareBuffers()
+    void PcmDecodeStreamer::prepareBuffers(std::size_t bufferCount, std::chrono::microseconds bufferDuration)
     {
-        constexpr std::chrono::milliseconds bufferDuration{ 100 };
+        const std::size_t sampleCountPerBuffer{ static_cast<std::size_t>(std::chrono::duration_cast<std::chrono::microseconds>(bufferDuration).count() * getPcmParameters().sampleRate / std::chrono::microseconds::period::den) };
-
-        const audio::PcmParameters& pcmParams{ getPcmParameters() };
-        std::size_t sampleCountPerBuffer{ static_cast<std::size_t>(std::chrono::duration_cast<std::chrono::microseconds>(bufferDuration).count() * pcmParams.sampleRate / std::chrono::microseconds::period::den) };
         const std::size_t bufferSize{ sampleCountToByteCount(sampleCountPerBuffer) };
 
         _buffers.resize(bufferCount);
@@ -127,11 +135,11 @@ namespace lms::audio::utils
             bufferDesc.isWritePending = true;
             buffer = { buffer.data(), sampleCountToByteCount(sampleCount) };
 
-            _outputStream.asyncWrite(buffer, [self = shared_from_this(), bufferIndex] {
+            _outputStream.asyncWrite(buffer, [this, bufferIndex] {
-                boost::asio::post(self->_strand, [self, bufferIndex] { self->onBufferWriteComplete(bufferIndex); });
+                boost::asio::post(_strand, [this, bufferIndex] { onBufferWriteComplete(bufferIndex); });
             });
 
-            boost::asio::post(_strand, [self = shared_from_this()] { self->decodeSome(); });
+            boost::asio::post(_strand, [this] { decodeSome(); });
         }
     }
 
@@ -185,10 +193,12 @@ namespace lms::audio::utils
 
     void PcmDecodeStreamer::notifyDecodeComplete()
     {
-        boost::asio::post(_ioContext, [self = shared_from_this(), cb = std::move(_decodeCompleteCallback)] {
+        LMS_LOG(AUDIO, DEBUG, "Decode complete notification");
-            LMS_LOG(AUDIO, DEBUG, "Decode complete notification");
+
-            cb(self->_aborted);
+        boost::asio::post(_ioContext, [this, cb = std::move(_decodeCompleteCallback)] {
-            self->_ioContext.get_executor().on_work_finished();
+            _pcmDecoder.reset();
+            cb(_aborted);
+            _ioContext.get_executor().on_work_finished();
         });
     }
 

src/libs/audio/impl/utils/PcmDecodeStreamer.hpp

@@ -19,14 +19,12 @@
 
 #pragma once
 
-#include <filesystem>
+#include <cstddef>
-#include <memory>
 #include <vector>
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/io_context_strand.hpp>
 
-#include "audio/PcmTypes.hpp"
 #include "audio/utils/IPcmDecodeStreamer.hpp"
 
 namespace lms::audio
@@ -37,7 +35,7 @@ namespace lms::audio
 
 namespace lms::audio::utils
 {
-    class PcmDecodeStreamer : public IPcmDecodeStreamer, public std::enable_shared_from_this<PcmDecodeStreamer>
+    class PcmDecodeStreamer : public IPcmDecodeStreamer
     {
     public:
         PcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& parameters);
@@ -47,12 +45,13 @@ namespace lms::audio::utils
         PcmDecodeStreamer& operator=(const PcmDecodeStreamer&) = delete;
 
     private:
-        void start(DecodeCompleteCallback cb) override;
+        void start(const std::filesystem::path& path, std::chrono::microseconds offset, DecodeCompleteCallback cb) override;
-        void abort() override; // will call DecodeCompleteCallback once done
+        void abort() override;
+        bool isComplete() const override;
 
         const audio::PcmParameters& getPcmParameters() const;
 
-        void prepareBuffers();
+        void prepareBuffers(std::size_t bufferCount, std::chrono::microseconds bufferDuration);
         bool isWritePending() const;
         void decodeSome();
         std::size_t readSamples(std::span<std::byte> buffer);
@@ -72,7 +71,6 @@ namespace lms::audio::utils
         audio::IAudioOutputStream& _outputStream;
         std::unique_ptr<audio::IPcmDecoder> _pcmDecoder;
 
-        static constexpr std::size_t bufferCount{ 4 };
         std::vector<BufferDesc> _buffers;
         std::size_t _nextBufferIndex{};
         bool _eofReached{};

src/libs/audio/include/audio/utils/IPcmDecodeStreamer.hpp

@@ -33,22 +33,26 @@ namespace lms::audio
 
 namespace lms::audio::utils
 {
+    // helper class to decode files to PCM samples, fed into the provided output stream
     class IPcmDecodeStreamer
     {
     public:
         virtual ~IPcmDecodeStreamer() = default;
 
         using DecodeCompleteCallback = std::function<void(bool aborted)>;
-        virtual void start(DecodeCompleteCallback cb) = 0;
+
-        virtual void abort() = 0; // will call DecodeCompleteCallback once done
+        // DecodeCompleteCallback is fired once the file has been fully decoded (but still buffered in output)
+        // You can start a new file only if the previous one is finished (i.e. once the callback is fired)
+        virtual void start(const std::filesystem::path& path, std::chrono::microseconds offset, DecodeCompleteCallback cb) = 0;
+        virtual void abort() = 0; // will call DecodeCompleteCallback once aborted
+        virtual bool isComplete() const = 0;
     };
 
     struct PcmDecodeStreamerParameters
     {
         audio::IAudioOutputStream& outputStream;
-        std::filesystem::path file;
+        std::size_t bufferCount;
-        std::chrono::microseconds offset;
+        std::chrono::milliseconds bufferDuration;
-        audio::PcmParameters pcmParameters;
     };
-    std::shared_ptr<IPcmDecodeStreamer> createPcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& params);
+    std::unique_ptr<IPcmDecodeStreamer> createPcmDecodeStreamer(boost::asio::io_context& ioContext, const PcmDecodeStreamerParameters& parameters);
 } // namespace lms::audio::utils

src/libs/services/jukebox/impl/JukeboxService.cpp

@@ -20,7 +20,9 @@
 #include "JukeboxService.hpp"
 
 #include <chrono>
+#include <cstdlib>
 #include <format>
+#include <thread>
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/post.hpp>
@@ -59,9 +61,6 @@ namespace lms::jukebox
         if (_decoder)
             _decoder->abort();
 
-        if (_outputStream)
-            _outputStream->flush();
-
         _ioContextRunner.wait();
         LMS_LOG(JUKEBOX, INFO, "Service stopped!");
     }
@@ -72,7 +71,6 @@ namespace lms::jukebox
 
         std::unique_lock lock{ _mutex };
 
-        deleteDecoder();
         if (trackIndex >= _tracks.size())
         {
             LMS_LOG(JUKEBOX, INFO, "Requested track index out of bound: stopping");
@@ -80,12 +78,15 @@ namespace lms::jukebox
             return;
         }
 
-        if (createDecoder(trackIndex, offset))
+        if (!_outputStream)
+            return;
+
+        abortDecoder();
+        if (startDecoder(trackIndex, offset))
         {
             _currentTrackIndex = trackIndex;
             _currentTrackPlaybackTimeOffset = _outputStream->getPlaybackTime();
             _currentTrackStartTimeOffset = offset;
-            startDecoder();
             _outputStream->resume();
         }
         // TODO if failure, switch to the next song?
@@ -202,10 +203,20 @@ namespace lms::jukebox
 
     void JukeboxService::onStreamReady()
     {
+        audio::utils::PcmDecodeStreamerParameters params{
+            .outputStream = *_outputStream,
+            .bufferCount = 50,
+            .bufferDuration = std::chrono::milliseconds{ 1000 },
+        };
+
+        _decoder = audio::utils::createPcmDecodeStreamer(_ioContext, params);
     }
 
-    bool JukeboxService::createDecoder(std::size_t trackIndex, std::chrono::microseconds offset)
+    bool JukeboxService::startDecoder(std::size_t trackIndex, std::chrono::microseconds offset)
     {
+        if (!_decoder)
+            return false;
+
         std::filesystem::path trackPath;
         {
             auto& session{ _db.getTLSSession() };
@@ -223,48 +234,42 @@ namespace lms::jukebox
 
         try
         {
-            audio::utils::PcmDecodeStreamerParameters params{
+            _decoder->start(trackPath, offset, [this](bool aborted) {
-                .outputStream = *_outputStream,
+                onDecodeFinished(aborted);
-                .file = trackPath,
+            });
-                .offset = offset,
-                .pcmParameters = _pcmParams,
-            };
-
-            _decoder = audio::utils::createPcmDecodeStreamer(_ioContext, params);
         }
         catch (const audio::Exception& e)
         {
-            LMS_LOG(JUKEBOX, ERROR, "Failed to create PCM decoder for track " << trackPath);
+            LMS_LOG(JUKEBOX, ERROR, "Failed to start PCM decoder for track " << trackPath);
             return false;
         }
 
         return true;
     }
 
-    void JukeboxService::deleteDecoder()
+    void JukeboxService::abortDecoder()
     {
+        // Must not be called from within owned io_context
         if (_decoder)
         {
             _decoder->abort();
-            _decoder.reset();
-        }
-    }
 
-    void JukeboxService::startDecoder()
+            // Should be hopefully short since flushing/aborting
-    {
+            while (!_decoder->isComplete())
-        _decoder->start([this](bool aborted) {
+                std::this_thread::yield(); // TODO: execute some io_context stuff?
-            onDecodeFinished(aborted);
+        }
-        });
     }
 
     void JukeboxService::onDecodeFinished(bool aborted)
     {
         if (aborted)
-            return; // already setup to play next song
+            return; // already setup to play next song, if needed
 
         std::unique_lock lock{ _mutex };
 
-        _decoder.reset();
+        _currentTrackPlaybackTimeOffset = {};
+        _currentTrackStartTimeOffset = {};
+
         if (!_currentTrackIndex)
         {
             _outputStream->pause();
@@ -274,15 +279,19 @@ namespace lms::jukebox
         if (++(*_currentTrackIndex) >= _tracks.size())
         {
             _currentTrackIndex.reset();
+            // let the output stream run out of data
             return;
         }
 
-        if (createDecoder(*_currentTrackIndex))
+        if (startDecoder(*_currentTrackIndex))
         {
-            startDecoder();
             _currentTrackPlaybackTimeOffset = _outputStream->getPlaybackTime() + _outputStream->getLatency();
             _currentTrackStartTimeOffset = {};
         }
+        else
+        {
+            _currentTrackIndex.reset();
+        }
         // TODO if failure, switch to the next song?
     }
 } // namespace lms::jukebox

src/libs/services/jukebox/impl/JukeboxService.hpp

@@ -64,9 +64,8 @@ namespace lms::jukebox
         void onContextReady();
         void onStreamReady();
 
-        bool createDecoder(std::size_t trackIndex, std::chrono::microseconds offset = {});
+        bool startDecoder(std::size_t trackIndex, std::chrono::microseconds offset = {});
-        void deleteDecoder();
+        void abortDecoder();
-        void startDecoder();
         void onDecodeFinished(bool aborted);
 
         // TODO: make configurable or use detected output params
@@ -91,6 +90,6 @@ namespace lms::jukebox
         std::unique_ptr<audio::IAudioOutputContext> _outputContext;
         std::unique_ptr<audio::IAudioOutputStream> _outputStream;
 
-        std::shared_ptr<audio::utils::IPcmDecodeStreamer> _decoder;
+        std::unique_ptr<audio::utils::IPcmDecodeStreamer> _decoder;
     };
 } // namespace lms::jukebox

src/tools/audioplay/LmsAudioPlay.cpp

@@ -72,13 +72,12 @@ namespace lms
         {
             audio::utils::PcmDecodeStreamerParameters params{
                 .outputStream = *_outputStream,
-                .file = _filePath,
+                .bufferCount = 2,
-                .offset = _offset,
+                .bufferDuration = std::chrono::milliseconds{ 100 },
-                .pcmParameters = _pcmParams,
             };
 
             _fileStreamer = audio::utils::createPcmDecodeStreamer(_ioContext, params);
-            _fileStreamer->start([this](bool aborted) {
+            _fileStreamer->start(_filePath, _offset, [this](bool aborted) {
                 if (aborted)
                     std::cerr << "Playback aborted!" << std::endl;
 
@@ -98,7 +97,7 @@ namespace lms
             });
 
             // Gives some time for the buffer to fill in
-            _playTimer.expires_from_now(std::chrono::milliseconds{ 50 });
+            _playTimer.expires_after(std::chrono::milliseconds{ 50 });
             _playTimer.async_wait([this](const boost::system::error_code& ec) {
                 if (ec)
                     return;