Pipeline_example.cpp

//------------------------------------------------------------------------------
// DecodePipelineData

using DecodePipelineCallback = std::function<void(std::shared_ptr<DecodedFrame>)>;

struct DecodePipelineData
{
    // Inputs
    DecodePipelineCallback Callback;
    std::shared_ptr<FrameInfo> Input;
    int PerspectiveIndex = -1;

    // Outputs
    std::shared_ptr<DecodedFrame> Output;
};


//------------------------------------------------------------------------------
// DecodePipelineElement

/*
    Processing pipeline:
    (1) Decompress and generate mesh
    (2) Decompress texture
*/

class DecodePipelineElement
{
public:
    void Initialize(
        std::shared_ptr<DecodePipelineElement> next_element,
        std::string element_name);
    ~DecodePipelineElement()
    {
        Shutdown();
    }
    void Shutdown();

    // Returns false if the queue overflowed
    void Process(std::shared_ptr<DecodePipelineData> data);

protected:
    std::shared_ptr<DecodePipelineElement> NextElement;
    std::string ElementName;
    DecodePipelineCallback Callback;

    core::WorkerQueue Worker;

    virtual bool Run(std::shared_ptr<DecodePipelineData> data) = 0;
};


//------------------------------------------------------------------------------
// DecodePipelineElement

void DecodePipelineElement::Initialize(
    std::shared_ptr<DecodePipelineElement> next_element,
    std::string element_name)
{
    NextElement = next_element;
    ElementName = element_name;

    Worker.Initialize(kMaxQueuedDecodes);
}

void DecodePipelineElement::Shutdown()
{
    Worker.Shutdown();
}

void DecodePipelineElement::Process(std::shared_ptr<DecodePipelineData> data)
{
    // Called from the worker thread of the previous element:

    const bool pushed = Worker.SubmitWork([this, data]()
    {
        const bool success = Run(data);

        // If operation failed:
        if (!success) {
            spdlog::error("Operation failed for stage {}: Dropped frame {}",
                ElementName, data->Input->BatchInfo->FrameNumber);
            return;
        }

        // If there is another element in the pipe:
        if (NextElement) {
            NextElement->Process(data);
        } else {
            data->Callback(data->Output);
        }
    });

    // If the queue overflowed to the next element:
    if (!pushed) {
        spdlog::error("Computer too slow for stage {}: Dropped frame {}",
            ElementName, data->Input->BatchInfo->FrameNumber);
    }
}


//------------------------------------------------------------------------------
// Element State

struct VideoDecoderElement : public DecodePipelineElement
{
    unsigned Width = 0;

    std::unique_ptr<mfx::VideoDecoder> IntelDecoder;

    BackreferenceChecker BackrefChecker;

    bool Run(std::shared_ptr<DecodePipelineData> data) override;
};

struct MeshDecompressorElement : public DecodePipelineElement
{
    int DepthWidth = 0;

    // Depth decompressor
    std::unique_ptr<lossless::DepthCompressor> LosslessDepth;
    std::unique_ptr<lossy::DepthCompressor> LossyDepth;

    // Mesh from depth
    std::unique_ptr<DepthMesher> Mesher;

    bool Run(std::shared_ptr<DecodePipelineData> data) override;
};