如何设计嵌入式并发架构: newosp 的事件驱动 + 固定线程池方案

template <typename PayloadVariant,
          uint32_t QueueDepth = 4096,    // 2 的幂
          uint32_t BatchSize = 256>
class AsyncBus;

struct alignas(64) RingBufferNode {
    std::atomic<uint32_t> sequence;          // CAS 序号控制
    MessageEnvelope<PayloadVariant> envelope; // header + payload
};

uint32_t ProcessBatch() {
    for (uint32_t i = 0; i < kBatchSize; ++i) {
        auto* node = &ring_[consumer_pos_ & kMask];
        if (node->sequence.load(acquire) != consumer_pos_ + 1) break;

        __builtin_prefetch(&ring_[(consumer_pos_ + 1) & kMask], 0, 1);
        DispatchToCallbacks(node->envelope);

        node->sequence.store(consumer_pos_ + kQueueDepth, release);
        ++consumer_pos_;
    }
}

template <typename Signature, size_t BufferSize = 2 * sizeof(void*)>
class FixedFunction;

// Bus 内部使用 32B SBO
using CallbackType = FixedFunction<void(const EnvelopeType&), 4 * sizeof(void*)>;
// sizeof(void*) = 8 → 32B，足够存储 [this + 1 指针] 的 lambda

// 传统方案: shared_ptr<Frame> → control block 堆分配 + 原子引用计数
// newosp: variant<RawFrame, ControlCmd, ...> → 直接嵌入 Envelope，sizeof 编译期确定

uint32_t depth = producer_pos_.load(relaxed) - cached_consumer_pos_;
if (depth >= AdmissionThreshold(priority)) {
    // 缓存可能过时，重新读取真实消费位置 (acquire)
    cached_consumer_pos_ = consumer_pos_.load(acquire);
    depth = producer_pos_.load(relaxed) - cached_consumer_pos_;
    if (depth >= AdmissionThreshold(priority)) {
        return false;  // 丢弃
    }
}

// 1. 单线程阻塞 (调试/单核)
osp::SingleThreadExecutor<Payload> exec;
exec.Spin();

// 2. 后台线程 + 休眠策略
osp::StaticExecutor<Payload, osp::PreciseSleepStrategy> exec(strategy);
exec.Start();

// 3. CPU 绑核
osp::PinnedExecutor<Payload, osp::YieldSleepStrategy> exec(/*cpu=*/2);
exec.Start();

// 4. 工业实时
osp::RealtimeConfig cfg;
cfg.sched_policy = SCHED_FIFO;
cfg.sched_priority = 80;
cfg.lock_memory = true;     // mlockall
cfg.cpu_affinity = 3;
osp::RealtimeExecutor<Payload, osp::PreciseSleepStrategy> exec(cfg);
exec.Start();

osp::WorkerPool<Payload> pool({.name = "sensor", .worker_num = 4});
pool.RegisterHandler<SensorData>([](const SensorData& d, const auto&) {
    heavy_computation(d);
});
pool.Start();  // 4 Worker + 1 Dispatcher = 5 线程

osp::IoPoller poller;
poller.Add(tcp_fd, osp::IoEvent::kReadable);
poller.Add(serial_fd, osp::IoEvent::kReadable);

while (running) {
    auto result = poller.Wait(/*timeout_ms=*/100);
    if (!result.has_value()) continue;
    for (uint32_t i = 0; i < result.value(); ++i) {
        auto& ev = poller.Results()[i];
        if (ev.events & osp::IoEvent::kReadable) {
            handle_fd(ev.fd);
        }
    }
}

void TransportReceiver::Run() {
    osp::IoPoller poller;
    poller.Add(socket_fd_, osp::IoEvent::kReadable);

    while (running_) {
        auto result = poller.Wait(100);
        if (!result.has_value() || result.value() == 0) continue;

        uint8_t buf[kMaxFrameSize];
        ssize_t n = ::recv(socket_fd_, buf, sizeof(buf), 0);
        if (n <= 0) continue;

        FrameHeaderV1 header;
        if (!ParseFrame(buf, n, &header)) continue;

        auto payload = Deserialize(buf + sizeof(header), header.type_idx);
        if (payload.has_value()) {
            bus_.Publish(std::move(payload.value()), header.sender_id);
        }
    }
}

#include <osp/bus.hpp>
#include <osp/static_node.hpp>
#include <osp/executor.hpp>
#include <osp/io_poller.hpp>
#include <osp/shutdown.hpp>

// ── 消息类型 (POD, trivially_copyable) ────────────────────────
struct SensorReading {
    uint32_t sensor_id;
    float value;
    uint64_t timestamp_ns;
};
struct ProcessedResult {
    uint32_t sensor_id;
    float filtered_value;
    uint8_t quality;
};
struct AlarmEvent {
    uint32_t sensor_id;
    uint8_t level;
    char message[64];
};

using Payload = std::variant<SensorReading, ProcessedResult, AlarmEvent>;
using Bus = osp::AsyncBus<Payload>;

static_assert(std::is_trivially_copyable_v<SensorReading>);
static_assert(std::is_trivially_copyable_v<ProcessedResult>);
static_assert(std::is_trivially_copyable_v<AlarmEvent>);

// ── Handler (编译期绑定) ──────────────────────────────────────
struct ProcessingHandler {
    Bus& bus;
    float ema_alpha = 0.3f;
    float ema_value = 0.0f;

    void operator()(const SensorReading& r, const osp::MessageHeader&) {
        // 指数移动平均滤波
        ema_value = ema_alpha * r.value + (1.0f - ema_alpha) * ema_value;
        uint8_t quality = (std::abs(r.value - ema_value) < 1.0f) ? 95 : 60;

        ProcessedResult result{r.sensor_id, ema_value, quality};
        bus.Publish(Payload(result), /*sender_id=*/2);
    }

    void operator()(const ProcessedResult& r, const osp::MessageHeader&) {
        // 质量低于阈值时报警
        if (r.quality < 50) {
            AlarmEvent alarm{};
            alarm.sensor_id = r.sensor_id;
            alarm.level = 1;
            std::snprintf(alarm.message, sizeof(alarm.message),
                         "quality=%u below threshold", r.quality);
            bus.Publish(Payload(alarm), /*sender_id=*/2);
        }
    }

    void operator()(const AlarmEvent& a, const osp::MessageHeader&) {
        OSP_LOG_WARN("Alarm", "sensor=%u level=%u: %s",
                     a.sensor_id, a.level, a.message);
    }
};

// ── Main ──────────────────────────────────────────────────────
int main() {
    auto& shutdown = osp::ShutdownManager::Instance();
    auto& bus = Bus::Instance();

    // 处理节点 (StaticNode, 编译期分发)
    ProcessingHandler handler{bus};
    osp::StaticNode<Payload, ProcessingHandler> processor("processor", 2, handler);

    // I/O 采集节点
    osp::Node<Payload> sensor_io("sensor_io", 1);

    // I/O 线程: IoPoller → recv → Publish
    std::thread io_thread([&]() {
        osp::IoPoller poller;
        poller.Add(sensor_fd, osp::IoEvent::kReadable);

        while (!shutdown.IsShutdown()) {
            auto result = poller.Wait(100);
            if (!result.has_value() || result.value() == 0) continue;

            for (uint32_t i = 0; i < result.value(); ++i) {
                auto& ev = poller.Results()[i];
                if (ev.events & osp::IoEvent::kReadable) {
                    SensorReading reading{};
                    ssize_t n = ::read(ev.fd, &reading, sizeof(reading));
                    if (n == sizeof(reading)) {
                        reading.timestamp_ns = osp::SteadyNowNs();
                        sensor_io.Publish(reading);
                    }
                }
            }
        }
    });

    // 消息调度: CPU 2 绑核，PreciseSleep 降低空闲功耗
    osp::PreciseSleepStrategy sleep_strategy(
        1'000'000ULL,   // 默认 1ms
        100'000ULL,     // 最小 100us
        10'000'000ULL   // 最大 10ms
    );
    osp::PinnedExecutor<Payload, osp::PreciseSleepStrategy> executor(2, sleep_strategy);
    executor.Start();

    shutdown.WaitForShutdown();
    executor.Stop();
    io_thread.join();
    return 0;
}