// 典型静态切分伪代码 - 生产环境禁用
int tile_cols = 4, tile_rows = 4;  // 固定 16 Tile
for (int t = 0; t < tile_cols * tile_rows; ++t) {
    thread_pool.submit(encode_tile(t));
}

// 零拷贝边界共享设计
struct TileBoundaryBuffer {
    alignas(64) uint8_t left[64][8];   // 64 行 × 8 像素
    alignas(64) uint8_t top[64][8];
    std::atomic<uint32_t> ready_flag{0};
};

// 编码线程写入
void write_boundary(TileBoundaryBuffer* buf, const Tile* tile) {
    av1_copy_tile_boundary(tile, buf->left, buf->top);
    buf->ready_flag.store(1, std::memory_order_release);
}

// 滤波线程读取
void read_boundary(const TileBoundaryBuffer* buf, Tile* tile) {
    while (!buf->ready_flag.load(std::memory_order_acquire)) _mm_pause();
    av1_apply_loop_filter(tile, buf->left, buf->top);
}

class FrameAwareThreadPool {
    // 每帧一个 TaskGraph，节点 = Tile 编码任务
    struct FrameTaskGraph {
        uint64_t frame_id;
        FrameType type;           // I/P/B
        std::vector<TileTask> tiles;
        std::atomic<int> remaining_tiles;
        std::chrono::steady_clock::time_point deadline; // 硬截止时间
    };

    // 线程本地队列 + 全局优先队列
    thread_local std::deque<TileTask*> local_queue_;
    PriorityQueue<FrameTaskGraph*> global_pq_; // 按 deadline 升序

    // 核心调度循环
    void worker_loop(int core_id) {
        pin_thread_to_core(core_id);
        while (running_) {
            TileTask* task = pop_local_or_steal();
            if (!task) { task = global_pq_.pop(); }
            if (task) {
                execute_tile(task);
                if (--task->graph->remaining_tiles == 0) {
                    on_frame_complete(task->graph);
                }
            }
        }
    }
};

// 重构前：AoS 导致伪共享
struct TileContext {
    FrameHeader hdr;
    MV mv[64][64];
    Coeff coeff[64][64];
    // 相邻 Tile 的 hdr 共享同一 Cache Line
};

// 重构后：热字段 SoA + 64B 对齐
struct TileContextHot {
    alignas(64) FrameHeader hdr;
    alignas(64) MV mv[64][64];
};
struct TileContextCold {
    Coeff coeff[64][64];  // 仅量化/熵编码访问
};

┌─────────────────────────────────────┐
│  Frame Level RC (每帧 1 次)          │
│  - 目标帧大小 = f(VBV, 场景复杂度)   │
│  - 分配 Tile 预算: Budget_i = w_i * │
└──────────────┬──────────────────────┘
               │
┌──────────────▼──────────────────────┐
│  Tile Level RC (并行无锁)            │
│  - Lambda = f(Budget_i, SATD)        │
│  - 编码结束原子累加 actual_bits      │
│  - 误差反馈下一帧 Frame RC           │
└─────────────────────────────────────┘

// 无锁累加器，避免全局锁竞争
struct FrameRateControl {
    std::atomic<int> total_bits{0};
    int target_bits;
    float lambda_base;

    void tile_done(int tile_bits) {
        total_bits.fetch_add(tile_bits, std::memory_order_relaxed);
    }

    float next_frame_lambda() {
        int actual = total_bits.load(std::memory_order_relaxed);
        float err = (actual - target_bits) / (float)target_bits;
        lambda_base *= (1.0f + 0.3f * err); // PI 控制器简化版
        return clamp(lambda_base, LAMBDA_MIN, LAMBDA_MAX);
    }
};

// 三级内存池设计
class Av1MemoryHierarchy {
    // L1: 线程本地 Ring Buffer（无锁、无原子、纯指针环）
    //     存放：TileContextHot、MV 缓存、Coeff 临时块
    //     大小：64KB / 线程，4KB 对齐，热数据 100% L1 命中
    thread_local static RingBuffer<64*1024> tls_ring_;

    // L2: NUMA 节点级 Slab Allocator（2MB HugePage 切分）
    //     存放：FrameHeader、DPB 参考帧、重构像素平面
    //     规格：64KB/128KB/256KB/512KB 4 个 Size Class
    //     分配/释放：单次 CAS，p99 < 50ns
    NumaSlabAllocator numa_slab_[MAX_NUMA_NODES];

    // L3: 进程级 HugePage 保留池（启动时锁定 4GB）
    //     用于：超大分辨率（4K/8K）帧缓冲、应急兜底
    //     策略：mmap(MAP_HUGETLB | MAP_POPULATE | MAP_LOCKED)
    HugePageReserve huge_reserve_;
};

graph LR
    A[AV1 Tile 编码线程] -->|写入重构像素| B(NUMA Slab: 重构帧缓冲)
    B -->|指针传递| C[Loop Filter / CDEF 就地处理]
    C -->|指针传递| D[熵编码输出: OBU 直接写入]
    D -->|指针传递| E[RTP 打包线程: iovec 指向 OBU 内存]
    E -->|sendmmsg| F[内核零拷贝: MSG_ZEROCOPY + 页锁定]
    F --> G[NIC DMA 直接取走]

// VPL 异步任务提交 - 无阻塞设计
class VplAccelerator {
    mfxSession session_;
    std::array<mfxTask, MAX_IN_FLIGHT> tasks_;
    std::atomic<uint32_t> head_{0}, tail_{0};

    // CPU 线程提交 ME 任务
    bool submit_me(const FrameTask* ft) {
        uint32_t idx = tail_.fetch_add(1) % MAX_IN_FLIGHT;
        mfxTask& task = tasks_[idx];
        
        // 1. 填充 VPL 参数：仅指针传递，零拷贝
        task.ext_param = { ft->input_surface, ft->mv_buffer };
        task.callback  = [this, ft](mfxStatus sts) { on_me_done(ft, sts); };
        
        // 2. 异步提交
        mfxStatus sts = MFXVideoCORE_SyncOperation(session_, task.syncp, 0); // 非阻塞
        return sts == MFX_ERR_NONE;
    }

    // 回调在 VPL 内部线程执行，极快返回
    void on_me_done(const FrameTask* ft, mfxStatus sts) {
        if (sts != MFX_ERR_NONE) { fallback_to_cpu_me(ft); return; }
        // 3. 推入 CPU TQ/熵编码队列
        cpu_tq_queue_.push(ft); 
    }
};

// 统一抽象，运行时加载 so/dll，零 ifdef 污染业务代码
class IHwAccel {
public:
    virtual ~IHwAccel() = default;
    virtual bool init(const CodecConfig& cfg) = 0;
    virtual bool submit_me(const FrameTask*) = 0;
    virtual bool submit_lf(const FrameTask*) = 0;
    virtual void flush() = 0;
    virtual Capability query_cap() const = 0; // 支持哪些分辨率/Profile
};

// 工厂模式自动探测
std::unique_ptr<IHwAccel> create_accelerator() {
    if (has_intel_vpl()) return std::make_unique<VplAccel>();
    if (has_amd_amf())   return std::make_unique<AmfAccel>();
    if (has_nvidia_nvenc()) return std::make_unique<NvencAccel>();
    return std::make_unique<CpuFallbackAccel>(); // 兜底
}

// bpf_program.c - 挂载到用户态 USDT 探针
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

struct encode_event_t {
    uint64_t ts_ns;
    uint32_t pid, tid;
    uint32_t frame_id;
    uint16_t tile_id;
    uint8_t  stage;      // 0=ME 1=TQ 2=LF 3=Entropy
    uint16_t latency_us;
    uint32_t l3_miss;    // 通过 PMU 读取
    uint32_t cpu_id;
};

struct {
    __uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
} events SEC(".maps");

SEC("uprobe/av1_encode_tile")
int trace_tile_entry(struct pt_regs *ctx) {
    // 读取函数参数：frame_id, tile_id, stage
    // 记录 TSC 时间戳到 thread-local map
    return 0;
}

SEC("uretprobe/av1_encode_tile")
int trace_tile_exit(struct pt_regs *ctx) {
    // 计算耗时，读取 PMU L3_MISS，发送到 perf ring buffer
    return 0;
}

// 业务代码植入 - 语义化属性，便于 Grafana/Tempo 聚合
void encode_tile(TileTask* task) {
    auto span = tracer->StartSpan("av1.tile.encode",
        { opentelemetry::trace::SpanKind::kInternal },
        { {"frame.id", task->frame_id},
          {"tile.id", task->tile_id},
          {"tile.pixels", task->pixel_count},
          {"codec.profile", "AV1_Main"},
          {"hw.accel", hw_accel_name()} });

    // 关键阶段子 Span
    SCOPED_SPAN("me", [&]{ run_me(task); });
    SCOPED_SPAN("tq_entropy", [&]{ run_tq_entropy(task); });
    
    span->SetAttribute("tile.latency_us", elapsed_us());
    span->End();
}

# PrometheusRule - 编码级 SLO
groups:
- name: av1-encoding-slo
  rules:
  - alert: TileEncodingLatencyP99High
    expr: histogram_quantile(0.99, rate(av1_tile_latency_bucket[1m])) > 8000
    for: 2m
    labels:
      severity: critical
      team: media-infra
    annotations:
      summary: "Tile 编码延迟 p99 超 8ms"
      runbook: "https://wiki/av1-tile-latency-debug"

  - alert: L3MissRateSpike
    expr: rate(av1_l3_miss_total[30s]) / rate(av1_tile_total[30s]) > 0.15
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "L3 Miss 率突增，疑似 NUMA 绑定失效或内存碎片"

# deployment.yaml - 硬性绑定 NUMA
apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      topologySpreadConstraints:
      - maxSkew: 1
        topologyKey: kubernetes.io/hostname
        whenUnsatisfiable: DoNotSchedule
        labelSelector:
          matchLabels:
            app: av1-encoder
      # 关键：要求独占 CPU + 巨页
      containers:
      - name: encoder
        resources:
          limits:
            cpu: "16"           # 独占 16 核
            memory: "32Gi"
            hugepages-2Mi: "4Gi" # 预留 2MB 大页
          requests: *limits
        env:
        - name: GOMAXPROCS
          value: "16"
        - name: NUMA_NODE
          valueFrom:
            fieldRef:
              fieldPath: status.numaNode # 需要 kubelet --topology-manager-policy=single-numa-node

// 启动时自动感知拓扑并绑定
void pin_process_to_numa() {
    // 1. 读取 cgroup cpuset.cpus 有效核心列表
    std::vector<int> allowed_cpus = read_cpuset_cpus();
    
    // 2. 通过 libnuma 映射 CPU -> NUMA Node
    std::map<int, std::vector<int>> node_to_cpus;
    for (int cpu : allowed_cpus) {
        int node = numa_node_of_cpu(cpu);
        node_to_cpus[node].push_back(cpu);
    }
    
    // 3. 选择内存最多的 NUMA Node 作为主节点
    int best_node = max_element(node_to_cpus, 
        [](auto& a, auto& b){ return numa_available_memory(a.first) < numa_available_memory(b.first); });
    
    // 4. 绑定当前进程/线程池
    numa_bind(best_node);                    // 内存分配策略
    pthread_setaffinity_np(pthread_self(), cpu_set_of(node_to_cpus[best_node]));
    
    // 5. 线程池工作线程按 CCX/Cluster 绑核
    thread_pool_.init_with_topology(node_to_cpus[best_node]);
}

# 确认无跨 NUMA 访问
numastat -p $(pidof av1_encoder) | grep numa_miss  # 应为 0
# 确认网卡中断亲和性
cat /proc/interrupts | grep eth0  # 中断向量应落在编码所在 Socket 的核心上

# Dockerfile - 多阶段构建 + 无发行版基础镜像
FROM scratch AS runtime
COPY --from=builder /usr/bin/av1_encoder /usr/bin/av1_encoder
COPY --from=builder /lib/x86_64-linux-gnu/libc.so.6 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libm.so.6 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libpthread.so.0 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libdl.so.2 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/librt.so.1 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libnuma.so.1 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libva.so.2 /lib/x86_64-linux-gnu/
COPY --from=builder /lib/x86_64-linux-gnu/libvpl.so.2 /lib/x86_64-linux-gnu/
USER 65532:65532  # non-root
ENTRYPOINT ["/usr/bin/av1_encoder"]

# CI 流水线强制扫描
syft packages dir:./output -o spdx-json > sbom.spdx.json
grype sbom:sbom.spdx.json --fail-on high --only-fixed
cosign sign --key env://COSIGN_PRIVATE_KEY $IMAGE_URI

// seccomp-profile.json - 仅允许编码必需系统调用
{
  "defaultAction": "SCMP_ACT_ERRNO",
  "architectures": ["SCMP_ARCH_X86_64"],
  "syscalls": [
    { "names": ["read", "write", "sendmmsg", "recvmmsg", "epoll_wait", "clock_gettime", "nanosleep", "futex", "mmap", "munmap", "mprotect", "mlock", "munlock", "exit_group", "rt_sigreturn", "sched_yield", "getcpu", "arch_prctl"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["ioctl"], "action": "SCMP_ACT_ALLOW", "args": [ { "index": 0, "value": 0x40046679, "op": "SCMP_CMP_EQ" } ] }  // 仅允许 VAAPI ioctl
  ]
}

# Pod SecurityContext
securityContext:
  runAsNonRoot: true
  runAsUser: 65532
  runAsGroup: 65532
  readOnlyRootFilesystem: true
  allowPrivilegeEscalation: false
  capabilities:
    drop: ["ALL"]
    add: ["IPC_LOCK", "SYS_RESOURCE"]  # 仅需 mlock + nice
  seccompProfile:
    type: Localhost
    localhostProfile: profiles/av1-encoder.json
  seLinuxOptions:
    level: "s0:c123,c456"  # MCS 标签隔离

# CMakeLists.txt - Highway 自动生成多版本目标文件
find_package(Highway REQUIRED)

hwy_add_library(av1_kernels
    SOURCES
        src/kernels/me_sad.cc
        src/kernels/tx_quant.cc
        src/kernels/lf_filter.cc
    HWY_TARGETS "AVX2;AVX512_VNNI;NEON;NEON_FP16;WASM_SIMD128"
    HWY_BASELINE "AVX2;NEON"  # 最低保障
)

target_link_libraries(av1_encoder PRIVATE av1_kernels)

// src/kernels/tx_quant.cc - 单一源码，编译出 5 个版本
#include "hwy/highway.h"
#include "hwy/ops/common.h"

namespace av1_kernels {
namespace HWY_NAMESPACE { // Highway 自动展开命名空间

void quantize_coeffs(const int32_t* input, int16_t* output, 
                     const int16_t* qmatrix, int count) {
    using namespace hwy;
    constexpr size_t N = Lanes<int16_t>(); // 自动适配 128/256/512-bit
    
    for (int i = 0; i < count; i += N) {
        auto in = Load(Int32<N>(), input + i);
        auto qm = Load(Int16<N>(), qmatrix + i);
        
        // 向量化量化：Q = round(Coeff * Scale / QMatrix)
        auto scaled = Mul(PromoteTo<Int32>(qm), in); // 扩展精度防溢出
        auto res = Round(Div(scaled, Broadcast(Int32<N>(QUANT_SHIFT))));
        
        Store(DemoteTo<Int16>(res), output + i);
    }
}

} // namespace HWY_NAMESPACE
} // namespace av1_kernels

// 运行时分发入口
HWY_EXPORT(quantize_coeffs);

// controller/rollout_controller.go
func (c *RolloutController) Reconcile(ctx context.Context, rs *appsv1.ReplicaSet) error {
    metrics := c.fetchMetrics(rs.Name)
    
    // 硬性熔断
    if metrics.JoinSuccessRate < 0.995 || metrics.FreezeRate > 0.001 {
        klog.ErrorS(nil, "L1 SLO breached, immediate rollback", "rs", rs.Name)
        return c.rollbackToStable(rs)
    }
    
    // 软性降级：关闭可选特性
    if metrics.EncodeLatencyP99 > 25_000_000 { // 25ms
        klog.Warning("Latency high, disabling CDEF/LR")
        c.patchFeatureGate(rs, "enable_cdef", "false")
        c.patchFeatureGate(rs, "enable_lr", "false")
    }
    
    // 逐步扩容
    if metrics.CPUUtil < 0.6 && metrics.MemGrowth < 50*1024*1024 {
        return c.scaleUp(rs, 1.2) // 每轮 +20%
    }
    return nil
}