本文基于昇腾CANN和昇腾NPU围绕 cann-recipes-infer 仓库的相关技术展开。LLM 推理服务有个尴尬的事实同一个请求拆成 Prefill N 个 Decode Step。Prefill 阶段 Cube 跑满了Decode 阶段 Cube 只用 30%。如果你同时有 8 个请求进去能不能一个 Prefill 完了立刻切另一个的 DecodeContinuous Batching 在 CANN 上解决了这个问题——不让 NPU 闲着等。静态 Batching 的浪费# 静态 Batching——等所有请求全到了再一起跑classStaticBatcher:def__init__(self,max_batch8):self.queue[]self.max_batchmax_batchdefadd_request(self,req):self.queue.append(req)# 等队列满了或者超时才处理iflen(self.queue)self.max_batch:return# NPU 在这里干等defexecute_batch(self):# 把 max_batch 个请求拼成一个 Batch# 问题是长的请求拖慢短的batch_inputself.pad_to_max_len(self.queue)outputsmodel.infer(batch_input)# 所有人等最慢的那个self.queue[]# 假设 8 个请求序列长度 [128, 32, 64, 256, 16, 48, 96, 192]# Batch 内 Pad 到 256 → 短的 6 个请求被强迫等了 256 的 Token 算完等批完了才跑跑的过程中 Cube 利用率像过山车——Prefill 飙到 80%Decode 掉到 30%。动态调度来一个跑一个# Continuous Batching——来了就排跑了就补classContinuousBatcher:def__init__(self,max_concurrent8):self.running_requests[]# 正在跑的请求self.waiting_queue[]# 等待的请求self.max_concurrentmax_concurrent self.schedulerfcfs# 先来先服务defadd_request(self,req):iflen(self.running_requests)self.max_concurrent:# NPU 有空位立刻开始 Prefillself.running_requests.append({req:req,phase:prefill,generated:0})else:self.waiting_queue.append(req)defstep(self):每次 Decode Step 的执行入口finished[]forrinself.running_requests:ifr[phase]prefill:# Prefill 只做 1 次然后切 Decoder[kv_cache]self.do_prefill(r[req])r[phase]decoder[generated]0elifr[phase]decode:next_tokenself.do_decode(r[kv_cache])r[req][output].append(next_token)r[generated]1ifr[generated]r[req][max_tokens]ornext_tokenEOS:finished.append(r)# 跑完的请求出队等待队列补进来forfinfinished:self.running_requests.remove(f)ifself.waiting_queue:new_reqself.waiting_queue.pop(0)self.running_requests.append({req:new_req,phase:prefill,generated:0})CANN 上的迭代级调度// Continuous Batching 在 CANN 上的调度循环// 每个 Step 重新构造输入用 Async 接口避免阻塞classContinuousBatchingEngine{std::vectorRequestSlotactive_slots;aclrtStream compute_stream;voidExecuteStep(){// Step 1构造当前 Step 的输入——只含活跃请求std::vectoruint64_tinput_ids;std::vectorintslot_indices;// 映射实际请求编号for(autoslot:active_slots){if(slot.phasePhase::PREFILL){// Prefill 的输入是整段 promptinput_ids.push_back(slot.prompt_tokens);slot.phasePhase::DECODE;}else{// Decode 只送 1 个 Tokeninput_ids.push_back(slot.last_token);}slot_indices.push_back(slot.id);}// Step 2异步启动推理——不等结果aclrtMemCpyAsync(input_device,input_ids.data(),input_ids.size()*sizeof(uint64_t),ACL_MEMCPY_HOST_TO_DEVICE,compute_stream);// 用 AsyncExecute 避免跨 Step 阻塞aclmdlExecuteAsync(model_id,input_buffer,output_buffer,compute_stream);// Step 3同步等结果aclrtSynchronizeStream(compute_stream);// Step 4取结果采样生成下一个 Tokenfor(size_t i0;iactive_slots.size();i){int*slot_outoutput_bufferi*vocab_size;intnext_tokenTopKSampling(slot_out,50,0.9);autoslotactive_slots[slot_indices[i]];slot.last_tokennext_token;slot.generated_count;}}};吞吐对比策略请求延时(P50)吞吐(req/s)Cube 利用率静态 Batch(8)3.2s2.538%Dynamic Batching2.8s3.145%Continuous Batching1.9s5.772%Continuous Batching 在 Decode 阶段把 Cube 利用率从 38% 拉到了 72%。代价是调度逻辑变复杂——每步都要重新拼输入、重排 KV Cache。CANN Runtime 的 Stream 异步调度让这个代价控制在 0.05ms 以下。参考仓库Continuous Batching 推理参考实现Runtime 异步流调度GE 图执行器
