千万级直播系统底层技术剖析：从DPDK网络到GPU转码的全栈性能优化

// DPDK收包循环（运行在独立核上）
static int lcore_rx_loop(void *arg) {
    struct lcore_params *p = (struct lcore_params*)arg;
    struct rte_mbuf *bufs[BURST_SIZE];
    while (!force_quit) {
        uint16_t nb = rte_eth_rx_burst(p->port_id, p->queue_id, bufs, BURST_SIZE);
        if (nb) {
            for (int i = 0; i < nb; ++i) {
                struct rte_mbuf *m = bufs[i];
                char *pkt_data = rte_pktmbuf_mtod(m, char*);
                if (is_rtmp_packet(pkt_data)) {
                    dispatch_to_rtmp_session(m);
                } else {
                    rte_pktmbuf_free(m);
                }
            }
        }
    }
    return 0;
}

struct tcp_stream {
    uint32_t src_ip, dst_ip;
    uint16_t src_port, dst_port;
    uint32_t snd_nxt, rcv_nxt;
    uint8_t state;               // SYN_RCVD, ESTABLISHED, FIN_WAIT
    RingBuffer* rx_ring;         // 无锁环形缓冲区
    uint64_t last_active_ts;
    // 柔性数组存储重传队列
};

class SlabAllocator {
public:
    void* allocate(size_t size) {
        int idx = calc_class(size);
        auto& free_list = slabs_[idx].free_list;
        if (!free_list.empty()) {
            void* ptr = free_list.back();
            free_list.pop_back();
            return ptr;
        }
        return allocate_new_slab(idx);
    }
    void deallocate(void* ptr, size_t size) {
        int idx = calc_class(size);
        slabs_[idx].free_list.push_back(ptr);
    }
private:
    struct Slab {
        std::vector<void*> free_list;
        char* base;  // mmap映射的大页内存
    };
    Slab slabs_[4];
};

enum ParseState {
    PARSE_BASIC_HEADER,
    PARSE_EXT_TIMESTAMP,
    PARSE_PAYLOAD
};

class RtmpDecoder {
public:
    bool feed(const uint8_t* data, size_t len) {
        while (len > 0) {
            switch (state_) {
            case PARSE_BASIC_HEADER: {
                uint8_t first = *data++;
                --len;
                fmt_ = (first >> 6) & 0x03;
                csid_ = first & 0x3F;
                if (csid_ == 0) { csid_ = *data++ + 64; --len; }
                else if (csid_ == 1) { csid_ = *(uint16_t*)data + 64; data += 2; len -= 2; }
                state_ = (fmt_ == 3) ? PARSE_PAYLOAD : PARSE_EXT_TIMESTAMP;
                break;
            }
            case PARSE_EXT_TIMESTAMP: {
                if (len < 4) return false;
                ext_ts_ = rte_be_to_cpu_32(*(uint32_t*)data);
                data += 4; len -= 4;
                state_ = PARSE_PAYLOAD;
                break;
            }
            case PARSE_PAYLOAD: {
                return assemble_chunk(data, len);
            }
            }
        }
        return true;
    }
};

int send_flv_tag(uint32_t stream_id, const uint8_t* payload, size_t payload_len,
                 uint8_t tag_type, uint32_t timestamp) {
    struct rte_mbuf *m = rte_pktmbuf_alloc(pool_);
    if (!m) return -ENOMEM;
    uint8_t* tag_hdr = rte_pktmbuf_prepend(m, 11);
    tag_hdr[0] = tag_type;
    tag_hdr[1] = (payload_len >> 16) & 0xFF;
    tag_hdr[2] = (payload_len >> 8) & 0xFF;
    tag_hdr[3] = payload_len & 0xFF;
    tag_hdr[4] = (timestamp >> 16) & 0xFF;
    tag_hdr[5] = (timestamp >> 8) & 0xFF;
    tag_hdr[6] = timestamp & 0xFF;
    tag_hdr[7] = (timestamp >> 24) & 0xFF;
    tag_hdr[8] = 0; tag_hdr[9] = 0; tag_hdr[10] = 0;
    uint8_t* dst = rte_pktmbuf_append(m, payload_len);
    rte_memcpy(dst, payload, payload_len);
    return rte_eth_tx_burst(port_id, queue_id, &m, 1);
}

class TranscodePipeline {
public:
    bool process_frame(AVFrame* src_frame) {
        cudaMemcpyAsync(gpu_frame_, src_frame->data[0], ..., cudaMemcpyHostToDevice);
        cudaLaunchKernel(yuv_to_nv12, ...);
        npp::resize(gpu_nv12, gpu_scaled, width, height, target_w, target_h);
        AVCodecContext* enc_ctx = get_encoder(target_resolution);
        AVPacket* pkt = av_packet_alloc();
        avcodec_send_frame(enc_ctx, gpu_scaled_as_avframe());
        while (avcodec_receive_packet(enc_ctx, pkt) == 0) {
            rtmp_sender_->send_video(pkt->data, pkt->size, pts);
        }
        return true;
    }
};

class EdgeScheduler {
public:
    std::string select_edge(const std::string& room_id, uint32_t client_ip) {
        uint32_t key = hash(room_id) ^ (client_ip & 0xFFFFFF00);
        auto it = ring_.upper_bound(key);
        if (it == ring_.end()) it = ring_.begin();
        return it->second;
    }
private:
    std::map<uint32_t, std::string> ring_;  // 30000个虚拟节点
};

bool should_drop_packet(const Packet& pkt, const Connection& conn) {
    if (get_system_load() < 0.7f) return false;
    if (conn.vip_level > 0) return false;
    if (pkt.frame_type == FRAME_B) return true;
    if (pkt.frame_type == FRAME_P && random() % 100 < 30) return true;
    return false;
}