【架构设计】C++ 服务端算子化：从单体逻辑到可组合计算图

// ── 数据流：节点、依赖、显式数据绑定 ──
ADD_NODE(A);
ADD_NODE(B);
ADD_NODE(C);
ADD_NODE(D);

ADD_EDGE(A, B);
ADD_EDGE(A, C);
ADD_EDGE(B, D);
ADD_EDGE(C, D);

ASSIGN_DATA(A, data_a, B, data_b);
ASSIGN_DATA(A, data_a, C, data_c);
ASSIGN_DATA(B, data_b, D, data_d);
ASSIGN_DATA(C, data_c, D, data_d);

// ── 控制流：与数据边正交，Runtime 解释执行 ──
DISABLE_NODE(B);                 // 热摘除：不执行 B，注入兜底值

// ── 各算子维护自己的 Schema（算子内可见，算子间不可见）──
struct SchemaA {           // A 的输出
    Span<ItemId> ids;
    Span<float> recall_scores;
    Span<uint8_t> sources;
};

struct SchemaB {           // B 的输入
    Span<ItemId> ids;
    Span<float> rank_features;
    Span<uint8_t> sources;
};

// ── 映射表：声明 A.data_a → B.data_b 的字段对齐规则 ──
BEGIN_MAPPING(A, data_a, B, data_b)
    MAP_DIRECT(ids, ids)
    MAP_RENAME(recall_scores, rank_features)
    MAP_TRANSFORM(sources, sources, CastSource)   // 可选变换函数
    MAP_DEFAULT(rank_features, 0.0f)              // A 未产出时兜底
END_MAPPING()

// ── 与 §3 建图 DSL 组合 ──
ADD_NODE(A);
ADD_NODE(B);
ADD_EDGE(A, B);
ASSIGN_DATA(A, data_a, B, data_b);
BIND_MAPPING(A, data_a, B, data_b, mapping_a2b);  // 边绑定映射表

// ── Runtime 在边上执行对齐（宏展开示意）──
Status AlignAndRun(RequestContext& ctx) {
    const SchemaA& raw = slots_.Get<A>("data_a");
    SchemaB& aligned = slots_.Alloc<B>("data_b");

    for (const auto& rule : mapping_a2b) {
        if (!rule.Apply(raw, &aligned)) {
            if (!rule.has_default) return Status::kError;
            rule.FillDefault(&aligned);
        }
    }
    return ops_["B"]->Execute(ctx, aligned, &slots_["data_c"]);
}

class Arena {
public:
    void* Allocate(size_t size, size_t align = alignof(std::max_align_t));
    template <typename T, typename... Args>
    T* New(Args&&... args);
    void Reset();  // 请求结束调用，O(1) 批量回收
};

struct ItemList {
    Item* items;   // 来自 ctx.arena
    size_t size;
};

// 算子输出写入 arena；下游只读或原地修改，生命周期 <= 本请求
Status RecallOp::Execute(RequestContext& ctx,
                         const Query& in,
                         ItemList* out) {
    out->items = ctx.arena->NewArray<Item>(in.limit * 2);
    out->size = Fetch(ctx, in, out->items);
    return Status::kOk;
}

// 请求入口
Status Handle(RequestContext& ctx, const Query& q, Response* resp) {
    Arena arena;
    ctx.arena = &arena;
    auto status = runtime.Execute(ctx, q, resp);
    arena.Reset();  // 500ms 后统一回收，中间算子无需 delete
    return status;
}

enum class Status { kOk, kSkip, kDegraded, kError };

class Operator {
public:
    virtual ~Operator() = default;
    virtual Status Execute(RequestContext& ctx,
                         const void* in, void* out) = 0;
};

// Runtime 统一包装——业务算子不继承、不调用这部分
class Runtime {
public:
    Status RunNode(int node_id, RequestContext& ctx) {
        auto& node = graph_.Node(node_id);
        ScopedSpan span(ctx.trace, node.name);       // 自动 trace
        ScopedTimer timer(&node_metrics_[node_id]);  // 自动耗时

        if (!policy_.ShouldRun(node_id)) {
            node_metrics_[node_id].skip_cnt++;
            return InjectFallback(node_id, ctx);
        }

        Status st = ops_[node_id]->Execute(ctx,
                                           LoadInput(node_id, ctx),
                                           AllocOutput(node_id, ctx));
        node_metrics_[node_id].Record(st);
        return st;
    }

private:
    GraphDef graph_;
    std::vector<Operator*> ops_;
    std::vector<NodeMetrics> node_metrics_;
    ControlPolicy policy_;
};

struct NodeMetrics {
    Counter ok, skip, degraded, error;
    Histogram latency_us;
    Histogram input_size, output_size;

    void Record(Status st) {
        switch (st) {
            case Status::kOk:       ok.Inc(); break;
            case Status::kSkip:     skip.Inc(); break;
            case Status::kDegraded: degraded.Inc(); break;
            default:                error.Inc(); break;
        }
    }
};

// 500ms 链路上，定位 tail latency 通常看 P99 最慢节点
// Runtime 按 node_id 上报，无需业务侧逐段埋点
void ExportMetrics(const std::vector<NodeMetrics>& m, const GraphDef& g) {
    for (int i = 0; i < g.size(); ++i) {
        Report(g.Node(i).name, m[i]);  // Recall / Rank / ReRank 各一条
    }
}