tracktion_LockFreeMultiThreadedNodePlayer.h

Go to the documentation of this file.

 /*
    ,--.                     ,--.     ,--.  ,--.
  ,-'  '-.,--.--.,--,--.,---.|  |,-.,-'  '-.`--' ,---. ,--,--,      Copyright 2024
  '-.  .-'|  .--' ,-.  | .--'|     /'-.  .-',--.| .-. ||      \   Tracktion Software
    |  |  |  |  \ '-'  \ `--.|  \  \  |  |  |  |' '-' '|  ||  |       Corporation
    `---' `--'   `--`--'`---'`--'`--' `---' `--' `---' `--''--'    www.tracktion.com
 
    Tracktion Engine uses a GPL/commercial licence - see LICENCE.md for details.
*/
 
#pragma once
 
#ifdef _MSC_VER
 #pragma warning (push)
 #pragma warning (disable: 4127)
#endif
 
#ifdef _MSC_VER
 #pragma warning (pop)
#endif
 
namespace tracktion { inline namespace graph
{
 
//==============================================================================
//==============================================================================
class LockFreeMultiThreadedNodePlayer
{
private:
    //==============================================================================
    template<typename Type>
    class LockFreeFifo
    {
    public:
        LockFreeFifo (int capacity)
            : fifo (std::make_unique<rigtorp::MPMCQueue<Type>> (static_cast<size_t> (capacity)))
        {
        }
 
        bool try_enqueue (Type&& item)      { return fifo->try_push (std::move (item)); }
        bool try_dequeue (Type& item)       { return fifo->try_pop (item); }
 
    private:
        std::unique_ptr<rigtorp::MPMCQueue<Type>> fifo;
    };
 
    struct PlaybackNode
    {
        PlaybackNode (Node& n)
            : node (n), numInputs (node.getDirectInputNodes().size())
        {}
 
        Node& node;
        const size_t numInputs;
        std::vector<Node*> outputs;
        std::atomic<size_t> numInputsToBeProcessed { 0 };
        std::atomic<bool> hasBeenQueued { true };
       #if JUCE_DEBUG
        std::atomic<bool> hasBeenDequeued { false };
       #endif
    };
 
    struct PreparedNode
    {
        std::unique_ptr<NodeGraph> graph;
        std::vector<std::unique_ptr<PlaybackNode>> playbackNodes;
        std::unique_ptr<LockFreeFifo<Node*>> nodesReadyToBeProcessed;
        std::unique_ptr<AudioBufferPool> audioBufferPool;
    };
 
public:
    //==============================================================================
    struct ThreadPool
    {
        ThreadPool (LockFreeMultiThreadedNodePlayer& p)
            : player (p)
        {
        }
 
        virtual ~ThreadPool() = default;
 
        virtual void createThreads (size_t numThreads, juce::AudioWorkgroup) = 0;
 
        virtual void clearThreads() = 0;
 
        virtual void signalOne() = 0;
 
        virtual void signal (int numToSignal) = 0;
 
        virtual void signalAll() = 0;
 
        virtual void waitForFinalNode() = 0;
 
        //==============================================================================
        void signalShouldExit()
        {
            threadsShouldExit = true;
            signalAll();
        }
 
        void resetExitSignal()
        {
            threadsShouldExit = false;
        }
 
        bool shouldExit() const
        {
            return threadsShouldExit.load (std::memory_order_acquire);
        }
 
        bool shouldWait()
        {
            if (shouldExit())
                return false;
 
            return player.numNodesQueued == 0;
        }
 
        bool isFinalNodeReady()
        {
            assert (currentPreparedNode);
 
            if (! currentPreparedNode.load()->graph)
                return true;
 
            return currentPreparedNode.load()->graph->rootNode->hasProcessed();
        }
 
        bool process()
        {
            if (auto cpn = currentPreparedNode.load())
                return player.processNextFreeNode (*cpn);
 
            return false;
        }
 
        void setCurrentNode (LockFreeMultiThreadedNodePlayer::PreparedNode* nodeInUse)
        {
            currentPreparedNode = nodeInUse;
 
            // If a nullptr is being set, you must stop the threads first.
            assert (nodeInUse || shouldExit());
        }
 
        LockFreeMultiThreadedNodePlayer& player;
 
    private:
        std::atomic<bool> threadsShouldExit { false };
        std::atomic<LockFreeMultiThreadedNodePlayer::PreparedNode*> currentPreparedNode { nullptr };
    };
 
    //==============================================================================
    LockFreeMultiThreadedNodePlayer();
 
    using ThreadPoolCreator = std::function<std::unique_ptr<ThreadPool> (LockFreeMultiThreadedNodePlayer&)>;
 
    LockFreeMultiThreadedNodePlayer (ThreadPoolCreator, juce::AudioWorkgroup = {});
 
    ~LockFreeMultiThreadedNodePlayer();
 
    //==============================================================================
    void setNumThreads (size_t);
 
    void setNode (std::unique_ptr<Node>);
 
    void setNode (std::unique_ptr<Node> newNode, double sampleRateToUse, int blockSizeToUse);
 
    void prepareToPlay (double sampleRateToUse, int blockSizeToUse);
 
    Node* getNode()
    {
        return rootNode;
    }
 
    int process (const Node::ProcessContext&);
 
    void clearNode();
 
    double getSampleRate() const
    {
        return sampleRate.load (std::memory_order_acquire);
    }
 
    int getBlockSize() const
    {
        return blockSize.load (std::memory_order_acquire);;
    }
 
    //==============================================================================
    void enablePooledMemoryAllocations (bool);
 
    /* @internal. */
    void enableNodeMemorySharing (bool shouldBeEnabled);
 
private:
    //==============================================================================
    std::atomic<size_t> numThreadsToUse { std::max ((size_t) 0, (size_t) std::thread::hardware_concurrency() - 1) };
    juce::Range<int64_t> referenceSampleRange;
    choc::buffer::FrameCount numSamplesToProcess = 0;
    std::atomic<bool> threadsShouldExit { false }, useMemoryPool { false };
 
    std::unique_ptr<ThreadPool> threadPool;
    juce::AudioWorkgroup audioWorkgroup;
 
    LockFreeObject<PreparedNode> preparedNodeObject;
    Node* rootNode = nullptr;
    NodeGraph* lastGraphPosted = nullptr;
    AudioBufferPool* lastAudioBufferPoolPosted = nullptr;
 
    std::atomic<size_t> numNodesQueued { 0 };
 
    //==============================================================================
    std::atomic<double> sampleRate { 44100.0 };
    std::atomic<int> blockSize { 512 };
    bool nodeMemorySharingEnabled = false;
 
    //==============================================================================
    std::unique_ptr<NodeGraph> prepareToPlay (std::unique_ptr<Node>, NodeGraph* oldGraph,
                                              double sampleRateToUse, int blockSizeToUse,
                                              bool useCurrentAudioBufferPool);
 
    //==============================================================================
    void clearThreads();
    void createThreads();
    void pause();
 
    //==============================================================================
    void postNewGraph (std::unique_ptr<NodeGraph>);
 
    //==============================================================================
    static void buildNodesOutputLists (PreparedNode&);
    void resetProcessQueue (PreparedNode&);
    Node* updateProcessQueueForNode (PreparedNode&, Node&);
    void processNode (PreparedNode&, Node&);
 
    //==============================================================================
    bool processNextFreeNode (PreparedNode&);
};
 
}}