tracktion_MultiThreadedNodePlayer.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
 
namespace tracktion { inline namespace graph
{
 
class MultiThreadedNodePlayer
{
public:
    //==============================================================================
    MultiThreadedNodePlayer();
 
    ~MultiThreadedNodePlayer();
 
    //==============================================================================
    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()
    {
        if (auto cpn = currentPreparedNode.load (std::memory_order_acquire);
            cpn != nullptr && cpn->graph != nullptr)
           return cpn->graph->rootNode.get();
 
        return nullptr;
    }
 
    int process (const Node::ProcessContext&);
 
    void clearNode();
 
    double getSampleRate() const
    {
        return sampleRate.load (std::memory_order_acquire);
    }
 
private:
    //==============================================================================
    std::atomic<size_t> numThreadsToUse { std::max ((size_t) 0, (size_t) std::thread::hardware_concurrency() - 1) };
    choc::buffer::FrameCount numSamplesToProcess;
    juce::Range<int64_t> referenceSampleRange;
    std::atomic<bool> threadsShouldExit { false };
 
    class ThreadPool;
    std::unique_ptr<ThreadPool> threadPool;
 
    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;
        choc::fifo::MultipleReaderMultipleWriterFIFO<Node*> nodesReadyToBeProcessed;
    };
 
    RealTimeSpinLock preparedNodeMutex;
    std::unique_ptr<PreparedNode> preparedNode;
    std::atomic<PreparedNode*> currentPreparedNode { nullptr };
 
    std::atomic<size_t> numNodesQueued { 0 };
    RealTimeSpinLock clearNodesLock;
 
    //==============================================================================
    std::atomic<double> sampleRate { 0.0 };
    int blockSize = 0;
 
    //==============================================================================
    std::unique_ptr<NodeGraph> prepareToPlay (std::unique_ptr<Node>, NodeGraph* oldGraph, double sampleRateToUse, int blockSizeToUse);
 
    //==============================================================================
    void clearThreads();
    void createThreads();
    static void pause();
 
    //==============================================================================
    void setNewGraph (std::unique_ptr<NodeGraph>);
 
    //==============================================================================
    static void buildNodesOutputLists (std::vector<Node*>&, std::vector<std::unique_ptr<PlaybackNode>>&);
    void resetProcessQueue();
    Node* updateProcessQueueForNode (Node&);
    void processNode (Node&);
 
    //==============================================================================
    bool processNextFreeNode();
};
 
}}