ApraPipes Pipeline Control Flow: How play, pause, step, and stop Work

ApraPipes implements a centralized PipeLine controller that propagates play, pause, step, and stop commands to source modules, which manage execution via internal command queues and state flags.

The ApraPipes library provides a C++ framework for building video and data processing pipelines using a modular architecture. Understanding the ApraPipes pipeline control flow is essential for managing real-time media processing, debugging frame-level operations, and ensuring clean shutdown sequences.

The Centralized Controller Architecture

The PipeLine class in base/include/PipeLine.h serves as the sole authority for control-flow operations. It maintains a linear list of modules but specifically targets source modules—the only components that generate frames—for play, pause, and step commands.

Control commands follow a top-down propagation pattern:

1. The pipeline receives a user command (play, pause, step, or stop)
2. The pipeline iterates through its module list
3. Commands execute only on modules with Module::SOURCE nature
4. Source modules translate these into internal state changes or command queue entries

This design ensures that transform and sink modules remain passive consumers while sources control the temporal flow of data.

Play and Pause: Starting and Halting Frame Generation

Pipeline-Level Control

In base/src/PipeLine.cpp, the play() and pause() methods toggle the pipeline's internal mPlay flag and forward the state to every source module:

void PipeLine::play()
{
    for (auto i = modules.begin(); i != modules.end(); i++)
    {
        if (i->get()->getNature() == Module::SOURCE)
        {
            i->get()->play(true);
        }
    }
    if (controlModule != nullptr) controlModule->play(true);
    mPlay = true;
}

The pause() method follows an identical pattern but passes false to each source and sets mPlay = false.

Module-Level Execution

Each source module receives the command through Module::play(bool) defined in base/src/Module.cpp. This method forwards to an overloaded version that creates a PlayPauseCommand:

bool Module::play(float speed, bool direction)
{
    if (!mRunning)
        return handlePausePlay(speed, direction);
    
    PlayPauseCommand ppCmd(speed, direction);
    return queuePlayPauseCommand(ppCmd);
}

If the module is not yet running (for example, when using run_all_threaded_withpause()), the call immediately executes handlePausePlay(), which updates the internal mPlay flag and speed. For active threads, the command enters a queue for asynchronous processing.

Step: Single-Frame Execution While Paused

The step command enables frame-accurate debugging by triggering exactly one produce() cycle while the pipeline remains paused. In PipeLine::step(), the pipeline verifies that mPlay is false, then calls queueStep() on every source module:

void PipeLine::step()
{
    if (mPlay) return;
    for (auto i = modules.begin(); i != modules.end(); i++)
    {
        if (i->get()->getNature() == Module::SOURCE)
        {
            i->get()->queueStep();
        }
    }
}

At the module level, queueStep() creates a StepCommand and enqueues it via queueCommand(). When the module's processing loop encounters this command in processSourceQue(), it executes produce() once and returns, yielding a single frame without resuming continuous playback.

Stop: Terminating the Pipeline

The stop() method initiates a graceful shutdown by transitioning the pipeline state to PL_STOPPING and invoking stop() on every source module plus the control module:

void PipeLine::stop()
{
    if (myStatus >= PL_STOPPING) return;
    myStatus = PL_STOPPING;
    for (auto i = modules.begin(); i != modules.end(); i++)
    {
        if (i->get()->getNature() == Module::SOURCE) i->get()->stop();
    }
    if (controlModule != nullptr) controlModule->stop();
}

For each source, Module::stop() terminates the internal processing thread and clears command queues. If the source was paused, the stop command ensures the mPlay flag is cleared through handlePausePlay().

Implementation Details in Source Code

The control flow relies on specific source files that define the interaction between the pipeline and its modules:

• base/include/PipeLine.h – Declares run_all_threaded_withpause(), play(), pause(), step(), and stop()
• base/src/PipeLine.cpp – Implements the iteration logic and state management
• base/include/Module.h – Defines the module interface including play(), queueStep(), and stop()
• base/src/Module.cpp – Contains handlePausePlay(), which updates mSpeed and the mPlay boolean flag

The handlePausePlay() method serves as the final destination for play/pause state changes:

bool Module::handlePausePlay(bool play)
{
    mPlay = play;
    notifyPlay(mPlay);
    mSpeed = mPlay ? 1 : 0;
    return true;
}

Practical Usage Example

The following pattern demonstrates the typical lifecycle of an ApraPipes pipeline using all four control states:

#include "PipeLine.h"
#include "Module.h"
#include <memory>

int main()
{
    auto pipeline = std::make_shared<PipeLine>("example");

    auto src   = std::make_shared<SourceModule>("camera", ModuleProps());
    auto xform = std::make_shared<TransformModule>("blur", ModuleProps());
    auto sink  = std::make_shared<SinkModule>("display", ModuleProps());

    pipeline->appendModule(src);
    pipeline->appendModule(xform);
    pipeline->appendModule(sink);
    pipeline->init();

    pipeline->run_all_threaded_withpause();

    pipeline->play();

    std::this_thread::sleep_for(std::chrono::seconds(3));
    pipeline->pause();

    pipeline->step();

    pipeline->stop();
    pipeline->wait_for_all();
}

This example starts the pipeline in a paused state, resumes processing, halts after three seconds, executes a single step for inspection, and finally terminates all threads.

Summary

• ApraPipes pipeline control flow is centralized through the PipeLine class, which delegates commands exclusively to source modules.
• Play and pause operations toggle the mPlay flag and either enqueue PlayPauseCommand objects or directly update state via handlePausePlay().
• Step commands bypass continuous execution, triggering a single produce() cycle through the StepCommand queue mechanism.
• Stop transitions the pipeline to PL_STOPPING state, terminates module threads, and clears pending commands.
• All control operations are thread-safe, utilizing command queues for running modules and direct state manipulation for initialized but non-running modules.

Frequently Asked Questions

How does ApraPipes ensure thread safety during play and pause transitions?

ApraPipes ensures thread safety by using command queues for active modules. When play() or pause() is called on a running source module, it creates a PlayPauseCommand and enqueues it via queuePlayPauseCommand(). The module's internal thread processes this command asynchronously through processSourceQue(), preventing race conditions between the controlling thread and the processing loop.

What is the difference between pause and stop in ApraPipes?

Pause maintains the pipeline's threaded state and module connections while halting frame generation at the source level; it sets mPlay = false but keeps threads alive. Stop transitions the pipeline to PL_STOPPING state, calls stop() on every source module to terminate their internal threads, clears command queues, and prepares the pipeline for destruction or re-initialization.

Can I use the step command while the pipeline is playing?

No, the step() method explicitly checks the mPlay flag and returns immediately if the pipeline is already playing. As implemented in base/src/PipeLine.cpp, stepping is strictly a debugging feature for paused pipelines. To step through frames, you must first call pause(), then invoke step() to trigger individual produce() cycles.

Which modules receive control commands in a multi-source pipeline?

Only modules with Module::SOURCE nature receive play, pause, and step commands. The PipeLine class iterates through its module list and filters using getNature() == Module::SOURCE before invoking control methods. Transform and sink modules do not receive these commands directly; they process data reactively based on what source modules produce.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}