Thread Safety Considerations for Developing Custom ApraPipes Modules: A Complete Guide

Custom ApraPipes modules must respect the framework's thread-per-module architecture by using thread-safe queues for data exchange, protecting mutable configuration with mutexes, and avoiding direct access to internal queue containers.

ApraPipes is an open-source C++ framework maintained by apra-labs/aprapipes for building high-performance media processing pipelines. When developing custom modules, understanding thread safety considerations is essential because the framework executes each module in its own std::thread, requiring strict adherence to synchronization protocols for shared data and queue-based communication.

Understanding the Thread-Per-Module Execution Model

The framework's concurrency architecture centers on PipeLine::run_all_threaded() defined in base/src/PipeLine.cpp. This method creates a dedicated std::thread for every module in the pipeline and invokes Module::operator()(), which serves as the thread's main processing entry point. According to the implementation at lines 57-66, the runtime also calls Utils::setModuleThreadName immediately after thread creation to aid debugging.

Because every module runs concurrently with its upstream and downstream neighbors, direct method calls between modules are prohibited. Instead, all communication flows through FrameContainerQueue instances that act as bounded buffers between thread boundaries.

Thread Safety Guarantees by Framework Component

The ApraPipes runtime provides specific synchronization guarantees for core infrastructure, but custom module developers must uphold corresponding contracts to prevent data races.

Module Lifecycle and Queue Management

The runtime guarantees that PipeLine::run_all_threaded() properly initializes threads and manages the Module::operator()() lifecycle loop. Developers must ensure that their implementation of operator() does not block indefinitely without yielding, as this would stall the pipeline termination sequence. The loop should check isRunning() frequently and use getInputFrame() and pushFrame() to interact with the module's FrameContainerQueue.

FrameContainerQueue in base/include/FrameContainerQueue.h inherits from a bounded buffer implementation that uses std::mutex for internal synchronization. Similarly, ThreadSafeQue<T> in base/include/ThreadSafeQue.h wraps std::deque with std::mutex and std::condition_variable, ensuring that all push, pop, and try_pop operations are atomic.

Property Changes and setProps Validation

The Module::setProps method in base/src/Module.cpp (lines 202-218) enforces that the qlen (queue length) property cannot be modified after construction. This validation is critical because changing the bounded buffer size post-initialization would violate the thread-safety guarantees of FrameContainerQueue. The method updates internal pacers and profilers atomically.

When overriding setProps to support runtime configuration changes (such as adjusting fps or custom parameters), developers must protect mutable members with std::mutex or std::atomic to synchronize access between the processing thread and external control threads.

Module Registration and Factory Safety

The declarative factory system uses std::call_once in base/include/declarative/ModuleRegistrations.h (lines 16-25) to ensure that ensureBuiltinModulesRegistered() executes exactly once, even when invoked from multiple threads during pipeline initialization. Custom modules registered via ModuleFactory inherit this protection automatically, preventing double-registration races and static initialization order fiascos.

Shared Hardware Resources

Many hardware-specific helpers protect their internal state with std::mutex. For example, NvArgusCameraHelper in base/include/NvArgusCameraHelper.h (lines 33-35) uses mutex protection for camera state, while H264Decoder in base/include/H264Decoder.h (lines 92-95) protects decoder state. When custom modules hold references to these shared resources, they can rely on the helper's internal locking, but must ensure that any additional mutable state in the custom module itself is independently protected.

Implementing Thread-Safe Custom Modules: Code Examples

Minimal Custom Transform Module with Mutex Protection

The following skeleton demonstrates proper synchronization for runtime configuration and correct queue handling:

// MyTransformModule.h
#pragma once
#include "Module.h"
#include "Frame.h"
#include <mutex>

class MyTransformModule : public Module {
public:
    MyTransformModule(const std::string &name, const ModuleProps &props)
        : Module(Module::TRANSFORM, name, props) {}

    void setProps(ModuleProps &props) override {
        // Respect base validation (qlen cannot change)
        Module::setProps(props);
        // Protect mutable members under lock
        std::lock_guard<std::mutex> lk(mConfigMutex);
        mGain = props.fps;  // example: reuse fps as gain
    }

    bool init() override {
        LOG_INFO << getId() << " initialized";
        return true;
    }

    void operator()() override {
        while (isRunning()) {
            // Thread-safe frame retrieval via FrameContainerQueue
            FramePtr in = getInputFrame();
            if (!in) { continue; }

            // Process using local data copy
            FramePtr out = makeFrame(in->size());
            float currentGain;
            {
                std::lock_guard<std::mutex> lk(mConfigMutex);
                currentGain = mGain;
            }
            
            std::transform(in->data(), in->data() + in->size(),
                           out->data(),
                           [currentGain](uint8_t v) { 
                               return static_cast<uint8_t>(v * currentGain); 
                           });

            // Thread-safe frame submission
            pushFrame(out);
        }
        LOG_INFO << getId() << " stopped";
    }

private:
    std::mutex mConfigMutex;
    float mGain{1.0f};
};

Key thread-safety elements in this implementation:

• Line 17-18: Module::setProps validates that qlen remains unchanged, enforcing the bounded buffer contract from base/src/Module.cpp.
• Line 20-21: std::lock_guard protects mGain from races between the processing thread and external configuration threads.
• Line 33: isRunning() provides a thread-safe exit condition for the main loop.
• Line 36: getInputFrame() internally uses FrameContainerQueue, which relies on ThreadSafeQue from base/include/ThreadSafeQue.h.
• Line 44-47: Local copy of mGain minimizes lock duration, preventing contention during pixel processing.

Integrating Thread-Safe Hardware Helpers

When integrating hardware decoders that provide internal synchronization:

#include "H264Decoder.h"

class MyDecoderModule : public Module {
public:
    MyDecoderModule(const std::string &name, const ModuleProps &props)
        : Module(Module::SOURCE, name, props), 
          decoder(std::make_shared<H264Decoder>()) {}

    bool init() override {
        decoder->open();  // Thread-safe initialization
        return true;
    }

    void operator()() override {
        while (isRunning()) {
            auto encoded = getInputFrame();
            if (!encoded) continue;

            // decode() is internally synchronized via std::mutex
            auto decoded = decoder->decode(encoded);
            pushFrame(decoded);
        }
    }

private:
    std::shared_ptr<H264Decoder> decoder;
};

The H264Decoder class in base/include/H264Decoder.h protects its internal state with a std::mutex (lines 92-95), allowing safe concurrent access from multiple module threads without additional locking in your code.

Thread-Safe Module Registration

Registering custom modules safely using the declarative factory:

#include "declarative/ModuleFactory.h"
#include "MyTransformModule.h"

static void registerMyModule() {
    // Guarded by std::call_once in ModuleRegistrations.h
    ModuleFactory::registerModule("my_transform",
        [](const std::string& name, const ModuleProps& props) {
            return boost::shared_ptr<Module>(new MyTransformModule(name, props));
        });
}

// Static initializer ensures single registration
static const bool myModuleRegistered = []() { 
    registerMyModule(); 
    return true; 
}();

The ensureBuiltinModulesRegistered() function in base/include/declarative/ModuleRegistrations.h uses std::call_once (lines 16-25) to prevent race conditions during registration, ensuring your custom module is registered exactly once even in multi-threaded initialization scenarios.

Critical Thread Safety Rules for Custom Module Development

To ensure reliable operation within the ApraPipes framework, adhere to these architectural constraints derived from the source code:

• Never access queue internals directly. Always use push(), pop(), or try_pop() on FrameContainerQueue rather than accessing the underlying std::deque directly, as the mutex protection in base/include/ThreadSafeQue.h only covers the public API.

• Respect the immutable queue length contract. The qlen property cannot be modified after module construction, as enforced by Module::setProps in base/src/Module.cpp (lines 202-218). Changing the bounded buffer size post-initialization would violate the thread-safety guarantees of FrameContainerQueue.

• Protect mutable configuration state. When overriding setProps to support runtime changes, use std::mutex with std::lock_guard to synchronize access between the processing thread and external control threads. Copy configuration values locally before processing to minimize lock duration.

• Keep the main loop responsive. The operator() method should check isRunning() frequently and avoid blocking indefinitely. Long-running operations should be broken into smaller units or performed on local data copies to ensure proper pipeline shutdown.

• Leverage existing thread-safe helpers. Rely on framework components like H264Decoder and NvArgusCameraHelper that already implement internal mutex protection, rather than adding redundant synchronization layers.

• Use framework-provided registration. Register custom modules via ModuleFactory to inherit the std::call_once protection provided by ensureBuiltinModulesRegistered() in base/include/declarative/ModuleRegistrations.h.

Summary

Developing thread-safe custom modules for ApraPipes requires understanding the framework's thread-per-module execution model and respecting the synchronization boundaries established by the runtime. Key takeaways include:

• Each module executes in its own std::thread created by PipeLine::run_all_threaded() in base/src/PipeLine.cpp, communicating exclusively through FrameContainerQueue instances that wrap ThreadSafeQue from base/include/ThreadSafeQue.h.
• The qlen property is immutable after construction, enforced by validation logic in Module::setProps at base/src/Module.cpp lines 202-218.
• Mutable configuration requires explicit synchronization using std::mutex when accessed from both setProps and operator().
• Hardware helpers like H264Decoder and NvArgusCameraHelper provide internal thread safety via std::mutex, allowing safe sharing across module boundaries.
• Module registration through ModuleFactory is thread-safe due to std::call_once protection in base/include/declarative/ModuleRegistrations.h.

Frequently Asked Questions

How does ApraPipes handle thread creation for custom modules?

The framework automatically manages thread creation via PipeLine::run_all_threaded() in base/src/PipeLine.cpp. This method instantiates a dedicated std::thread for every module and invokes Module::operator()() as the thread's entry point. The runtime also calls Utils::setModuleThreadName immediately after creation to set debugging-friendly thread names. Developers implement the operator() method to define processing logic but never manually create or manage threads, as the framework handles lifecycle synchronization and graceful shutdown through the isRunning() mechanism.

Can I change the queue length (qlen) of a module after it has been constructed?

No, the qlen property is immutable after module construction. The Module::setProps method in base/src/Module.cpp (lines 202-218) explicitly validates that queue length remains unchanged during property updates. This restriction exists because FrameContainerQueue in base/include/FrameContainerQueue.h implements a bounded buffer pattern where the capacity is fixed at initialization. Modifying the buffer size post-construction would violate the thread-safety guarantees of the underlying ThreadSafeQue implementation and could lead to race conditions during frame production and consumption.

What synchronization primitive should I use for mutable configuration in custom modules?

Use std::mutex with std::lock_guard to protect mutable configuration members that are accessed from both the processing thread and external control threads. When overriding setProps to handle runtime-adjustable parameters like fps or custom gain values, lock the mutex during assignment to prevent races with the processing loop in operator(). Minimize lock duration by copying protected values to local variables before entering processing loops, as demonstrated in the MyTransformModule example. For simple atomic types, std::atomic may be used instead, but std::mutex provides the necessary protection for complex configuration structures.

Is the ModuleFactory registration thread-safe for dynamically loaded custom modules?

Yes, module registration through ModuleFactory is inherently thread-safe due to std::call_once protection. The ensureBuiltinModulesRegistered() function in base/include/declarative/ModuleRegistrations.h (lines 16-25) uses std::call_once to guarantee that registration code executes exactly once, even when invoked from multiple threads during pipeline initialization. Custom modules registered via ModuleFactory::registerModule inherit this protection automatically, preventing race conditions, double-registration errors, and static initialization order fiascos in multi-threaded environments.

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