// 缓存行对齐，避免伪共享
struct alignas(64) Slot {
    std::atomic<uint64_t> sequence{0};  // 序列号，用于 CAS 同步
    RtpPacket* packet{nullptr};         // 指向内存池中对象的指针
};

class MpmcRingQueue {
    Slot* slots_;
    const uint32_t capacity_;           // 必须为 2 的幂次
    const uint32_t mask_;
    alignas(64) std::atomic<uint64_t> head_{0};  // 生产者位置
    alignas(64) std::atomic<uint64_t> tail_{0};  // 消费者位置
    // ...
};

// 生产者批量入队，返回实际写入数量
uint32_t TryEnqueueBulk(RtpPacket** packets, uint32_t count) {
    uint64_t pos = head_.fetch_add(count, std::memory_order_relaxed);
    uint32_t written = 0;
    for (uint32_t i = 0; i < count; ++i) {
        Slot& slot = slots_[(pos + i) & mask_];
        // 等待槽位可用（自旋或让权，视延迟敏感度调整）
        while (slot.sequence.load(std::memory_order_acquire) != pos + i) {
            std::this_thread::yield();
        }
        slot.packet = packets[i];
        slot.sequence.store(pos + i + 1, std::memory_order_release);
        ++written;
    }
    return written;
}

// 消费者批量出队
uint32_t TryDequeueBulk(RtpPacket** out, uint32_t max_count) {
    uint64_t pos = tail_.fetch_add(max_count, std::memory_order_relaxed);
    uint32_t read = 0;
    for (uint32_t i = 0; i < max_count; ++i) {
        Slot& slot = slots_[(pos + i) & mask_];
        uint64_t seq = slot.sequence.load(std::memory_order_acquire);
        if (seq != pos + i + 1) break;  // 数据未就绪
        out[i] = slot.packet;
        slot.packet = nullptr;
        slot.sequence.store(pos + i + capacity_ + 1, std::memory_order_release);
        ++read;
    }
    // 归还未读取的预留槽位（简化处理，生产环境需更严谨的回退机制）
    if (read < max_count) tail_.fetch_sub(max_count - read, std::memory_order_relaxed);
    return read;
}

// 线程本地缓存
thread_local struct ThreadCache {
    // 每个 Size Class 维护一个自由链表头
    void* free_lists_[kNumClasses]{};
    uint16_t free_counts_[kNumClasses]{};
    
    // 批量从 CentralCache 取块
    void* Allocate(size_t size_class) {
        if (free_counts_[size_class] == 0) {
            Refill(size_class);
        }
        void* obj = free_lists_[size_class];
        free_lists_[size_class] = *reinterpret_cast<void**>(obj);
        --free_counts_[size_class];
        return obj;
    }
    
    // 批量归还到 CentralCache
    void Deallocate(void* obj, size_t size_class) {
        if (free_counts_[size_class] >= kMaxFreePerClass) {
            Flush(size_class);
        }
        *reinterpret_cast<void**>(obj) = free_lists_[size_class];
        free_lists_[size_class] = obj;
        ++free_counts_[size_class];
    }
} tls_cache;

// 全局 CentralCache（多线程共享，仅用原子操作保护链表头）
class CentralCache {
    std::atomic<void*> central_free_lists_[kNumClasses]{};
    std::mutex class_mutexes_[kNumClasses];  // 仅在 Slab 分配/回收时持有
    
    void* FetchFromCentral(size_t size_class, uint32_t batch) {
        std::lock_guard<std::mutex> lk(class_mutexes_[size_class]);
        void* head = central_free_lists_[size_class].load(std::memory_order_relaxed);
        // ... 取出 batch 个对象，更新链表头 ...
        return head;
    }
    
    void ReturnToCentral(void* list_head, size_t size_class, uint32_t count) {
        std::lock_guard<std::mutex> lk(class_mutexes_[size_class]);
        // ... 头插法归还 ...
    }
};

struct RtpPacket {
    std::atomic<uint16_t> ref_cnt{1};
    uint16_t size_class;
    uint8_t payload[];  // 柔性数组，紧随头部
};

// 入队前增加引用
void Retain(RtpPacket* pkt, uint16_t subscribers) {
    pkt->ref_cnt.fetch_add(subscribers - 1, std::memory_order_relaxed);
}

// 发送完成后释放
void Release(RtpPacket* pkt) {
    if (pkt->ref_cnt.fetch_sub(1, std::memory_order_acq_rel) == 1) {
        ReturnToPool(pkt);  // 归还内存池
    }
}

// 每个工作线程持有：本地入队指针 + 远端入队指针数组
struct WorkerContext {
    MpmcRingQueue* local_queue;                    // 同 Node 生产者入队
    MpmcRingQueue* remote_queues[kMaxNumaNodes];   // 跨 Node 生产者入队
    
    // 网络线程调用：自动路由到正确队列
    bool EnqueuePacket(RtpPacket* pkt, uint16_t src_numa_node) {
        if (src_numa_node == current_numa_node_) {
            return local_queue->TryEnqueue(pkt);
        } else {
            // 远端入队：批量聚合减少跨总线原子操作
            return remote_queues[src_numa_node]->TryEnqueueBulk(&pkt, 1);
        }
    }
};

# 远端队列占比，>15% 需扩容或调整分片
sfu_remote_queue_ratio = sum(rate(sfu_queue_remote_enqueue_total[1m])) / sum(rate(sfu_queue_total_enqueue_total[1m]))

# 跨 NUMA 内存访问延迟（需配合 perf c2c 或 Intel PCM 采集）
numa_remote_access_latency_ns

struct RtpPacketExt : RtpPacket {
    uint8_t  spatial_id;      // SVC 空间层 / Simulcast 编码流 ID
    uint8_t  temporal_id;     // 时间层
    uint16_t layer_bitmap;    // 该包所属层位图（用于快速过滤）
    uint32_t frame_id;        // 归属帧 ID，用于关键帧对齐
};

struct LayeredQueue {
    // 优先级：KeyFrame > High Spatial > Low Spatial > Audio/Control
    MpmcRingQueue queues_[kPriorityCount];  
    uint8_t weights_[kPriorityCount] = {50, 30, 15, 5};  // DRR 权重
    
    // Deficit Round Robin 调度出队
    uint32_t DequeueBulk(RtpPacket** out, uint32_t max) {
        uint32_t total = 0;
        for (int p = 0; p < kPriorityCount && total < max; ++p) {
            deficit_[p] += weights_[p];
            uint32_t can_take = std::min(deficit_[p], max - total);
            uint32_t taken = queues_[p].TryDequeueBulk(out + total, can_take);
            deficit_[p] -= taken;
            total += taken;
        }
        return total;
    }
};

// 使用 C++20 [[clang::annotate]] 或 GCC __attribute__ 标注原子语义
#define ACQUIRE  __attribute__((acquire))
#define RELEASE  __attribute__((release))
#define RELAXED  __attribute__((relaxed))

// 示例：序列号存储必须显式标注
slot.sequence.store(pos + 1, std::memory_order_release) RELEASE;

// 1. 构造极端时序：注入随机 yield/睡眠
class ChaosScheduler {
    void InjectJitter(std::atomic<uint64_t>* var) {
        if (rand() % 1000 == 0) std::this_thread::yield();  // 0.1% 概率让权
    }
};

// 2. 状态机模型检查（参考 CDSchecker 思想）
// 将队列抽象为：State = {head, tail, slots[].seq, slots[].ptr}
// 验证不变量：∀i, slots[i].seq ∈ {head, head+1, ..., tail+capacity}
// 线性化点：Enqueue 成功 CAS sequence；Dequeue 成功 CAS sequence

// 仅在 Debug/Canary 版本开启
#ifdef ENABLE_SHADOW_VALIDATION
    std::mutex shadow_mtx;
    std::deque<RtpPacket*> shadow_queue;
    
    void ShadowEnqueue(RtpPacket* pkt) {
        std::lock_guard lk(shadow_mtx);
        shadow_queue.push_back(pkt);
    }
    
    void ValidateConsistency() {
        // 定期抽样比对无锁队列出队序列与 shadow_queue 是否一致
    }
#endif

NIC (XDP) → XSK Ring (Kernel) → User-space UMEM (内存池) → SFU 无锁队列 → Worker

struct TraceHeader {
    uint64_t trace_id;      // 全局唯一，关联客户端 SDK 上报
    uint64_t span_id;       // 当前跳点
    uint8_t  flags;         // sampled=1
};

# 统计无锁队列 CAS 失败重试分布
bpftrace -e 'uprobe:libsfu.so:MpmcRingQueue::TryEnqueue /arg0/ { @[ustack()] = count(); }'

# 监控内存池跨 NUMA 分配
bpftrace -e 'uprobe:libsfu.so:CentralCache::AllocateOnNode { @numa[arg1] = count(); }'

# 采样包生命周期延迟（入队→出队→发送完成）
bpftrace -e '
  uprobe:libsfu.so:Enqueue { @start[arg0] = nsecs; }
  uprobe:libsfu.so:Dequeue /@start[arg0]/ { @latency = hist(nsecs - @start[arg0]); delete(@start[arg0]); }
'