P2-02 · mTCP 源码架构：core/tcp_in/eventpoll 三个入口

对应路线图：Phase 2 (Day 50-91) ｜ DPDK 24.11.6 LTS ｜内核对照 v6.6
代码出处：source/mtcp/mtcp/src/，引用均标 文件:行号
上一篇：P2-01 mTCP 论文精读｜下一篇：P2-03 TCP 状态机对照

一句话定位

mTCP 整个栈可以压缩成一句话：一个 while 循环扮演"用户态软中断"，每轮做九件事。看懂 RunMainLoop，就懂了 mTCP 的全部调度哲学。

注：本仓库的 mTCP 是 fork 版本，入口叫 RunMainLoop（core.c:762），由 MTCPRunThread（core.c:1153）拉起；上游旧版叫 mtcp_run_event_loop，职责相同。

入口一：core.c 主循环（RunMainLoop 九步）

RunMainLoop 单轮时序

static void
RunMainLoop(struct mtcp_thread_context *ctx)        // core.c:762
{
    mtcp_manager_t mtcp = ctx->mtcp_manager;                // core.c:764
    // ...
    while ((!ctx->done || mtcp->flow_cnt) && !ctx->exit) {   // core.c:776
        // ...
        gettimeofday(&cur_ts, NULL);
        mtcp->cur_ts = TIMEVAL_TO_TS(&cur_ts);          // ① 取时间戳 :781-783

        for (rx_inf = 0; rx_inf < CONFIG.eths_num; rx_inf++) {
            recv_cnt = mtcp->iom->recv_pkts(ctx, rx_inf);    // ② 批量收包 :789
            for (i = 0; i < recv_cnt; i++) {
                pktbuf = mtcp->iom->get_rptr(mtcp->ctx, rx_inf, i, &len);
                if (pktbuf != NULL)
                    ProcessPacket(mtcp, rx_inf, ts, pktbuf, len);  // ③ :795
            }
        }

        if (mtcp->flow_cnt > 0) {
            CheckRtmTimeout(mtcp, ts, CONFIG.max_concurrency);   // ④ 定时器 :824
            CheckTimewaitExpire(mtcp, ts, CONFIG.max_concurrency);  // :825
            // ...
        }

        if (mtcp->ep) {
            FlushEpollEvents(mtcp, ts);                 // ⑤ 批量事件 :834
        }

        if (mtcp->flow_cnt > 0) {
            HandleApplicationCalls(mtcp, ts);           // ⑥ 应用调用 :839
        }

        WritePacketsToChunks(mtcp, ts);                 // ⑦ 写发送块 :842

        for (tx_inf = 0; tx_inf < CONFIG.eths_num; tx_inf++)
            mtcp->iom->send_pkts(ctx, tx_inf);          // ⑧ 批量发送 :847
        mtcp->iom->select(ctx);                         // ⑨ 空闲才睡 :860
    }
}

两个最关键的细节：① ④⑤⑥⑦的顺序不能乱——先查超时（重传/回收），再刷事件（应用才能看到新数据），再消费应用攒下的调用（connect/send/close），最后才写发送块，保证应用视角一致性；② ⑨ 的 select 是"没活干才睡"——有包收就不睡（这正是 P1-05 说的"把中断换成轮询"），但流空闲时也让出 CPU，避免空转烧核。

入口二：tcp_in.c 状态机（每包都过这里）

ProcessPacket（eth_in.c:9）按以太类型分发：ETH_P_IP → ProcessIPv4Packet，最终落到 ProcessTCPPacket（tcp_in.c:1205）：

int
ProcessTCPPacket(mtcp_manager_t mtcp,
     uint32_t cur_ts, const int ifidx, const struct iphdr *iph, int ip_len)
{                                                       // tcp_in.c:1205
    struct tcphdr* tcph = (struct tcphdr *) ((u_char *)iph + (iph->ihl << 2));
    uint8_t *payload    = (uint8_t *)tcph + (tcph->doff << 2);
    int payloadlen = ip_len - (payload - (u_char *)iph);
    // ...
    s_stream.saddr = iph->daddr;                        // 注意：4-tuple 是"本端视角"
    s_stream.sport = tcph->dest;                        // 与网卡方向相反
    s_stream.daddr = iph->saddr;
    s_stream.dport = tcph->source;                      // tcp_in.c:1247-1250

    if (!(cur_stream = StreamHTSearch(mtcp->tcp_flow_table, &s_stream))) {
        /* not found in flow table */
        cur_stream = CreateNewFlowHTEntry(mtcp, cur_ts, iph, ip_len, tcph,
                seq, ack_seq, payloadlen, window);    // tcp_in.c:1252-1258
    }

    /* Validate sequence. if not valid, ignore the packet */
    if (cur_stream->state > TCP_ST_SYN_RCVD) {
        ret = ValidateSequence(mtcp, cur_stream,
                cur_ts, tcph, seq, ack_seq, payloadlen);   // tcp_in.c:1261-1263
    }
    // ...
    switch (cur_stream->state) {                        // tcp_in.c:1298 状态分发
    case TCP_ST_LISTEN:
        Handle_TCP_ST_LISTEN(mtcp, cur_ts, cur_stream, tcph);
        break;
    case TCP_ST_ESTABLISHED:
        Handle_TCP_ST_ESTABLISHED(mtcp, cur_ts, cur_stream, tcph,
                seq, ack_seq, payload, payloadlen, window);
        break;
    // ... 共 11 个状态分支，逐状态对比见 P2-03
    }
}

流表是每核私有的拉链 hash：StreamHTSearch 就是算 hash 下标 + TAILQ_FOREACH 遍历桶（fhash.c:109-126），hash 函数 HashFlow 对 12 字节 4-tuple 做 Jenkins 风格混合后 & (NUM_BINS_FLOWS - 1)（tcp_stream.c:82-94，13 万桶）。

入口三：eventpoll.c 事件桥（应用线程的接口）

应用线程不碰包路径，只调 mtcp_epoll_wait（eventpoll.c:363）等事件。栈线程侧只做入队 AddEpollEvent（eventpoll.c:569）：

inline int
AddEpollEvent(struct mtcp_epoll *ep,
        int queue_type, socket_map_t socket, uint32_t event)
{                                                       // eventpoll.c:569
    if (socket->events & event) {
        return 0;               /* 同一事件只入队一次 */
    }
    if (queue_type == MTCP_EVENT_QUEUE) {
        eq = ep->mtcp_queue;    /* 栈线程 → mtcp_queue */
    } else if (queue_type == USR_EVENT_QUEUE) {
        eq = ep->usr_queue;     /* 应用线程 → usr_queue */
    }
    index = eq->end++;
    socket->events |= event;
    eq->events[index].sockid = socket->id;              // eventpoll.c:604-609
}

状态机在关键跃迁点调 RaiseReadEvent/RaiseWriteEvent/RaiseErrorEvent（tcp_stream.c:125/146/198），它们内部就是 AddEpollEvent(..., MTCP_EVENT_QUEUE, ...)。然后主循环 ⑤ 统一 flush（core.c:461-472 把 mtcp_queue 搬进 usr_queue）+ 单次 pthread_cond_signal（core.c:483）。这是"批量事件"的完整闭环。

🐧 内核对照

环节	Linux v6.6	mTCP
收包循环	软中断 `net_rx_action`（budget 限流）→ tcp_v4_rcv（tcp_ipv4.c:1999）	`RunMainLoop` 每轮 burst（core.c:789），无预算
连接查找	`__inet_lookup_established` 全局 ehash（tcp_ipv4.c:1803）	每核私有 fhash（fhash.c:109）
空闲行为	NAPI 空闲后重新开中断	`iom->select` 睡眠（core.c:860）

面试 3 问

RunMainLoop 九步顺序为什么是"超时→事件→应用调用→发送"？ → 超时先处理保证重传/回收优先；事件在应用调用之前刷，应用读到最新状态；发送放最后让本轮的 ACK/重传/新数据一起进发送块（core.c:824-847 行序）。
ProcessTCPPacket 的 4-tuple 为什么是"反的"？ → 流表 key 是本端视角：收到的包 saddr=对端，所以存表时 saddr 取 iph->daddr（tcp_in.c:1247-1250）。
mTCP 是纯轮询吗？ → 不是：有包收就转（core.c:776 循环），无流无包时 iom->select 让出 CPU（core.c:860），是"轮询 + 可选睡眠"的混合模型。

🔧 动手实验

在 source/mtcp/ 按 autotools 流程构建（README.md/README.config 有 DPDK 路径配置说明），跑 apps/example/epserver -f epserver.conf + epwget（配置模板 config/sample_mtcp.conf：io = dpdk、port = dpdk0、max_concurrency = 10000）。用 gdb 在 RunMainLoop 每步各打一个断点，统计 1 秒内各断点命中次数——"九个动作的负载画像"。数据落点 perf-data/P2/02-eventloop.md。

下一篇：P2-03 · 🐧 对照：mTCP 与内核 TCP 状态机逐状态对比