From a71cef2d528869510107143464e49575106a51fd Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Fri, 17 Jul 2026 15:32:09 +0800 Subject: [PATCH 1/3] =?UTF-8?q?cuda:=20COLI=5FGROUP=5FASYNC=3D1=20?= =?UTF-8?q?=E2=80=94=20async=20expert-group=20issue/take=20with=20CPU/GPU?= =?UTF-8?q?=20overlap=20at=20decode=20(opt-in,=20+6-8%=20measured)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- c/backend_cuda.cu | 69 ++++++++++++++++++++++++ c/backend_cuda.h | 10 ++++ c/glm.c | 134 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 213 insertions(+) diff --git a/c/backend_cuda.cu b/c/backend_cuda.cu index 188330d9..ce39f7d9 100644 --- a/c/backend_cuda.cu +++ b/c/backend_cuda.cu @@ -29,6 +29,7 @@ typedef struct { cudaStream_t stream; void *group_desc; size_t group_desc_cap; size_t tensor_count, tensor_bytes; + int group_pending; size_t group_pending_bytes; /* async expert-group in flight (Inc.4) */ } DeviceContext; typedef struct { @@ -738,6 +739,74 @@ extern "C" int coli_cuda_expert_group(ColiCudaTensor *const *gates, return 1; } +/* ---- Async expert group (Inc.4): issue/take split of coli_cuda_expert_group ---- + * The measured cost of the sync call at decode is ~0.45 ms/call of HOST-side wait + * (stream sync + staging), vs ~0.18 ms of actual GPU work — 70% tax, paid ~5x per + * layer because a token's 8 experts scatter across devices. issue() stages and + * launches on the device stream and returns immediately; take() syncs and hands + * back the pinned result rows. One issue may be outstanding per device; moe() + * takes at each layer end, which also orders the next layer's reuse of the ctx + * scratch buffers. Small batches only (decode/spec): bigger totals keep the sync + * path with its TC variants. Numerics are the sync path's small-batch kernels, + * so greedy output is byte-identical by construction. */ +extern "C" int coli_cuda_expert_group_issue(ColiCudaTensor *const *gates, + ColiCudaTensor *const *ups, + ColiCudaTensor *const *downs, + const int *rows, int count, + const float *x) { + if (!gates || !ups || !downs || !rows || !x || count < 1 || count > 64) return 0; + ColiCudaTensor *first=gates[0]; + if (!first) return 0; + int device=first->device,D=first->I,I=first->O,total=0; + GroupDesc host[64]; + for(int c=0;cdevice!=device||u->device!=device||d->device!=device|| + g->I!=D||u->I!=D||g->O!=I||u->O!=I||d->I!=I||d->O!=D) return 0; + host[c]={g->weights,u->weights,d->weights,g->scales,u->scales,d->scales, + g->fmt,u->fmt,d->fmt,rows[c],total}; + total+=rows[c]; + } + if(total>8) return 0; /* decode-scale only */ + DeviceContext *ctx=find_ctx(device); if(!ctx||ctx->group_pending||!select_ctx(ctx)) return 0; + size_t xb=(size_t)total*D*sizeof(float), ib=(size_t)total*I*sizeof(float); + if(!reserve(&ctx->x,&ctx->x_cap,xb)||!reserve(&ctx->y,&ctx->y_cap,xb)|| + !reserve(&ctx->gate,&ctx->gate_cap,ib)||!reserve(&ctx->up,&ctx->up_cap,ib)|| + !reserve_pinned(&ctx->host_x,&ctx->host_x_cap,xb)|| + !reserve_pinned(&ctx->host_y,&ctx->host_y_cap,xb)) return 0; + std::memcpy(ctx->host_x,x,xb); + if(!cuda_ok(cudaMemcpyAsync(ctx->x,ctx->host_x,xb,cudaMemcpyHostToDevice,ctx->stream), + "expert group issue upload")) return 0; + for(int c=0;cgate+(size_t)host[c].offset*I,*u16=ctx->up+(size_t)host[c].offset*I; + float *x16=ctx->x+(size_t)host[c].offset*D,*y16=ctx->y+(size_t)host[c].offset*D; + quant_matmul<<stream>>>(g16,x16, + host[c].g,host[c].gs,host[c].gf,r,D,I,row_bytes(host[c].gf,D)); + quant_matmul<<stream>>>(u16,x16, + host[c].u,host[c].us,host[c].uf,r,D,I,row_bytes(host[c].uf,D)); + silu_mul<<<(unsigned)(((size_t)r*I+255)/256),256,0,ctx->stream>>>(g16,u16,(size_t)r*I); + quant_matmul<<stream>>>(y16,g16, + host[c].d,host[c].ds,host[c].df,r,I,D,row_bytes(host[c].df,I)); + } + if(!cuda_ok(cudaGetLastError(),"expert group issue launch")|| + !cuda_ok(cudaMemcpyAsync(ctx->host_y,ctx->y,xb,cudaMemcpyDeviceToHost,ctx->stream), + "expert group issue download")) return 0; + ctx->group_pending=1; ctx->group_pending_bytes=xb; + { std::lock_guard lock(g_group_stats_mu); + g_group_calls++; g_group_experts+=(uint64_t)count; g_group_rows+=(uint64_t)total; } + return 1; +} + +extern "C" const float *coli_cuda_expert_group_take(int device) { + DeviceContext *ctx=find_ctx(device); + if(!ctx||!ctx->group_pending) return nullptr; + ctx->group_pending=0; + if(!select_ctx(ctx)) return nullptr; + if(!cuda_ok(cudaStreamSynchronize(ctx->stream),"expert group take")) return nullptr; + return ctx->host_y; +} + extern "C" int coli_cuda_attention_absorb(ColiCudaTensor *w,float *ctx,const float *q, const float *latent,const float *rope,int H,int Q, diff --git a/c/backend_cuda.h b/c/backend_cuda.h index edddbfe0..cca31a1c 100644 --- a/c/backend_cuda.h +++ b/c/backend_cuda.h @@ -67,6 +67,16 @@ COLI_CUDA_DLLEXPORT int coli_cuda_shared_mlp_w4a16(ColiCudaTensor *gate, ColiCud /* Packed group of same-shaped experts. Inputs and outputs contain sum(rows) * consecutive [D] rows in call order. */ +/* Async issue/take split of the group call below (Inc.4): issue launches on the + * device stream and returns; take syncs and returns the pinned result rows (valid + * until the next issue on that device). Small totals only (<=8 rows); one + * outstanding issue per device. */ +COLI_CUDA_DLLEXPORT int coli_cuda_expert_group_issue(ColiCudaTensor *const *gates, + ColiCudaTensor *const *ups, + ColiCudaTensor *const *downs, + const int *rows, int count, const float *x); +COLI_CUDA_DLLEXPORT const float *coli_cuda_expert_group_take(int device); + COLI_CUDA_DLLEXPORT int coli_cuda_expert_group(ColiCudaTensor *const *gates, ColiCudaTensor *const *ups, ColiCudaTensor *const *downs, diff --git a/c/glm.c b/c/glm.c index 70da3201..f3c9071c 100644 --- a/c/glm.c +++ b/c/glm.c @@ -3245,6 +3245,11 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int } #ifdef COLI_CUDA ESlot *group_e[64]; int group_n[64]; int ngroup=0; + /* Inc.4 overlap stash: pass-1 packing kept for the take phase after the CPU loop */ + ESlot *eg_e[64]; int eg_n[64], eg_row[64][4], eg_npg=0; float eg_w[64][4]; + int dev_nc0[COLI_CUDA_MAX_DEVICES], dev_off0[COLI_CUDA_MAX_DEVICES], + dev_total0[COLI_CUDA_MAX_DEVICES], dev_which0[COLI_CUDA_MAX_DEVICES][64]; + memset(dev_nc0,0,sizeof(dev_nc0)); (void)eg_npg; (void)dev_total0; (void)dev_off0; #endif #ifdef COLI_METAL if(g_metal_enabled){ @@ -3273,9 +3278,82 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int free(mxg); free(mrows); free(mrw); } #undef MB_BUILD +#endif +#ifdef COLI_CUDA + /* Inc.4 pass 1: collect the VRAM-resident experts' groups and ISSUE them async + * BEFORE the CPU loop below, so the GPU computes its share while the CPU works + * through the RAM-tier/miss rows — t_emm becomes max(cpu, gpu) instead of the + * sum. Only resident experts are collected (misses are never cuda_eligible), so + * no pipe_wait is needed here; the CPU loop keeps its own waits. Any issue + * failure drops the layer back to the collect-in-loop + sync-group path. */ + int early_issued=0, done_j[64]={0}; + { + static int g_group_async2=-1; + if(g_group_async2<0) g_group_async2=getenv("COLI_GROUP_ASYNC")?atoi(getenv("COLI_GROUP_ASYNC")):0; + if(!metal_done && g_group_async2 && group_enabled && S<=4 && g_cuda_enabled && + g_cuda_ndev>0 && !omp_in_parallel()){ + ESlot *pg_e[64]; int pg_n[64], pg_j[64], npg=0; + int prow[64][4]; float pw[64][4]; + for(int j=0;jg.cuda_eligible&&e->u.cuda_eligible&&e->d.cuda_eligible)) continue; + int nr=0; + for(int s=0;sg.cuda_device==g_cuda_devices[di]) pd_total[di]+=pg_n[q]; + for(int di=1;dig.cuda_device==device){ + int nc=pd_nc[di]++; ESlot *e=pg_e[q]; + pd_g[di][nc]=e->g.cuda; pd_u[di][nc]=e->u.cuda; pd_d[di][nc]=e->d.cuda; + pd_rows[di][nc]=pg_n[q]; pd_which[di][nc]=q; + for(int r=0;rt_emm+=now_s()-tg0; + for(int q=0;qgpu_expert_calls++; + } + } else { + for(int di=0;di=1 routing invariant. @@ -3322,6 +3400,35 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int m->cpu_expert_rows+=(uint64_t)nr;} } #ifdef COLI_CUDA + /* Inc.4 take phase: the CPU loop above ran while the GPU computed the issued + * groups — collect them now. A failed device recomputes its experts on the CPU + * (expert_host_ensure reloads slabs released by CUDA_RELEASE_HOST). */ + if(early_issued){ + double tg1=now_s(); + for(int di=0;dig,&e->u,nr); + for(int64_t z=0;z<(int64_t)nr*I;z++) gg[z]=siluf(gg[z])*uu[z]; + matmul_qt(hh,gg,&e->d,nr); + for(int r=0;rt_emm+=now_s()-tg1; + } ColiCudaTensor *dev_g[COLI_CUDA_MAX_DEVICES][64],*dev_u[COLI_CUDA_MAX_DEVICES][64]; ColiCudaTensor *dev_d[COLI_CUDA_MAX_DEVICES][64]; int dev_rows[COLI_CUDA_MAX_DEVICES][64],dev_which[COLI_CUDA_MAX_DEVICES][64]; @@ -3343,6 +3450,33 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int } } double tg=now_s(); + /* Inc.4: at decode scale, issue every device's group WITHOUT syncing, then take + * them all — one stream sync per device per layer instead of a full staged + * round-trip per call (measured: ~70% of the sync call is host-side wait). + * Any issue failure drains what was issued and the whole layer falls back to + * the sync path below, which recomputes from group_x (idempotent). */ + int async_done=0; + static int g_group_async=-1; + if(g_group_async<0) g_group_async=getenv("COLI_GROUP_ASYNC")?atoi(getenv("COLI_GROUP_ASYNC")):0; + if(g_group_async && S<=4 && g_cuda_ndev>0){ + int issued[COLI_CUDA_MAX_DEVICES]={0}, all=1; + for(int di=0;di1) schedule(static) for(int di=0;di Date: Fri, 17 Jul 2026 15:41:38 +0800 Subject: [PATCH 2/3] cuda: cache layernorm weights on-device in pipe_layer_sparse (kill 152 sync H2D/token) + overlap-window profiling counters --- c/glm.c | 32 +++++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 7 deletions(-) diff --git a/c/glm.c b/c/glm.c index f3c9071c..80365c48 100644 --- a/c/glm.c +++ b/c/glm.c @@ -178,6 +178,7 @@ typedef struct { KVState *kv; ESlot **ecache; int *ecn; int ecap; /* LRU expert per-layer */ float **kv_dev_L, **kv_dev_R; int *kv_dev_valid; /* ombra KV su device (decode) */ + float **ln_dev; /* in_ln/post_ln cached on device: [layer*2+{0,1}] (Inc.4) */ ESlot ws[64]; /* working set del layer corrente (load paralleli) */ ESlot **pin; int *npin; /* HOT-STORE: expert pinnati in RAM (mai evicted) */ uint32_t **eusage; /* contatori persistenti (per STATS/PIN) */ @@ -249,6 +250,7 @@ static int qt_cuda_update(QT *t){ t->fmt==1?(const void*)t->q8:(const void*)t->q4; return coli_cuda_tensor_update(t->cuda,weights,t->s); } +static double g_ovl_issue,g_ovl_cpu,g_ovl_take,g_ovl_mark; /* Inc.4 overlap-window split (OVL report) */ static void cuda_stats_print(void){ size_t n=0,b=0; coli_cuda_stats(-1,&n,&b); fprintf(stderr,"[CUDA] resident set: %zu tensors, %.2f GB VRAM\n",n,b/1e9); @@ -264,6 +266,9 @@ static void cuda_stats_print(void){ getenv("COLI_CUDA_PROFILE")?"; timing sotto":""); if(calls&&getenv("COLI_CUDA_PROFILE")) fprintf(stderr, "[CUDA] expert groups timing: H2D %.1f ms | kernel %.1f ms | D2H %.1f ms\n",h2d,kernel,d2h); + if(g_ovl_issue+g_ovl_cpu+g_ovl_take>0) fprintf(stderr, + "[CUDA] overlap window: pack+issue %.2fs | cpu-rows %.2fs | take(sync+acc) %.2fs\n", + g_ovl_issue,g_ovl_cpu,g_ovl_take); } static int parse_cuda_devices(const char *list, int *out){ if(!list||!*list) return 0; @@ -3326,10 +3331,11 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int for(int di=0;dit_emm+=now_s()-tg1; + m->t_emm+=now_s()-tg1; g_ovl_take+=now_s()-tg1; } ColiCudaTensor *dev_g[COLI_CUDA_MAX_DEVICES][64],*dev_u[COLI_CUDA_MAX_DEVICES][64]; ColiCudaTensor *dev_d[COLI_CUDA_MAX_DEVICES][64]; @@ -3923,17 +3930,28 @@ static int pipe_layer_sparse(Model *m, Layer *l, int li, float *x_dev, int S, in if(!l->sh_gate.cuda_eligible||!l->sh_up.cuda_eligible||!l->sh_down.cuda_eligible|| !qt_cuda_upload(&l->sh_gate)||!qt_cuda_upload(&l->sh_up)||!qt_cuda_upload(&l->sh_down)|| l->sh_gate.cuda_device!=dev||l->sh_up.cuda_device!=dev||l->sh_down.cuda_device!=dev) return 0; - float *w_in =coli_cuda_pipe_scratch(dev,8,(size_t)D*4); - float *w_post=coli_cuda_pipe_scratch(dev,9,(size_t)D*4); + /* Inc.4: the layernorm weights are constants — upload once per layer and keep them + * on the layer's device, instead of two synchronous 24 KB uploads per layer per + * token (152 sync H2D/token measured on the profile). */ + if(!m->ln_dev) m->ln_dev=calloc((size_t)(c->n_layers+1)*2,sizeof(float*)); + float *w_in=m->ln_dev[(size_t)li*2], *w_post=m->ln_dev[(size_t)li*2+1]; + if(!w_in){ + w_in=coli_cuda_pipe_alloc(dev,(size_t)D*4); + if(!w_in||!coli_cuda_pipe_upload(dev,w_in,l->in_ln,(size_t)D*4)) return 0; + m->ln_dev[(size_t)li*2]=w_in; + } + if(!w_post){ + w_post=coli_cuda_pipe_alloc(dev,(size_t)D*4); + if(!w_post||!coli_cuda_pipe_upload(dev,w_post,l->post_ln,(size_t)D*4)) return 0; + m->ln_dev[(size_t)li*2+1]=w_post; + } float *nrm_d=coli_cuda_pipe_scratch(dev,10,xb); float *y_d =coli_cuda_pipe_scratch(dev,11,xb); float *sg_d =coli_cuda_pipe_scratch(dev,12,(size_t)S*sI*4); float *su_d =coli_cuda_pipe_scratch(dev,13,(size_t)S*sI*4); float *snap =coli_cuda_pipe_scratch(dev,14,xb); - if(!w_in||!w_post||!nrm_d||!y_d||!sg_d||!su_d||!snap) return 0; + if(!nrm_d||!y_d||!sg_d||!su_d||!snap) return 0; if(!coli_cuda_pipe_peer_copy(dev,snap,dev,x_dev,xb)) return 0; /* snapshot per il fallback */ - if(!coli_cuda_pipe_upload(dev,w_in,l->in_ln,(size_t)D*4)|| - !coli_cuda_pipe_upload(dev,w_post,l->post_ln,(size_t)D*4)) return 0; double ta=now_s(); if(!coli_cuda_pipe_rmsnorm(dev,nrm_d,x_dev,w_in,S,D,c->eps)) return 0; /* DSA: i layer con indexer FULL cachano k_idx dalla nrm pre-attention (CPU, piccolo) */ From 5978d050380799d9c779a11455eef38ba299b63e Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Sat, 18 Jul 2026 18:44:27 +0800 Subject: [PATCH 3/3] fix Windows async expert loader --- c/backend_loader.c | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/c/backend_loader.c b/c/backend_loader.c index bbcb8ca0..1d6608b7 100644 --- a/c/backend_loader.c +++ b/c/backend_loader.c @@ -41,6 +41,11 @@ typedef int (*fn_expert_mlp)(ColiCudaTensor *gate, ColiCudaTensor *up typedef int (*fn_expert_group)(ColiCudaTensor *const *gates, ColiCudaTensor *const *ups, ColiCudaTensor *const *downs, const int *rows, int count, float *y, const float *x); +typedef int (*fn_expert_group_issue)(ColiCudaTensor *const *gates, + ColiCudaTensor *const *ups, + ColiCudaTensor *const *downs, + const int *rows, int count, const float *x); +typedef const float * (*fn_expert_group_take)(int device); typedef int (*fn_attention_absorb)(ColiCudaTensor *kv_b, float *ctx, const float *q, const float *latent, const float *rope, int H, int Q, int R, int V, int K, int T, float attention_scale); @@ -99,6 +104,8 @@ static struct { fn_group_stats group_stats; fn_expert_mlp expert_mlp; fn_expert_group expert_group; + fn_expert_group_issue expert_group_issue; + fn_expert_group_take expert_group_take; fn_attention_absorb attention_absorb; fn_tensor_upload tensor_upload; fn_matmul matmul; @@ -193,6 +200,8 @@ static int coli_cuda_load(void){ RESOLVE(group_stats, fn_group_stats) RESOLVE(expert_mlp, fn_expert_mlp) RESOLVE(expert_group, fn_expert_group) + RESOLVE(expert_group_issue, fn_expert_group_issue) + RESOLVE(expert_group_take, fn_expert_group_take) RESOLVE(attention_absorb, fn_attention_absorb) RESOLVE(tensor_upload, fn_tensor_upload) RESOLVE(matmul, fn_matmul) @@ -288,6 +297,19 @@ int coli_cuda_expert_group(ColiCudaTensor *const *gates, ColiCudaTensor *const * return g_cuda.expert_group(gates, ups, downs, rows, count, y, x); } +int coli_cuda_expert_group_issue(ColiCudaTensor *const *gates, + ColiCudaTensor *const *ups, + ColiCudaTensor *const *downs, + const int *rows, int count, const float *x){ + if(!g_cuda.available) return 0; + return g_cuda.expert_group_issue(gates, ups, downs, rows, count, x); +} + +const float *coli_cuda_expert_group_take(int device){ + if(!g_cuda.available) return NULL; + return g_cuda.expert_group_take(device); +} + int coli_cuda_attention_absorb(ColiCudaTensor *kv_b, float *ctx, const float *q, const float *latent, const float *rope, int H, int Q, int R, int V, int K, int T, float attention_scale){