┌─────────────────────────────────────────────────────────────┐
│                    编码管线调度层 (Pipeline Scheduler)       │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │  帧任务队列  │→ │ 依赖图解析器 │→ │ 优先级调度器 │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└─────────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
        ┌──────────┐    ┌──────────┐    ┌──────────┐
        │ Worker 0 │    │ Worker 1 │    │ Worker N │  ← 帧级线程池
        └──────────┘    └──────────┘    └──────────┘
              │               │               │
              ▼               ▼               ▼
        ┌──────────────────────────────────────────┐
        │           WPP 行任务调度器 (Row Scheduler) │
        │  ┌─────┐ ┌─────┐ ┌─────┐      ┌─────┐    │
        │  │Row 0│→│Row 1│→│Row 2│ →...→│Row H│    │
        │  └─────┘ └─────┘ └─────┘      └─────┘    │
        └──────────────────────────────────────────┘

// 行组任务：调度最小单元，减少任务分发开销
struct RowGroupTask {
    uint32_t frame_id;
    uint32_t start_ctu_row;      // 起始 CTU 行
    uint32_t num_rows;           // 行组高度（典型 4~8 行）
    uint32_t dep_rows_completed; // 已完成的前置依赖行数（原子计数）
    CabacContextSnapshot ctx_snapshot; // 入口上下文快照
    std::atomic<uint32_t> progress{0}; // 进度：已完成 CTU 数
};

// Worker 本地上下文，避免频繁访问全局状态
struct WorkerLocalContext {
    CabacContext cabac;          // 本地 CABAC 状态机
    ReconstructBuffer recon;     // 本地重构像素缓存（行组级）
    BitstreamWriter bs_writer;   // 码流写入器
    StatsCollector stats;        // 性能计数器
};

void Worker::run_wpp_loop(FrameContext& frame) {
    while (auto row_group = row_scheduler_.pop_ready_row_group(frame)) {
        // 1. 加载入口上下文（仅首行组需从全局加载）
        if (row_group->start_ctu_row == 0) {
            local_ctx_.cabac = frame.global_cabac_entry;
        } else {
            local_ctx_.cabac = row_group->ctx_snapshot;
        }

        // 2. 编码行组内所有 CTU 行
        for (uint32_t r = 0; r < row_group->num_rows; ++r) {
            encode_ctu_row(frame, row_group->start_ctu_row + r, local_ctx_);
            row_group->progress.fetch_add(ctus_per_row, std::memory_order_release);
        }

        // 3. 行组结束：发布上下文快照供下一行组消费
        if (row_group->start_ctu_row + row_group->num_rows < frame.height_in_ctu_rows) {
            RowGroupTask* next = frame.row_groups[row_group->start_ctu_row + row_group->num_rows];
            next->ctx_snapshot = local_ctx_.cabac.snapshot();
            // 原子递增依赖计数，唤醒下一行组
            if (next->dep_rows_completed.fetch_add(1, std::memory_order_acq_rel) == 1) {
                row_scheduler_.push_ready(next);
            }
        }

        // 4. 码流合并：行组级 bitstream 追加到帧级缓冲区
        frame.bitstream.append(local_ctx_.bs_writer.data());
    }
}

class NumaAwareThreadPool {
public:
    explicit NumaAwareThreadPool(size_t worker_per_numa) {
        auto topo = hwloc::Topology::discover();
        for (auto& node : topo.numa_nodes()) {
            for (size_t i = 0; i < worker_per_numa; ++i) {
                auto cpu = node.allocate_cpu(); // 独占物理核
                workers_.emplace_back([this, cpu, node_id = node.id()] {
                    pin_thread_to_cpu(cpu);
                    set_numa_preferred(node_id);
                    worker_loop();
                });
                numa_workers_[node_id].push_back(workers_.back().get_id());
            }
        }
    }

    // 提交帧任务时优先调度到同 NUMA 节点的 Worker
    void submit(EncodeTask task) {
        size_t target_numa = select_numa_node(task); // 结合 DPB 所在内存节点
        queues_[target_numa].push(std::move(task));
    }
private:
    std::vector<std::thread> workers_;
    std::unordered_map<int, LockFreeQueue<EncodeTask>> queues_;
};

struct DpbSlot {
    std::atomic<int> ref_count{0};      // 正在被多少帧引用
    std::atomic<uint64_t> release_fence{0}; // 完成编码的帧序列号
    std::mutex mtx;                     // 保护像素数据搬移
    PictureBuffer pic;
};

// 编码前获取参考帧
bool acquire_ref_frames(EncodeTask& task, std::vector<DpbSlot*>& refs) {
    for (auto idx : task.ref_indices) {
        DpbSlot* slot = dpb_[idx];
        int expected = slot->ref_count.load();
        while (expected >= 0 && !slot->ref_count.compare_exchange_weak(expected, expected + 1)) {}
        if (expected < 0) return false; // 帧已被标记释放
        refs.push_back(slot);
    }
    return true;
}

// 编码完成后释放
void release_ref_frames(const std::vector<DpbSlot*>& refs, uint64_t frame_seq) {
    for (auto slot : refs) {
        if (slot->ref_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
            slot->release_fence.store(frame_seq, std::memory_order_release);
            // 通知 DPB 回收器可复用该槽位
            dpb_recycler_.notify_available(slot);
        }
    }
}

// 关键节点打点（使用 OpenTelemetry / Prometheus）
struct EncodeTelemetry {
    uint64_t frame_id;
    uint64_t enqueue_ts;      // 入队时间
    uint64_t schedule_ts;     // 调度开始
    uint64_t first_row_ts;    // 首行组启动
    uint64_t last_row_ts;     // 末行组完成
    uint64_t pack_ts;         // 打包完成
    uint32_t worker_id;
    uint32_t numa_node;
    FrameType frame_type;
    uint32_t qp;
    uint32_t bit_size;
};

graph TD
    A[帧任务入队] --> B{帧类型?}
    B -->|I/P 关键帧| C[阶段一: 全局预估器]
    B -->|B 帧/非参考帧| D[简化预估: 继承参考帧 Lambda]
    C --> E[特征提取: 纹理/运动/残差直方图]
    E --> F[轻量级 R-Q 模型推理<br/>(Linear / Tiny MLP)]
    F --> G[输出: Frame Lambda + CTU 行级 QP 调制表]
    G --> H[下发至 Worker / RowGroup]
    H --> I[阶段二: 局部 RDO 执行]
    I --> J[实际 Bits 回收]
    J --> K[阶段三: 误差反馈校正]
    K --> L[更新全局 R-Q 模型参数<br/>(EMA 平滑)]
    L --> C

class GlobalBandwidthArbiter {
    TokenBucket global_bucket_{total_upload_mbps * 0.95}; // 预留 5% 信令
    std::map<StreamId, TokenBucket> stream_buckets_;
    std::mutex mtx_;

public:
    // 编码前申请预算，返回本帧可用 bits 上限
    uint64_t request_budget(StreamId sid, FrameType type, uint64_t estimated_bits) {
        std::lock_guard lk(mtx_);
        // 优先级：主视频 > 辅流 > 录制
        double weight = get_priority_weight(sid, type);
        uint64_t fair_share = global_bucket_.available() * weight;
        uint64_t grant = std::min(estimated_bits * 1.2, fair_share); // 允许 20% 突发
        stream_buckets_[sid].consume(grant);
        global_bucket_.consume(grant);
        return grant;
    }

    // 编码后归还差额
    void return_unused(StreamId sid, uint64_t unused_bits) {
        std::lock_guard lk(mtx_);
        stream_buckets_[sid].return_tokens(unused_bits);
        global_bucket_.return_tokens(unused_bits);
    }
};

# GCC/Clang 编译旗（生产构建强制开启）
-fno-fast-math -fno-associative-math -fno-signed-zeros -fno-trapping-math
-frounding-math -fsignaling-nans
# 关键：禁用融合乘加 (FMA) 指令重排，或显式使用 _mm256_fmadd_ps 并固定顺序

#include <xmmintrin.h>
void enforce_fp_determinism() {
    _MM_SET_FLUSH_ZERO_MODE(_MM_FLUSH_ZERO_ON);   // 逐下溢归零
    _MM_SET_DENORMALS_ZERO_MODE(_MM_DENORMALS_ZERO_ON); // 非规格数归零
    _MM_SET_ROUNDING_MODE(_MM_ROUND_NEAREST);     // 统一舍入模式
    // 禁用精度异常掩码，捕获潜在数值问题
    _mm_setcsr(_mm_getcsr() | 0x1F80); 
}

# tools/determinism_check.py
def compare_bitstreams(bin_a, bin_b, tolerance=0):
    """基于 NAL 单元语义比对，而非字节比对"""
    nal_a = parse_annexb(bin_a)
    nal_b = parse_annexb(bin_b)
    assert len(nal_a) == len(nal_b), "NAL count mismatch"
    for i, (na, nb) in enumerate(zip(nal_a, nal_b)):
        assert na.type == nb.type, f"NAL {i} type diff"
        # 允许 trailing_bits 填充字节差异
        if na.payload[:-1] != nb.payload[:-1]: 
            diff = find_first_diff_bit(na.payload, nb.payload)
            raise DeterminismError(f"NAL {i} bit diff at {diff}")

class WcBitstreamBuffer {
    void* base_ = nullptr;
    size_t capacity_ = 0;
    std::atomic<size_t> write_pos_{0};

public:
    explicit WcBitstreamBuffer(size_t cap) : capacity_(align_up(cap, 4096)) {
        // 1. 申请物理连续大页 (1GB HugeTLB 最佳，2MB 次之)
        int fd = open("/dev/hugepages/encoder_bs", O_CREAT | O_RDWR, 0755);
        base_ = mmap(nullptr, capacity_, PROT_READ | PROT_WRITE, 
                     MAP_SHARED | MAP_HUGETLB | MAP_POPULATE, fd, 0);
        // 2. 设置 PAT/MSR 将该物理页标记为 WC (Write-Combining)
        //    需 root 权限或 CAP_SYS_ADMIN，通常由初始化服务一次性配置
        set_memory_wc(base_, capacity_); 
    }

    // 无锁追加，单生产者（编码线程）多消费者（网络发送线程）
    bool append(const void* data, size_t len) {
        size_t pos = write_pos_.fetch_add(len, std::memory_order_relaxed);
        if (pos + len > capacity_) return false; // 环形缓冲区需处理回绕
        // 非临时存储指令 (MOVNTDQA / _mm256_stream_si256) 直写 WC 内存
        const __m256i* src = static_cast<const __m256i*>(data);
        __m256i* dst = reinterpret_cast<__m256i*>(static_cast<char*>(base_) + pos);
        for (size_t i = 0; i < len / 32; ++i) _mm256_stream_si256(dst + i, src[i]);
        // 剩余字节标量写入
        return true;
    }
};

# 将 L3 Cache 划分为 3 个 Class of Service (CLOS)
# CLOS 0 (High Pri): 编码 Worker 核心 -> 占 70% L3 Ways (保护重构像素/参考帧)
# CLOS 1 (Low Pri) : 网络发送/日志/监控 -> 占 15% L3 Ways
# CLOS 2 (System)  : OS/Container Runtime -> 占 15% L3 Ways

# 配置示例
pqos -a "llc:0=0x7fff0000;llc:1=0x00007fff;llc:2=0x00008000"
pqos -e "llc:0=0-27;llc:1=28-55;llc:2=56-63"  # 绑定核心到 CLOS

# encoding-session.yaml
apiVersion: media.example.com/v1alpha1
kind: EncodingSession
metadata:
  name: conf-1080p-main-001
spec:
  codec: vvc
  resolution: "1920x1080"
  fps: 30
  # 硬拓扑需求
  topology:
    cpu:
      sockets: 1              # 独占 1 个 Socket
      coresPerSocket: 14      # 独占 14 物理核 (含 HT 共 28 逻辑核)
      numaNode: 0             # 指定 NUMA 节点
      exclusive: true         # 独占调度
    memory:
      hugepages: "2Mi"        # 必须 2M 巨页
      limit: 8Gi
      numaLocal: true         # 内存本地化
    devices:
      - name: "intel.com/qat" # 可选：QAT 压缩加速卡
        count: 1
  # 业务 SLA
  sla:
    maxEncodingLatencyMs: 10
    targetBitrateKbps: 4500

// runtime/TopologyProbe.cpp
struct ContainerTopology {
    std::vector<int> allowed_cpus;      // cgroup v2 cpuset.cpus.effective
    std::vector<int> allowed_mems;      // cgroup v2 memory.numa_stat
    std::map<int, std::vector<int>> numa_to_cpus; // 解析 /sys/fs/cgroup/cpuset/cpus.partition
};

ContainerTopology probe_container_topo() {
    ContainerTopology topo;
    // 1. 解析 cgroup v2 cpuset
    parse_cpuset("/sys/fs/cgroup/cpuset.cpus.effective", topo.allowed_cpus);
    // 2. 映射 CPU -> NUMA Node (读取 /sys/devices/system/cpu/cpuX/topology/node_id)
    for (int cpu : topo.allowed_cpus) {
        int node = read_numa_node(cpu);
        topo.numa_to_cpus[node].push_back(cpu);
    }
    // 3. 校验独占性：若 allowed_cpus 非连续或含 HT 兄弟核，降级策略/报警
    validate_exclusivity(topo);
    return topo;
}

// 线程池初始化自适应
ThreadPool::ThreadPool(const ContainerTopology& topo) {
    for (auto& [node, cpus] : topo.numa_to_cpus) {
        for (int cpu : cpus) {
            // 跳过 HT 兄弟核 (仅取物理核首个)
            if (is_hyperthread_sibling(cpu, cpus)) continue; 
            workers_.emplace_back([cpu, node] {
                pin_thread(cpu);
                set_mempolicy(MPOL_BIND, node);
                worker_loop();
            });
        }
    }
}

# 使用 bpftrace 实时查看编码线程在哪个 NUMA 节点访问远程内存
bpftrace -e '
tracepoint:syscalls:sys_enter_sched_setaffinity 
/ comm == "vvc_encoder" / 
{ printf("TID %d -> CPU %dn", args->pid, args->cpuset); }

kprobe:handle_mm_fault
/ comm == "vvc_encoder" && arg2 != 0 /  // 缺页中断
{ @numa_miss[pid, numa_node_of_addr(arg0)] = count(); }
'

// 编码器侧：生产者
struct EncodedFrameSlot {
    std::atomic<uint64_t> seq_num{0};      // 单调递增帧序号
    std::atomic<uint32_t> payload_size{0}; // 有效载荷长度
    std::atomic<uint32_t> flags{0};        // KEY_FRAME | FEC_REQUIRED ...
    uint64_t capture_timestamp_us;         // 采集时间戳 (用于 NTP 同步)
    uint64_t encode_finish_ts;             // 编码完成时间 (用于延迟统计)
    // 柔性数组成员：紧随其后为 NALU 数据 (WC 内存)
    alignas(64) char payload[]; 
};

class LockFreeFrameRing {
    // 单生产者(编码线程) 单消费者(打包线程) 无锁环形队列
    // 利用 C++20 atomic_ref 或手写 acquire/release 语义
};

void RtpPacketizer::run(LockFreeFrameRing& ring, PacedSender& pacer) {
    while (auto slot = ring.try_pop()) {
        // 1. 直接在 slot->payload 上构建 RTP Header (前置预留 12+ 字节)
        //    避免 memcpy 到新 buffer
        RtpHeader* hdr = reinterpret_cast<RtpHeader*>(slot->payload - RTP_HEADROOM);
        build_rtp_header(hdr, slot->seq_num, slot->capture_timestamp_us, slot->flags);

        // 2. 分片 (FU-A / VVC AP) —— 仅修改指针/长度元数据，不拷贝 payload
        std::vector<RtpPacket> packets = fragment_vvc_nalu(
            slot->payload, slot->payload_size, MTU_SIZE
        );

        // 3. 推入 Pacing 队列 (基于 Token Bucket / GCC 算法)
        for (auto& pkt : packets) {
            // pkt.data 指向 slot->payload 内部偏移地址
            // pkt.custom_deleter = [slot] { slot->ref_count--; if(0) ring.recycle(slot); };
            pacer.enqueue(std::move(pkt));
        }
    }
}

# chaos/encoding_chaos.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: vvc-encoder-cpu-stress
spec:
  action: stress
  mode: one
  selector:
    labelSelectors:
      app: vvc-encoder
  stressors:
    cpu:
      workers: 4              # 抢占 4 核模拟共置干扰
      load: 80                # 80% 负载
      options: ["--cpu-method", "fft"] # 浮点密集型干扰
  duration: "5m"
  scheduler:
    cron: "@every 30m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: vvc-egress-latency
spec:
  action: delay
  mode: all
  selector:
    labelSelectors:
      app: vvc-encoder
  delay:
    latency: "50ms"
    correlation: "25"
    jitter: "10ms"
  direction: egress
  target:
    selector:
      labelSelectors:
        app: media-gateway
    mode: all