template <typename T, size_t Capacity>
class alignas(64) MPSCRingBuffer {
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be power of 2");
    
    // 数据区：存放 T 类型元素，Capacity 必须为 2 的幂次，便于位运算取模
    T slots_[Capacity];
    
    // 生产者侧：多线程竞争写入，需原子操作
    alignas(64) std::atomic<size_t> head_{0};   // 写入位置（生产者索引）
    alignas(64) std::atomic<size_t> tail_{0};   // 读取位置（消费者索引）
    
    // 缓存行填充，防止 head_ 与 tail_ 伪共享
    char pad_[64 - 2 * sizeof(std::atomic<size_t>)];
};

bool try_push(const T& item) noexcept {
    size_t head = head_.load(std::memory_order_relaxed);
    for (;;) {
        size_t tail = tail_.load(std::memory_order_acquire); // Acquire: 同步消费者已读位置
        if (head - tail >= Capacity) return false;           // 队列满
        
        // 尝试 CAS 占位：仅 head_ 变更，数据区暂不写入
        if (head_.compare_exchange_weak(
                head, head + 1, 
                std::memory_order_acq_rel,   // 成功：AcqRel（Release+Acquire）
                std::memory_order_relaxed))  // 失败：Relaxed 重试
        {
            break; // 成功占位，head 为旧值，对应槽位索引
        }
        // CAS 失败：head 已被其他线程更新，循环重试
    }
    
    // 【关键】数据写入必须在 CAS 成功后，且带 Release 语义
    slots_[head & (Capacity - 1)] = item; // 普通赋值，依赖后续 Release fence
    std::atomic_thread_fence(std::memory_order_release); // 确保数据写入对消费者可见
    return true;
}

bool try_pop(T& out) noexcept {
    size_t tail = tail_.load(std::memory_order_relaxed);
    size_t head = head_.load(std::memory_order_acquire); // Acquire: 同步生产者发布的数据
    
    if (head == tail) return false; // 队列空
    
    // 读取数据：Acquire 语义保证读到完整数据
    out = slots_[tail & (Capacity - 1)];
    
    // 更新 tail_：Release 语义通知生产者槽位可复用
    tail_.store(tail + 1, std::memory_order_release);
    return true;
}

size_t try_pop_batch(T* batch, size_t max_count) noexcept {
    size_t tail = tail_.load(std::memory_order_relaxed);
    size_t head = head_.load(std::memory_order_acquire);
    size_t available = head - tail;
    size_t count = std::min(available, max_count);
    
    for (size_t i = 0; i < count; ++i) {
        batch[i] = slots_[(tail + i) & (Capacity - 1)];
    }
    tail_.store(tail + count, std::memory_order_release);
    return count;
}

┌─────────────────────────────────────────────────────────────┐
│                    应用层（媒体业务逻辑）                      │
│  MPSCRingBuffer<RefCountedBuffer>  ← 仅传递智能指针/句柄     │
├─────────────────────────────────────────────────────────────┤
│                    共享内存池（ShmPool）                       │
│  预分配 4KB/16KB 块，引用计数管理，跨进程/线程零拷贝传递      │
├─────────────────────────────────────────────────────────────┤
│                    内核旁路 / XDP / DPDK                       │
│  网卡 DMA 直接写入用户态巨页，避免 skb 拷贝、协议栈开销        │
└─────────────────────────────────────────────────────────────┘

// 使用 C++20 std::atomic_ref 实现线程安全引用计数
class RefCountedBuffer {
    alignas(64) std::atomic<uint32_t> ref_cnt_{1};
    uint32_t capacity_;
    uint32_t size_;
    uint8_t data_[]; // 柔性数组，紧随对象后
    
public:
    void retain() noexcept { ref_cnt_.fetch_add(1, std::memory_order_relaxed); }
    
    void release() noexcept {
        if (ref_cnt_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
            ShmPool::instance().deallocate(this); // 归还内存池
        }
    }
    
    uint8_t* data() noexcept { return data_; }
    uint32_t size() const noexcept { return size_; }
    void set_size(uint32_t s) noexcept { size_ = s; }
};

// DPDK/XDP 接收回调（运行在独立 Poll 线程）
void on_packet_received(rte_mbuf* mbuf) {
    // 1. 从内存池获取缓冲区（无锁分配，O(1)）
    RefCountedBuffer* buf = ShmPool::instance().allocate(mbuf->pkt_len);
    
    // 2. 单次 memcpy：网卡 DMA 区 → 用户态共享内存（不可避免的一次拷贝）
    //    若使用 AF_XDP + XSK_MAP，可实现真正零拷贝（内核映射同一物理页）
    rte_memcpy(buf->data(), rte_pktmbuf_mtod(mbuf, void*), mbuf->pkt_len);
    buf->set_size(mbuf->pkt_len);
    
    // 3. 入队：仅传递指针，零拷贝
    while (!ring_buffer.try_push(buf)) {
        // 队列满：背压策略（丢包/阻塞/扩容）
        handle_backpressure(buf);
    }
    
    // 4. 释放 mbuf 回网卡 RX 环
    rte_pktmbuf_free(mbuf);
}

// 媒体转发线程：从队列取包，直接转发给下游 WebRTC Transport
void forward_loop() {
    RefCountedBuffer* batch[64];
    while (running_) {
        size_t n = ring_buffer.try_pop_batch(batch, 64);
        for (size_t i = 0; i < n; ++i) {
            // 零拷贝：增加引用计数，传递给多个下游 Track
            batch[i]->retain(); 
            for (auto& track : downstream_tracks_) {
                track->send_rtp(batch[i]->data(), batch[i]->size());
            }
            batch[i]->release(); // 发送完成释放
        }
    }
}

# GCC/Clang 关键编译选项
-O3 -march=native -mtune=native 
-fno-strict-aliasing 
-fomit-frame-pointer 
-DNDEBUG

struct RingBufferMetrics {
    std::atomic<uint64_t> push_total{0};
    std::atomic<uint64_t> push_failed{0};
    std::atomic<uint64_t> pop_total{0};
    std::atomic<uint64_t> pop_empty{0};
    // 延迟直方图（使用 HDR Histogram 或 Prometheus Histogram）
    hdr_histogram* push_latency_ns;
    hdr_histogram* pop_latency_ns;
};

sequenceDiagram
    participant P1 as 生产者 Core 0 (L1/L2)
    participant Bus as 总线/互联 (Mesh/Ring)
    participant P2 as 消费者 Core 15 (L1/L2)
    participant Mem as 内存控制器
    
    P1->>Bus: RFO (Request For Ownership) - 独占缓存行
    Bus-->>P1: 授予 Modified (M) 状态
    P1->>P1: 修改 head_ 值 (Store Buffer)
    P1->>Bus: Writeback / Snoop Response (可选)
    P2->>Bus: Read Request (Acquire Load)
    Bus-->>P2: 数据转发 (Forward from P1 or Mem)
    P2->>P2: 缓存行变为 Shared (S) / Exclusive (E)

// 生产者线程局部存储
thread_local size_t local_head = 0;
thread_local size_t local_batch = 64; // 批量申请大小

bool try_push_batched(T* items, size_t count) {
    // 1. 本地缓冲累积（无原子操作，极快）
    // ... 将 items 放入 thread_local 环形缓冲 ...
    
    // 2. 累积满 batch 或定时刷新时，一次性 CAS 全局 head
    if (should_flush()) {
        size_t expected = global_head_.load(std::memory_order_relaxed);
        while (!global_head_.compare_exchange_weak(
                expected, expected + count,
                std::memory_order_acq_rel, std::memory_order_relaxed)) {
            // 竞争失败极少，因为批量大，CAS 频率降低 64 倍
        }
        // 3. 批量拷贝数据到全局 slots_ (可用 SIMD memcpy)
        // 4. Release Fence
    }
}

// 消费者阻塞等待
T pop_blocking() {
    size_t tail = tail_.load(std::memory_order_relaxed);
    while (true) {
        size_t head = head_.load(std::memory_order_acquire);
        if (head != tail) {
            T item = slots_[tail & mask];
            tail_.store(tail + 1, std::memory_order_release);
            return item;
        }
        // 【C++20 核心特性】原子等待：内核级休眠，零 CPU 占用
        // 仅当 head_ 发生变化时被唤醒，避免虚假唤醒
        head_.wait(tail, std::memory_order_relaxed); 
        tail = tail_.load(std::memory_order_relaxed); // 更新 tail 继续循环
    }
}

// 生产者唤醒 (在 Release Fence 之后调用)
head_.notify_one(); // 或 notify_all()

class PriorityMPSCRing {
    MPSCRingBuffer<RtpPacket, 4096> delta_ring_;   // P/B 帧、音频
    MPSCRingBuffer<RtpPacket, 1024> keyframe_ring_; // I 帧、关键音频帧
    std::atomic<uint32_t> keyframe_pending_{0};     // 信号量
    
public:
    // 生产者入队
    void push(RtpPacket&& pkt) {
        if (pkt.is_keyframe()) {
            keyframe_ring_.try_push(std::move(pkt));
            keyframe_pending_.fetch_add(1, std::memory_order_release);
            keyframe_pending_.notify_one(); // 唤醒消费者
        } else {
            delta_ring_.try_push(std::move(pkt));
        }
    }
    
    // 消费者：优先消费关键帧，饥饿保护
    bool pop(RtpPacket& out) {
        // 1. 优先尝试关键帧环
        if (keyframe_pending_.load(std::memory_order_acquire) > 0) {
            if (keyframe_ring_.try_pop(out)) {
                keyframe_pending_.fetch_sub(1, std::memory_order_relaxed);
                return true;
            }
        }
        // 2. 退回普通环
        if (delta_ring_.try_pop(out)) return true;
        
        // 3. 双环均空，阻塞等待关键帧信号量 (避免忙等普通环)
        keyframe_pending_.wait(0, std::memory_order_relaxed);
        return false; // 被唤醒后下次循环重试
    }
};

class NackBuffer {
    // 固定大小数组，索引 = seq_num % Capacity
    // 利用序列号单调递增特性，天然解决 ABA 问题
    struct Slot {
        std::atomic<uint16_t> seq{0}; // 0 表示空闲
        RtpPacket pkt;
    };
    alignas(64) Slot slots_[Capacity]; // Capacity = 2^16 = 65536 (覆盖 16bit seq)
    
public:
    // 生产者：写入新包
    void store(uint16_t seq, RtpPacket&& pkt) {
        Slot& slot = slots_[seq & (Capacity - 1)];
        slot.pkt = std::move(pkt);
        // Release: 发布 seq，消费者可见
        slot.seq.store(seq, std::memory_order_release); 
    }
    
    // 消费者 (NACK 处理线程)：查找重传包
    bool retrieve(uint16_t seq, RtpPacket& out) {
        Slot& slot = slots_[seq & (Capacity - 1)];
        // Acquire: 同步生产者的 pkt 写入
        uint16_t stored_seq = slot.seq.load(std::memory_order_acquire);
        if (stored_seq == seq) {
            out = std::move(slot.pkt);
            slot.seq.store(0, std::memory_order_relaxed); // 标记空闲 (可选，或由新包覆盖)
            return true;
        }
        return false; // 包已被覆盖或未收到
    }
};

// Ingress 侧：创建共享内存并传递 fd
int create_shm_buffer(size_t size) {
    int fd = memfd_create("media_buf", MFD_CLOEXEC | MFD_ALLOW_SEALING);
    ftruncate(fd, size);
    // 关键：加密封，防止恶意/误写入篡改大小
    fcntl(fd, F_ADD_SEALS, F_SEAL_SHRINK | F_SEAL_GROW | F_SEAL_SEAL); 
    return fd;
}

// 通过 UDS 发送 fd (SCM_RIGHTS)
void send_fd(int unix_sock, int fd, const PacketMeta& meta) {
    struct msghdr msg = {};
    struct iovec iov = { &meta, sizeof(meta) };
    msg.msg_iov = &iov; msg.msg_iovlen = 1;
    
    char cmsg_buf[CMSG_SPACE(sizeof(int))];
    msg.msg_control = cmsg_buf; msg.msg_controllen = sizeof(cmsg_buf);
    struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
    cmsg->cmsg_level = SOL_SOCKET; cmsg->cmsg_type = SCM_RIGHTS; cmsg->cmsg_len = CMSG_LEN(sizeof(int));
    *(int*)CMSG_DATA(cmsg) = fd;
    
    sendmsg(unix_sock, &msg, 0);
}

// RDMA 连接建立后，注册共享内存池为 Memory Region
struct ibv_mr* mr = ibv_reg_mr(pd, shm_pool_base, shm_pool_size, 
                               IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);

// 生产者：构建 Work Request，直接发送远端
void rdma_send_packet(uint64_t remote_addr, uint32_t rkey, const RtpPacket& pkt) {
    struct ibv_sge sge = { (uint64_t)pkt.data(), pkt.size(), mr->lkey };
    struct ibv_send_wr wr = {}, *bad_wr = nullptr;
    wr.wr_id = (uint64_t)&pkt; // 用于 completion 回调释放 refcnt
    wr.sg_list = &sge; wr.num_sge = 1;
    wr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM; // 利用 IMM 字段传 seq_num
    wr.imm_data = htonl(pkt.sequence_number());
    wr.wr.rdma.remote_addr = remote_addr;
    wr.wr.rdma.rkey = rkey;
    wr.send_flags = IBV_SEND_SIGNALED;
    
    ibv_post_send(qp, &wr, &bad_wr);
}

// 完成队列 (CQ) 轮询线程：回收 Buffer
void cq_poller() {
    struct ibv_wc wc;
    while (ibv_poll_cq(cq, 1, &wc) > 0) {
        if (wc.status != IBV_WC_SUCCESS) handle_error(wc);
        // wc.wr_id 指向原始 Packet，释放引用计数
        static_cast<RefCountedBuffer*>(wc.wr_id)->release();
    }
}

// 共享内存中原始字节流
void* shm_ptr = mmap(...); 
// 直接在任意地址上施加原子语义，无需 placement new
std::atomic_ref<uint64_t> head(*static_cast<uint64_t*>(shm_ptr));
head.fetch_add(1, std::memory_order_acq_rel);

struct alignas(std::hardware_destructive_interference_size) CacheLinePaddedAtomic {
    std::atomic<size_t> value;
};

// 生产者
ptr.store(new_data, std::memory_order_release);

// 消费者：指针依赖天然建立顺序 (无需额外 fence)
Node* p = ptr.load(std::memory_order_consume); // 理论最优，但编译器支持不一
// 稳健写法：
Node* p = ptr.load(std::memory_order_acquire); // 稍重但绝对安全
process(p->data); // 数据依赖

struct PopAwaitable {
    MPSCRingBuffer<Packet, 4096>& ring;
    Packet* out;
    
    bool await_ready() noexcept { return ring.try_pop(*out); }
    
    void await_suspend(std::coroutine_handle<> h) noexcept {
        // 注册到事件循环：当 ring 有数据时 resume
        // 可结合 eventfd + epoll / io_uring 实现
        reactor::instance().wait_for_ring_data(ring, h); 
    }
    
    Packet await_resume() noexcept { return *out; }
};

// 业务代码极其简洁
auto process_loop() -> Task<void> {
    while (true) {
        Packet pkt = co_await PopAwaitable{ring, &pkt};
        co_await process_packet(pkt); // 后续处理也可协程化
    }
}

# 必开编译旗帜
-Wall -Wextra -Wpedantic 
-Wthread-safety-analysis (Clang Thread Safety Analysis) 
-fsanitize=thread (TSAN) -fsanitize=undefined (UBSAN) 
-fno-omit-frame-pointer (方便 perf/profiling)

class MPSCRingBuffer {
    std::atomic<size_t> head_ GUARDED_BY(head_); // 逻辑锁概念
    std::atomic<size_t> tail_ GUARDED_BY(tail_);
    
    // 标注需要独占访问的成员函数 (仅文档/静态检查用)
    void push(T&& item) EXCLUSIVE_LOCK_FUNCTION(head_) { ... }
    bool try_pop(T& out) EXCLUSIVE_LOCK_FUNCTION(tail_) { ... }
};

// 线性化点记录器 (生产环境采样 1%)
struct LinearizationPoint {
    enum Op { PUSH, POP };
    Op op;
    uint64_t seq;      // 全局单调序列号 (rdtsc)
    uint64_t thread_id;
    void* item_ptr;    // 数据指针
    bool success;
};

// 离线分析：将所有线程记录的 Point 按 seq 排序，验证是否存在合法的顺序历史
// 即：每个 Pop 必须匹配一个先于它的成功 Push，且 FIFO 顺序
bool verify_linearizability(std::vector<LinearizationPoint>& log) {
    std::sort(log.begin(), log.end(), [](auto& a, auto& b){ return a.seq < b.seq; });
    std::queue<void*> expected_queue;
    for (auto& p : log) {
        if (p.op == PUSH && p.success) expected_queue.push(p.item_ptr);
        if (p.op == POP && p.success) {
            if (expected_queue.empty() || expected_queue.front() != p.item_ptr) 
                return false; // 违反 FIFO 或读取未写入数据
            expected_queue.pop();
        }
    }
    return true;
}