The shared-state problem

The typical approach to multi-threading: shared data protected by mutexes.

 1class SensorManager {
 2    std::mutex   mtx_;
 3    SensorData   latest_;       // shared — both threads touch it
 4    bool         updated_ = false;
 5public:
 6    void updateFromSensor() {   // called from sensor thread
 7        std::lock_guard lk(mtx_);
 8        latest_ = readSensor();
 9        updated_ = true;
10    }
11
12    SensorData getLatest() {    // called from display thread
13        std::lock_guard lk(mtx_);
14        updated_ = false;
15        return latest_;
16    }
17};

This works for two threads, but as the number of callers grows, lock contention increases, and the reasoning about which thread owns which state becomes harder. Every method is a potential deadlock site.

The active object pattern eliminates shared state: the object owns its own thread and processes all calls as messages from a queue. No external thread ever touches its internals directly.

The active object

 1#include <thread>
 2#include <queue>
 3#include <mutex>
 4#include <condition_variable>
 5#include <functional>
 6#include <atomic>
 7
 8class ActiveObject {
 9public:
10    ActiveObject() : running_(true), thread_([this] { run(); }) {}
11
12    ~ActiveObject() {
13        post([this] { running_ = false; });  // poison pill
14        thread_.join();
15    }
16
17    // Post a message (callable) to the internal queue — non-blocking
18    void post(std::function<void()> msg) {
19        {
20            std::lock_guard<std::mutex> lk(mtx_);
21            queue_.push(std::move(msg));
22        }
23        cv_.notify_one();
24    }
25
26private:
27    void run() {
28        while (running_) {
29            std::function<void()> msg;
30            {
31                std::unique_lock<std::mutex> lk(mtx_);
32                cv_.wait(lk, [this] { return !queue_.empty(); });
33                msg = std::move(queue_.front());
34                queue_.pop();
35            }
36            msg();  // execute in this thread — no external lock needed
37        }
38    }
39
40    std::atomic<bool>              running_;
41    std::thread                    thread_;
42    std::mutex                     mtx_;
43    std::condition_variable        cv_;
44    std::queue<std::function<void()>> queue_;
45};

All state lives inside the object. The thread executes messages sequentially — no two messages run concurrently, so no internal mutex is needed on the state.

A concrete active sensor

 1class SensorService : private ActiveObject {
 2    // State — only accessed from the internal thread
 3    SensorData latest_;
 4    std::vector<std::function<void(SensorData)>> subscribers_;
 5
 6public:
 7    // Public API — called from any thread, posts to internal queue
 8    void requestRead() {
 9        post([this] { doRead(); });
10    }
11
12    void subscribe(std::function<void(SensorData)> cb) {
13        post([this, cb = std::move(cb)] mutable {
14            subscribers_.push_back(std::move(cb));
15        });
16    }
17
18    // Returns a future — caller can wait for the result
19    std::future<SensorData> getLatest() {
20        auto promise = std::make_shared<std::promise<SensorData>>();
21        auto future  = promise->get_future();
22        post([this, promise] {
23            promise->set_value(latest_);
24        });
25        return future;
26    }
27
28private:
29    // Runs only in the internal thread
30    void doRead() {
31        latest_ = readSensor();          // no lock — only this thread accesses latest_
32        for (auto& cb : subscribers_)
33            cb(latest_);
34    }
35};

Usage from multiple threads:

 1SensorService sensor;
 2
 3// Display thread — subscribes to updates
 4sensor.subscribe([](SensorData d) {
 5    display.update(d.temperature);    // called in sensor's thread — be careful!
 6});
 7
 8// Control thread — periodic reads
 9while (true) {
10    sensor.requestRead();
11    std::this_thread::sleep_for(std::chrono::milliseconds(100));
12}
13
14// Query thread — wait for a value
15auto future = sensor.getLatest();
16auto data = future.get();   // blocks until the active object responds

The callback ownership problem

In the subscriber example, the callback display.update runs in the sensor’s thread, not the display’s thread. If display has its own thread and its own state, this is a thread-safety violation.

The correct fix: make the display also an active object, and have callbacks post to the display’s queue:

 1class DisplayService : private ActiveObject {
 2public:
 3    void update(SensorData d) {
 4        post([this, d] { doUpdate(d); });
 5    }
 6private:
 7    void doUpdate(SensorData d) {
 8        lcd.draw(d.temperature);   // only runs in DisplayService's thread
 9    }
10};
11
12DisplayService display;
13
14sensor.subscribe([&display](SensorData d) {
15    display.update(d);   // posts to display's queue — non-blocking, thread-safe
16});

Each active object owns its state and processes messages from its own thread. Communication between them is always via post — never via direct calls.

FreeRTOS variant

On FreeRTOS, the active object pattern maps directly to a task + queue:

 1class ActiveSensorTask {
 2    QueueHandle_t queue_;
 3    TaskHandle_t  task_;
 4
 5    struct Message {
 6        enum class Type { Read, Subscribe, Stop } type;
 7        // payload...
 8    };
 9
10public:
11    ActiveSensorTask() {
12        queue_ = xQueueCreate(16, sizeof(Message));
13        xTaskCreate(taskFn, "Sensor", 512, this, 2, &task_);
14    }
15
16    void requestRead() {
17        Message msg{Message::Type::Read};
18        xQueueSend(queue_, &msg, 0);
19    }
20
21    void requestStop() {
22        Message msg{Message::Type::Stop};
23        xQueueSend(queue_, &msg, portMAX_DELAY);
24    }
25
26private:
27    static void taskFn(void* param) {
28        static_cast<ActiveSensorTask*>(param)->run();
29    }
30
31    void run() {
32        Message msg;
33        while (true) {
34            if (xQueueReceive(queue_, &msg, portMAX_DELAY) == pdTRUE) {
35                if (msg.type == Message::Type::Stop) break;
36                if (msg.type == Message::Type::Read) doRead();
37            }
38        }
39        vTaskDelete(nullptr);
40    }
41
42    void doRead() {
43        SensorData d = readSensor();
44        // notify other tasks via their queues
45    }
46};

The structure is identical: a task owns its state, receives typed messages from a queue, processes them sequentially. No mutex needed on internal state.

Bounded queue and back-pressure

The std::queue in the basic implementation grows without bound — a slow consumer can exhaust memory. Use a bounded queue with a post that signals back-pressure:

1bool post(std::function<void()> msg) {
2    std::lock_guard<std::mutex> lk(mtx_);
3    if (queue_.size() >= maxQueueSize_) return false;  // reject
4    queue_.push(std::move(msg));
5    cv_.notify_one();
6    return true;
7}

On FreeRTOS, xQueueSend with timeout 0 returns errQUEUE_FULL — the caller decides whether to drop, retry, or escalate.

Active object vs actor model

The active object pattern is the C++ version of the actor model (Erlang, Akka). Each object is an actor — isolated state, message-passing communication.

The difference: actors typically have a runtime that routes messages between objects by name or address. Active objects wire up their own queues explicitly — simpler, no framework, but less dynamic.

For small embedded systems, the active object pattern with FreeRTOS tasks and queues gives the benefits of actor isolation without a framework.

Summary

  • Active object = one thread + one queue + all state inside
  • External threads post messages (callables or typed structs) — never call methods directly on shared state
  • The internal thread executes messages sequentially — no internal mutex needed
  • Callbacks that post to another active object’s queue keep the invariant: no cross-thread state access
  • Maps directly to FreeRTOS task + queue — same pattern, different runtime
  • Use bounded queues and handle back-pressure explicitly

What’s next