低延迟GPU数据包处理

// CPU实现
void processPacketCPU(const Packet* packet, PacketResult* result, int packetId) {
    // 核心数据包处理逻辑
}
 
// GPU实现
__device__ void processPacketGPU(const Packet* packet, PacketResult* result, int packetId) {
    // 相同的核心逻辑，但作为设备函数
}

// 为数据包缓冲区分配固定内存
cudaHostAlloc(&h_packets, packet_buffer_size, cudaHostAllocDefault);

// 分配零拷贝内存
cudaHostAlloc(&h_packets, packet_buffer_size, cudaHostAllocMapped);
cudaHostGetDevicePointer(&d_packets, h_packets, 0);

// 创建用于流水线的流
cudaStream_t streams[NUM_STREAMS];
for (int i = 0; i < NUM_STREAMS; i++) {
    cudaStreamCreate(&streams[i]);
}
 
// 流水线执行
for (int i = 0; i < NUM_BATCHES; i++) {
    int stream_idx = i % NUM_STREAMS;
    // 异步传输批次i到GPU
    cudaMemcpyAsync(d_packets[i], h_packets[i], batch_size, 
                    cudaMemcpyHostToDevice, streams[stream_idx]);
    // 处理批次i
    processPacketsKernel<<<grid, block, 0, streams[stream_idx]>>>(
        d_packets[i], d_results[i], batch_size);
    // 异步将结果传回
    cudaMemcpyAsync(h_results[i], d_results[i], result_size,
                   cudaMemcpyDeviceToHost, streams[stream_idx]);
}

__global__ void persistentKernel(volatile int* work_queue, volatile int* queue_size,
                                 PacketBatch* batches) {
    while (true) {
        // 检查新工作
        if (threadIdx.x == 0 && blockIdx.x == 0) {
            // 等待新批次（自旋等待或睡眠）
            while (*queue_size == 0);
            // 获取批次索引
            batch_idx = atomicAdd((int*)queue_size, -1);
        }
        // 使用共享内存向所有线程广播batch_idx
        __shared__ int s_batch_idx;
        if (threadIdx.x == 0) s_batch_idx = batch_idx;
        __syncthreads();
        
        // 使用我们的核心函数处理指定批次的数据包
        processPacketGPU(&batches[s_batch_idx].packets[tid], &results[tid], tid);
        
        // 发出完成信号
        if (threadIdx.x == 0 && blockIdx.x == 0) {
            batches[s_batch_idx].status = COMPLETED;
        }
    }
}

// 创建并捕获CUDA图
cudaGraph_t graph;
cudaGraphExec_t graphExec;
 
// 将操作捕获到图中
cudaStreamBeginCapture(stream, cudaStreamCaptureModeGlobal);
for (int i = 0; i < PIPELINE_DEPTH; i++) {
    cudaMemcpyAsync(...); // 复制输入
    kernel<<<...>>>(...);  // 处理
    cudaMemcpyAsync(...); // 复制输出
}
cudaStreamEndCapture(stream, &graph);
 
// 将图编译为可执行文件
cudaGraphInstantiate(&graphExec, graph, NULL, NULL, 0);
 
// 使用新数据重复执行图
for (int batch = 0; batch < NUM_BATCHES; batch++) {
    updateGraphInputs(batch); // 更新内存地址
    cudaGraphLaunch(graphExec, stream);
}