diff --git a/.gitignore b/.gitignore index 203a322..ce6a80c 100644 --- a/.gitignore +++ b/.gitignore @@ -60,3 +60,6 @@ c/backend_metal_test c/tests/test_tier c/mio_env/ c/bench/ +c/tests/test_decode_batch +c/tests/test_i4_acc512 +c/tests/test_idot diff --git a/c/backend_cuda.cu b/c/backend_cuda.cu index 3bb853b..9ce142f 100644 --- a/c/backend_cuda.cu +++ b/c/backend_cuda.cu @@ -18,12 +18,14 @@ struct ColiCudaTensor { typedef struct { int device; + int compute_major,compute_minor; float *x, *y, *gate, *up; size_t x_cap, y_cap, gate_cap, up_cap; uint8_t *qx; float *qscale; size_t qx_cap, qscale_cap; float *host_x,*host_y; size_t host_x_cap,host_y_cap; float *aq,*al,*ar,*ac; size_t aq_cap,al_cap,ar_cap,ac_cap; + float *pipe_buf[24]; size_t pipe_cap[24]; /* scratch persistenti del resident pipeline */ cudaStream_t stream; void *group_desc; size_t group_desc_cap; size_t tensor_count, tensor_bytes; @@ -120,6 +122,68 @@ __global__ static void silu_mul(float *gate, const float *up, size_t n) { } } +/* Four warps share one A tile and compute 16x64 outputs. This matters for + * prefill: the first prototype reloaded/converter A once per 16 output cols. */ +__global__ static void w4a16_matmul(float *y,const float *x,const uint8_t *w, + const float *scale,int M,int K,int N){ +#if __CUDA_ARCH__ >= 700 + using namespace nvcuda;int warp=threadIdx.x>>5,lane=threadIdx.x&31; + int m0=blockIdx.y*16,n0=blockIdx.x*64+warp*16; + __shared__ __half ah[256],bh[4][256]; + wmma::fragment acc;wmma::fill_fragment(acc,0.f); + size_t rb=(size_t)(K+1)/2; + for(int k0=0;k0>1)];int a=(gk&1)?q>>4:q&15; + v=(float)(a&8?a-16:a)*scale[gn];} + bh[warp][z]=__float2half(v); /* [Ntile,Ktile] == B col-major */ + } + __syncthreads(); + wmma::fragment af; + wmma::fragment bf; + wmma::load_matrix_sync(af,ah,16);wmma::load_matrix_sync(bf,bh[warp],16); + wmma::mma_sync(acc,af,bf,acc);__syncthreads(); + } + __shared__ float out[4][256];wmma::store_matrix_sync(out[warp],acc,16,wmma::mem_row_major);__syncwarp(); + for(int z=lane;z<256;z+=32){int m=z/16,n=z%16; + if(m0+mFP16 conversion of A. */ +__global__ static void w4a16_gate_up(float *gate,float *up,const float *x, + const uint8_t *gw,const uint8_t *uw,const float *gs,const float *us, + int M,int K,int N){ +#if __CUDA_ARCH__ >= 700 + using namespace nvcuda;int warp=threadIdx.x>>5,lane=threadIdx.x&31,which=warp&1,tile=warp>>1; + int m0=blockIdx.y*16,n0=blockIdx.x*64+tile*16;const uint8_t *w=which?uw:gw; + const float *scale=which?us:gs;float *y=which?up:gate;size_t rb=(size_t)(K+1)/2; + __shared__ __half ah[256],bh[8][256]; + wmma::fragment acc;wmma::fill_fragment(acc,0.f); + for(int k0=0;k0>1)];int a=(gk&1)?q>>4:q&15; + v=(float)(a&8?a-16:a)*scale[gn];}bh[warp][z]=__float2half(v);} + __syncthreads(); + wmma::fragment af; + wmma::fragment bf; + wmma::load_matrix_sync(af,ah,16);wmma::load_matrix_sync(bf,bh[warp],16); + wmma::mma_sync(acc,af,bf,acc);__syncthreads(); + } + __shared__ float out[8][256];wmma::store_matrix_sync(out[warp],acc,16,wmma::mem_row_major);__syncwarp(); + for(int z=lane;z<256;z+=32){int m=z/16,n=z%16; + if(m0+m=S)return; const float *xs=x+(size_t)s*K; float v=0; for(int i=threadIdx.x;i=S||nt<1)return; + extern __shared__ float sm[];float *qa=sm,*cl=qa+K,*scores=cl+K,*red=scores+T; + const float *qs=q+((size_t)s*H+h)*(Q+R); + for(int k=tid;k>1;n;n>>=1){if(tid>1;n;n>>=1){if(tid= bytes) return 1; if (*ptr) cudaFree(*ptr); @@ -286,6 +381,7 @@ extern "C" int coli_cuda_init(const int *devices, int count) { if (!select_ctx(ctx)) { g_nctx = 0; return 0; } cudaDeviceProp prop{}; if (!cuda_ok(cudaGetDeviceProperties(&prop, device), "device properties")) { g_nctx = 0; return 0; } + ctx->compute_major=prop.major;ctx->compute_minor=prop.minor; if(!cuda_ok(cudaStreamCreateWithFlags(&ctx->stream,cudaStreamNonBlocking),"stream creation")){ g_nctx=0;return 0; } @@ -307,6 +403,7 @@ extern "C" void coli_cuda_shutdown(void) { if (ctx->qx) cudaFree(ctx->qx); if (ctx->qscale) cudaFree(ctx->qscale); if(ctx->aq)cudaFree(ctx->aq);if(ctx->al)cudaFree(ctx->al);if(ctx->ar)cudaFree(ctx->ar);if(ctx->ac)cudaFree(ctx->ac); + for(int b=0;b<24;b++) if(ctx->pipe_buf[b]) cudaFree(ctx->pipe_buf[b]); if (ctx->host_x) cudaFreeHost(ctx->host_x); if (ctx->host_y) cudaFreeHost(ctx->host_y); if (ctx->stream) cudaStreamDestroy(ctx->stream); @@ -453,6 +550,34 @@ extern "C" int coli_cuda_expert_mlp(ColiCudaTensor *gate, ColiCudaTensor *up, return 1; } +extern "C" int coli_cuda_shared_mlp_w4a16(ColiCudaTensor *gate,ColiCudaTensor *up, + ColiCudaTensor *down,float *y,const float *x,int S){ + if(!gate||!up||!down||!x||!y||S<1||gate->fmt!=2||up->fmt!=2||down->fmt!=2|| + gate->device!=up->device||gate->device!=down->device||gate->I!=up->I|| + gate->O!=up->O||down->I!=gate->O||down->O!=gate->I)return 0; + DeviceContext *ctx=find_ctx(gate->device);if(!select_ctx(ctx)||ctx->compute_major<7)return 0; + int D=gate->I,I=gate->O;size_t xb=(size_t)S*D*sizeof(float),ib=(size_t)S*I*sizeof(float); + if(!reserve(&ctx->x,&ctx->x_cap,xb)||!reserve(&ctx->gate,&ctx->gate_cap,ib)|| + !reserve(&ctx->up,&ctx->up_cap,ib)||!reserve(&ctx->y,&ctx->y_cap,xb)|| + !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), + "shared w4a16 input upload"))return 0; + dim3 hidden((unsigned)((I+63)/64),(unsigned)((S+15)/16)); + dim3 output((unsigned)((D+63)/64),(unsigned)((S+15)/16)); + w4a16_gate_up<<stream>>>(ctx->gate,ctx->up,ctx->x, + (const uint8_t*)gate->weights,(const uint8_t*)up->weights,gate->scales,up->scales,S,D,I); + silu_mul<<<(unsigned)(((size_t)S*I+255)/256),256,0,ctx->stream>>>(ctx->gate,ctx->up,(size_t)S*I); + w4a16_matmul<<stream>>>(ctx->y,ctx->gate,(const uint8_t*)down->weights,down->scales,S,I,D); + if(!cuda_ok(cudaGetLastError(),"shared w4a16 launch")|| + !cuda_ok(cudaMemcpyAsync(ctx->host_y,ctx->y,xb,cudaMemcpyDeviceToHost,ctx->stream), + "shared w4a16 output download")|| + !cuda_ok(cudaStreamSynchronize(ctx->stream),"shared w4a16 synchronize"))return 0; + std::memcpy(y,ctx->host_y,xb); + return 1; +} + extern "C" int coli_cuda_expert_group(ColiCudaTensor *const *gates, ColiCudaTensor *const *ups, ColiCudaTensor *const *downs, @@ -510,6 +635,38 @@ extern "C" int coli_cuda_expert_group(ColiCudaTensor *const *gates, silu_mul<<<(unsigned)(((size_t)total*I+255)/256),256,0,ctx->stream>>>(ctx->gate,ctx->up,(size_t)total*I); quantize_s4_rows<<stream>>>(ctx->qx,ctx->qscale,ctx->gate,total,I); grouped_s4_wmma<<stream>>>(ctx->y,ctx->qx,ctx->qscale,dev,I,D,2); + }else if(all_s4&&ctx->compute_major>=7&&getenv("COLI_CUDA_TC_W4A16")&& + atoi(getenv("COLI_CUDA_TC_W4A16"))){ + /* W4A16 Tensor Core per gruppo: attivazioni fp16 per tile (lossless al + * contrario del path W4A4), un lancio per expert dentro lo stream — + * l'overhead di lancio e' trascurabile rispetto ai GEMM. */ + int tc16_min=getenv("COLI_CUDA_TC_W4A16_MIN")?atoi(getenv("COLI_CUDA_TC_W4A16_MIN")):16; + int off16=0; + for(int c=0;cgate+(size_t)off16*I,*u16=ctx->up+(size_t)off16*I; + float *x16=ctx->x+(size_t)off16*D,*y16=ctx->y+(size_t)off16*D; + if(r>=tc16_min){ + dim3 hg16((unsigned)((I+63)/64),(unsigned)((r+15)/16)); + dim3 og16((unsigned)((D+63)/64),(unsigned)((r+15)/16)); + w4a16_gate_up<<stream>>>(g16,u16,x16, + (const uint8_t*)host[c].g,(const uint8_t*)host[c].u,host[c].gs,host[c].us,r,D,I); + silu_mul<<<(unsigned)(((size_t)r*I+255)/256),256,0,ctx->stream>>>(g16,u16,(size_t)r*I); + w4a16_matmul<<stream>>>(y16,g16, + (const uint8_t*)host[c].d,host[c].ds,r,I,D); + }else{ + /* piccoli batch: tile TC quasi vuoti + overhead di lancio — il + * kernel naive per-elemento resta piu' veloce (misurato in decode) */ + 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)); + } + off16+=r; + } }else if(all_s4&&(!getenv("COLI_CUDA_W4_PACKED")||atoi(getenv("COLI_CUDA_W4_PACKED")))){ dim3 hg((unsigned)I,(unsigned)max_rows,(unsigned)count),og((unsigned)D,(unsigned)max_rows,(unsigned)count); int dual=!getenv("COLI_CUDA_DUAL_PROJ")||atoi(getenv("COLI_CUDA_DUAL_PROJ")); @@ -547,6 +704,7 @@ extern "C" int coli_cuda_expert_group(ColiCudaTensor *const *gates, return 1; } + extern "C" int coli_cuda_attention_absorb(ColiCudaTensor *w,float *ctx,const float *q, const float *latent,const float *rope,int H,int Q, int R,int V,int K,int T,float scale){ @@ -569,6 +727,49 @@ extern "C" int coli_cuda_attention_absorb(ColiCudaTensor *w,float *ctx,const flo return 1; } +static int attention_absorb_batch_run(ColiCudaTensor *w,ColiCudaTensor *proj,float *out, + const float *q,const float *latent,const float *rope,int S,int H,int Q,int R,int V, + int K,int T,float scale){ + if(!w||!out||!q||!latent||!rope||S<1||H<1||Q<1||R<1||V<1||K<1||K>512|| + T8192||w->I!=K||w->O!=H*(Q+V))return 0; + if(proj&&(proj->device!=w->device||proj->I!=H*V))return 0; + DeviceContext *dc=find_ctx(w->device);if(!select_ctx(dc))return 0; + size_t qb=(size_t)S*H*(Q+R)*sizeof(float),lb=(size_t)T*K*sizeof(float); + size_t rb=(size_t)T*R*sizeof(float),cb=(size_t)S*H*V*sizeof(float); + if(!reserve(&dc->aq,&dc->aq_cap,qb)||!reserve(&dc->al,&dc->al_cap,lb)|| + !reserve(&dc->ar,&dc->ar_cap,rb)||!reserve(&dc->ac,&dc->ac_cap,cb))return 0; + if(!cuda_ok(cudaMemcpyAsync(dc->aq,q,qb,cudaMemcpyHostToDevice,dc->stream),"attention batch q upload")|| + !cuda_ok(cudaMemcpyAsync(dc->al,latent,lb,cudaMemcpyHostToDevice,dc->stream),"attention batch latent upload")|| + !cuda_ok(cudaMemcpyAsync(dc->ar,rope,rb,cudaMemcpyHostToDevice,dc->stream),"attention batch rope upload"))return 0; + size_t shared=(size_t)(2*K+T+256)*sizeof(float); + attention_absorb_batch_kernel<<stream>>>(dc->ac,dc->aq,dc->al, + dc->ar,w->weights,w->scales,w->fmt,S,H,Q,R,V,K,T,scale); + if(!cuda_ok(cudaGetLastError(),"attention batch launch"))return 0; + const float *src=dc->ac;size_t ob=cb; + if(proj){ + ob=(size_t)S*proj->O*sizeof(float);if(!reserve(&dc->y,&dc->y_cap,ob))return 0; + quant_matmul<<O,S),256,0,dc->stream>>>(dc->y,dc->ac,proj->weights, + proj->scales,proj->fmt,S,proj->I,proj->O,row_bytes(proj->fmt,proj->I)); + if(!cuda_ok(cudaGetLastError(),"attention o_proj launch"))return 0;src=dc->y; + } + if(!cuda_ok(cudaMemcpyAsync(out,src,ob,cudaMemcpyDeviceToHost,dc->stream), + proj?"attention projected output download":"attention batch context download")|| + !cuda_ok(cudaStreamSynchronize(dc->stream),"attention batch synchronize"))return 0; + return 1; +} + +extern "C" int coli_cuda_attention_absorb_batch(ColiCudaTensor *w,float *ctx,const float *q, + const float *latent,const float *rope,int S,int H,int Q,int R,int V,int K,int T, + float scale){ + return attention_absorb_batch_run(w,nullptr,ctx,q,latent,rope,S,H,Q,R,V,K,T,scale); +} + +extern "C" int coli_cuda_attention_project_batch(ColiCudaTensor *w,ColiCudaTensor *proj, + float *out,const float *q,const float *latent,const float *rope,int S,int H,int Q, + int R,int V,int K,int T,float scale){ + return attention_absorb_batch_run(w,proj,out,q,latent,rope,S,H,Q,R,V,K,T,scale); +} + extern "C" void coli_cuda_tensor_free(ColiCudaTensor *tensor) { if (!tensor) return; DeviceContext *ctx = find_ctx(tensor->device); @@ -590,3 +791,234 @@ extern "C" size_t coli_cuda_tensor_bytes(const ColiCudaTensor *tensor) { extern "C" int coli_cuda_tensor_device(const ColiCudaTensor *tensor) { return tensor ? tensor->device : -1; } + +/* ==== resident-pipeline primitives (Inc.0, 2026-07-13) ==== + * Device-side building blocks so the residual stream can stay on the layer's + * home device across a whole layer. Control flow stays on CPU; only the data + * plane lives here. All entry points take DEVICE pointers (no transfers) — + * the caller owns staging via the pipe buffer API below. */ + +__global__ static void pipe_rmsnorm_rows(float *y,const float *x,const float *w, + int D,float eps,int xstride,int ystride){ + const float *xr=x+(size_t)blockIdx.x*xstride; float *yr=y+(size_t)blockIdx.x*ystride; + __shared__ double sh[256]; + double a=0; for(int i=threadIdx.x;i0;s>>=1){ if(threadIdx.x=24||!select_ctx(ctx)) return NULL; + if(!reserve(&ctx->pipe_buf[slot],&ctx->pipe_cap[slot],bytes)) return NULL; + return ctx->pipe_buf[slot]; +} +extern "C" void *coli_cuda_pipe_alloc(int device,size_t bytes){ + DeviceContext *ctx=find_ctx(device); if(!select_ctx(ctx)) return NULL; + void *p=NULL; + if(!cuda_ok(cudaMalloc(&p,bytes),"pipe alloc")) return NULL; + return p; +} +extern "C" void coli_cuda_pipe_free(int device,void *p){ + DeviceContext *ctx=find_ctx(device); if(!p||!select_ctx(ctx)) return; + cudaFree(p); +} +extern "C" int coli_cuda_pipe_upload(int device,void *dst,const void *src,size_t bytes){ + DeviceContext *ctx=find_ctx(device); if(!select_ctx(ctx)) return 0; + return cuda_ok(cudaMemcpy(dst,src,bytes,cudaMemcpyHostToDevice),"pipe upload"); +} +extern "C" int coli_cuda_pipe_download(int device,const void *src,void *dst,size_t bytes){ + DeviceContext *ctx=find_ctx(device); if(!select_ctx(ctx)) return 0; + return cuda_ok(cudaMemcpy(dst,src,bytes,cudaMemcpyDeviceToHost),"pipe download"); +} +extern "C" int coli_cuda_pipe_rmsnorm(int device,float *y_dev,const float *x_dev, + const float *w_dev,int S,int D,float eps){ + DeviceContext *ctx=find_ctx(device); + if(S<1||D<1||!select_ctx(ctx)) return 0; + pipe_rmsnorm_rows<<>>(y_dev,x_dev,w_dev,D,eps,D,D); + return cuda_ok(cudaGetLastError(),"pipe rmsnorm"); +} +extern "C" int coli_cuda_pipe_rmsnorm_s(int device,float *y_dev,const float *x_dev, + const float *w_dev,int S,int D,float eps, + int xstride,int ystride){ + DeviceContext *ctx=find_ctx(device); + if(S<1||D<1||xstride>>(y_dev,x_dev,w_dev,D,eps,xstride,ystride); + return cuda_ok(cudaGetLastError(),"pipe rmsnorm strided"); +} +extern "C" int coli_cuda_pipe_rope(int device,float *v_dev,const int *pos_dev, + int rows,int stride,int offset,int R,int heads, + float theta){ + DeviceContext *ctx=find_ctx(device); + if(rows<1||R<2||R>256||heads<1||!select_ctx(ctx)) return 0; + pipe_rope_rows<<>>(v_dev,pos_dev,0,stride,offset,R,heads,theta); + return cuda_ok(cudaGetLastError(),"pipe rope"); +} +extern "C" int coli_cuda_pipe_rope_base(int device,float *v_dev,int pos_base,int rows, + int stride,int offset,int R,int heads,float theta){ + DeviceContext *ctx=find_ctx(device); + if(rows<1||R<2||R>256||heads<1||!select_ctx(ctx)) return 0; + pipe_rope_rows<<>>(v_dev,NULL,pos_base,stride,offset,R,heads,theta); + return cuda_ok(cudaGetLastError(),"pipe rope base"); +} +extern "C" int coli_cuda_pipe_copy2d(int device,float *dst,int dpitch,const float *src, + int spitch,int width,int height){ + DeviceContext *ctx=find_ctx(device); if(!select_ctx(ctx)) return 0; + return cuda_ok(cudaMemcpy2D(dst,(size_t)dpitch*4,src,(size_t)spitch*4, + (size_t)width*4,height,cudaMemcpyDeviceToDevice),"pipe copy2d"); +} +/* attention batch + fused o_proj with DEVICE-resident q/latent/rope: the whole + * upstream projection chain stayed on this device, so nothing is uploaded here. + * Only the final [S,O] projection is downloaded to host. */ +extern "C" int coli_cuda_attention_project_batch_dev(ColiCudaTensor *w,ColiCudaTensor *proj, + float *out,const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale){ + if(!w||!proj||!out||!q_dev||!latent_dev||!rope_dev||S<1||H<1||Q<1||R<1||V<1|| + K<1||K>512||T8192||w->I!=K||w->O!=H*(Q+V)|| + proj->device!=w->device||proj->I!=H*V)return 0; + DeviceContext *dc=find_ctx(w->device);if(!select_ctx(dc))return 0; + size_t cb=(size_t)S*H*V*sizeof(float); + if(!reserve(&dc->ac,&dc->ac_cap,cb))return 0; + size_t shared=(size_t)(2*K+T+256)*sizeof(float); + attention_absorb_batch_kernel<<stream>>>(dc->ac,q_dev,latent_dev, + rope_dev,w->weights,w->scales,w->fmt,S,H,Q,R,V,K,T,scale); + if(!cuda_ok(cudaGetLastError(),"pipe attention launch"))return 0; + size_t ob=(size_t)S*proj->O*sizeof(float); + if(!reserve(&dc->y,&dc->y_cap,ob))return 0; + quant_matmul<<O,S),256,0,dc->stream>>>(dc->y,dc->ac,proj->weights, + proj->scales,proj->fmt,S,proj->I,proj->O,row_bytes(proj->fmt,proj->I)); + if(!cuda_ok(cudaGetLastError(),"pipe o_proj launch"))return 0; + if(!cuda_ok(cudaMemcpyAsync(out,dc->y,ob,cudaMemcpyDeviceToHost,dc->stream),"pipe attention download")|| + !cuda_ok(cudaStreamSynchronize(dc->stream),"pipe attention sync"))return 0; + return 1; +} +extern "C" int coli_cuda_pipe_silu_mul(int device,float *gate_dev,const float *up_dev, + size_t n){ + DeviceContext *ctx=find_ctx(device); if(!n||!select_ctx(ctx)) return 0; + silu_mul<<<(unsigned)((n+255)/256),256>>>(gate_dev,up_dev,n); + return cuda_ok(cudaGetLastError(),"pipe silu mul"); +} +extern "C" int coli_cuda_pipe_add(int device,float *x_dev,const float *t_dev,size_t n){ + DeviceContext *ctx=find_ctx(device); if(!n||!select_ctx(ctx)) return 0; + pipe_add_n<<<(unsigned)((n+255)/256),256>>>(x_dev,t_dev,n); + return cuda_ok(cudaGetLastError(),"pipe add"); +} +extern "C" int coli_cuda_pipe_rows_add(int device,float *x_dev,const float *partial_dev, + const int *rows_dev,int nrows,int D){ + DeviceContext *ctx=find_ctx(device); if(nrows<1||D<1||!select_ctx(ctx)) return 0; + pipe_rows_add<<>>(x_dev,partial_dev,rows_dev,D); + return cuda_ok(cudaGetLastError(),"pipe rows add"); +} +/* GEMM with device-resident activations: same quant_matmul kernel as + * coli_cuda_matmul, zero host transfers. */ +extern "C" int coli_cuda_pipe_gemm(ColiCudaTensor *t,float *y_dev,const float *x_dev, + int S){ + if(!t||S<1) return 0; + DeviceContext *ctx=find_ctx(t->device); if(!select_ctx(ctx)) return 0; + dim3 grid((unsigned)t->O,(unsigned)S); + quant_matmul<<>>(y_dev,x_dev,t->weights,t->scales,t->fmt,S,t->I,t->O, + row_bytes(t->fmt,t->I)); + return cuda_ok(cudaGetLastError(),"pipe gemm"); +} +/* copia diretta scheda->scheda (P2P se disponibile, altrimenti staging driver) */ +extern "C" int coli_cuda_pipe_peer_copy(int dst_dev,float *dst,int src_dev, + const float *src,size_t bytes){ + if(!dst||!src) return 0; + if(dst_dev==src_dev){ DeviceContext *c=find_ctx(dst_dev); if(!select_ctx(c)) return 0; + return cuda_ok(cudaMemcpy(dst,src,bytes,cudaMemcpyDeviceToDevice),"pipe intra copy"); } + return cuda_ok(cudaMemcpyPeer(dst,dst_dev,src,src_dev,bytes),"pipe peer copy"); +} +/* come attention_project_batch_dev ma l'uscita di o_proj RESTA sul device (out_dev). */ +extern "C" int coli_cuda_attention_project_batch_dev_out(ColiCudaTensor *w,ColiCudaTensor *proj, + float *out_dev,const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale){ + if(!w||!proj||!out_dev||!q_dev||!latent_dev||!rope_dev||S<1||H<1||Q<1||R<1||V<1|| + K<1||K>512||T8192||w->I!=K||w->O!=H*(Q+V)|| + proj->device!=w->device||proj->I!=H*V)return 0; + DeviceContext *dc=find_ctx(w->device);if(!select_ctx(dc))return 0; + size_t cb=(size_t)S*H*V*sizeof(float); + if(!reserve(&dc->ac,&dc->ac_cap,cb))return 0; + size_t shared=(size_t)(2*K+T+256)*sizeof(float); + attention_absorb_batch_kernel<<stream>>>(dc->ac,q_dev,latent_dev, + rope_dev,w->weights,w->scales,w->fmt,S,H,Q,R,V,K,T,scale); + if(!cuda_ok(cudaGetLastError(),"pipe attention launch (dev out)"))return 0; + quant_matmul<<O,S),256,0,dc->stream>>>(out_dev,dc->ac,proj->weights, + proj->scales,proj->fmt,S,proj->I,proj->O,row_bytes(proj->fmt,proj->I)); + if(!cuda_ok(cudaGetLastError(),"pipe o_proj launch (dev out)"))return 0; + return cuda_ok(cudaStreamSynchronize(dc->stream),"pipe attention sync (dev out)"); +} +/* absorb batch con TUTTO su device (q/latent/rope gia' residenti sulla scheda + * dello shard, ctx resta sul device): il cuore della attention head-shardata + * dentro il pipeline. Nessun trasferimento host. */ +extern "C" int coli_cuda_attention_absorb_batch_dev(ColiCudaTensor *w,float *ctx_dev, + const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale){ + if(!w||!ctx_dev||!q_dev||!latent_dev||!rope_dev||S<1||H<1||Q<1||R<1||V<1|| + K<1||K>512||T8192||w->I!=K||w->O!=H*(Q+V))return 0; + DeviceContext *dc=find_ctx(w->device);if(!select_ctx(dc))return 0; + size_t shared=(size_t)(2*K+T+256)*sizeof(float); + attention_absorb_batch_kernel<<stream>>>(ctx_dev,q_dev,latent_dev, + rope_dev,w->weights,w->scales,w->fmt,S,H,Q,R,V,K,T,scale); + if(!cuda_ok(cudaGetLastError(),"pipe shard attention launch"))return 0; + return cuda_ok(cudaStreamSynchronize(dc->stream),"pipe shard attention sync"); +} +/* absorb per il DECODE con KV gia' residente: carica solo q (poche KB), + * latent/rope arrivano dall'ombra device. ctx torna a host (S piccolo). */ +extern "C" int coli_cuda_attention_absorb_kvdev(ColiCudaTensor *w,float *ctx,const float *q, + const float *latent_dev,const float *rope_dev,int H,int Q,int R,int V,int K,int T, + float scale){ + if(!w||!ctx||!q||!latent_dev||!rope_dev||H<1||Q<1||R<1||V<1||K<1||K>512||T<1||T>8192|| + w->I!=K||w->O!=H*(Q+V))return 0; + DeviceContext *dc=find_ctx(w->device);if(!select_ctx(dc))return 0; + size_t qb=(size_t)H*(Q+R)*sizeof(float),cb=(size_t)H*V*sizeof(float); + if(!reserve(&dc->aq,&dc->aq_cap,qb)||!reserve(&dc->ac,&dc->ac_cap,cb))return 0; + if(!cuda_ok(cudaMemcpyAsync(dc->aq,q,qb,cudaMemcpyHostToDevice,dc->stream),"kvdev q upload"))return 0; + size_t shared=(size_t)(2*K+T+256)*sizeof(float); + attention_absorb_batch_kernel<<stream>>>(dc->ac,dc->aq,latent_dev, + rope_dev,w->weights,w->scales,w->fmt,1,H,Q,R,V,K,T,scale); + if(!cuda_ok(cudaGetLastError(),"kvdev absorb launch")|| + !cuda_ok(cudaMemcpyAsync(ctx,dc->ac,cb,cudaMemcpyDeviceToHost,dc->stream),"kvdev ctx download")|| + !cuda_ok(cudaStreamSynchronize(dc->stream),"kvdev absorb sync"))return 0; + return 1; +} +extern "C" int coli_cuda_pipe_sync(int device){ + DeviceContext *ctx=find_ctx(device); if(!select_ctx(ctx)) return 0; + return cuda_ok(cudaDeviceSynchronize(),"pipe sync"); +} diff --git a/c/backend_cuda.h b/c/backend_cuda.h index 533bbea..c84a265 100644 --- a/c/backend_cuda.h +++ b/c/backend_cuda.h @@ -56,6 +56,14 @@ COLI_CUDA_DLLEXPORT int coli_cuda_matmul(ColiCudaTensor **tensor, COLI_CUDA_DLLEXPORT int coli_cuda_expert_mlp(ColiCudaTensor *gate, ColiCudaTensor *up, ColiCudaTensor *down, float *y, const float *x, int S); +/* Prefill-oriented shared expert path. INT4 weights stay packed in global + * memory, activations are converted to FP16 per tile, and Tensor Cores + * accumulate into FP32. Unlike COLI_CUDA_TC_INT4 this does not quantize the + * activation to INT4. */ +int coli_cuda_shared_mlp_w4a16(ColiCudaTensor *gate, ColiCudaTensor *up, + ColiCudaTensor *down, float *y, + const float *x, int S); + /* Packed group of same-shaped experts. Inputs and outputs contain sum(rows) * consecutive [D] rows in call order. */ COLI_CUDA_DLLEXPORT int coli_cuda_expert_group(ColiCudaTensor *const *gates, @@ -69,6 +77,21 @@ COLI_CUDA_DLLEXPORT int coli_cuda_attention_absorb(ColiCudaTensor *kv_b,float *c const float *latent,const float *rope,int H,int Q, int R,int V,int K,int T,float attention_scale); +/* Causal MLA absorption for S contiguous rows from one sequence. The KV + * arrays contain T rows ending at the final query; query s attends T-S+s+1 + * rows. One transfer and one launch replace S host round-trips. */ +COLI_CUDA_DLLEXPORT int coli_cuda_attention_absorb_batch(ColiCudaTensor *kv_b,float *ctx,const float *q, + const float *latent,const float *rope,int S, + int H,int Q,int R,int V,int K,int T, + float attention_scale); + +/* Same attention batch followed immediately by resident o_proj on the same + * device. Only the final [S,D] tensor crosses back to the host. */ +COLI_CUDA_DLLEXPORT int coli_cuda_attention_project_batch(ColiCudaTensor *kv_b,ColiCudaTensor *o_proj, + float *out,const float *q,const float *latent, + const float *rope,int S,int H,int Q,int R, + int V,int K,int T,float attention_scale); + COLI_CUDA_DLLEXPORT void coli_cuda_tensor_free(ColiCudaTensor *tensor); COLI_CUDA_DLLEXPORT size_t coli_cuda_tensor_bytes(const ColiCudaTensor *tensor); COLI_CUDA_DLLEXPORT int coli_cuda_tensor_device(const ColiCudaTensor *tensor); @@ -77,8 +100,47 @@ COLI_CUDA_DLLEXPORT int coli_cuda_tensor_device(const ColiCudaTensor *tensor); int coli_cuda_tensor_update(ColiCudaTensor *tensor, const void *weights, const float *scales); +/* ---- resident-pipeline primitives (Inc.0): device-pointer entry points ---- */ +float *coli_cuda_pipe_scratch(int device,int slot,size_t bytes); +void *coli_cuda_pipe_alloc(int device,size_t bytes); +void coli_cuda_pipe_free(int device,void *p); +int coli_cuda_pipe_upload(int device,void *dst,const void *src,size_t bytes); +int coli_cuda_pipe_download(int device,const void *src,void *dst,size_t bytes); +int coli_cuda_pipe_rmsnorm(int device,float *y_dev,const float *x_dev, + const float *w_dev,int S,int D,float eps); +int coli_cuda_pipe_rope(int device,float *v_dev,const int *pos_dev,int rows, + int stride,int offset,int R,int heads,float theta); +int coli_cuda_pipe_silu_mul(int device,float *gate_dev,const float *up_dev,size_t n); +int coli_cuda_pipe_add(int device,float *x_dev,const float *t_dev,size_t n); +int coli_cuda_pipe_rows_add(int device,float *x_dev,const float *partial_dev, + const int *rows_dev,int nrows,int D); +int coli_cuda_pipe_gemm(ColiCudaTensor *t,float *y_dev,const float *x_dev,int S); +int coli_cuda_pipe_rmsnorm_s(int device,float *y_dev,const float *x_dev, + const float *w_dev,int S,int D,float eps, + int xstride,int ystride); +int coli_cuda_pipe_rope_base(int device,float *v_dev,int pos_base,int rows, + int stride,int offset,int R,int heads,float theta); +int coli_cuda_pipe_copy2d(int device,float *dst,int dpitch,const float *src, + int spitch,int width,int height); +int coli_cuda_attention_project_batch_dev(ColiCudaTensor *kv_b,ColiCudaTensor *o_proj, + float *out,const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale); +int coli_cuda_attention_absorb_batch_dev(ColiCudaTensor *kv_b_shard,float *ctx_dev, + const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale); +int coli_cuda_attention_absorb_kvdev(ColiCudaTensor *kv_b,float *ctx,const float *q, + const float *latent_dev,const float *rope_dev,int H,int Q,int R,int V,int K,int T, + float scale); +int coli_cuda_pipe_peer_copy(int dst_dev,float *dst,int src_dev, + const float *src,size_t bytes); +int coli_cuda_attention_project_batch_dev_out(ColiCudaTensor *kv_b,ColiCudaTensor *o_proj, + float *out_dev,const float *q_dev,const float *latent_dev,const float *rope_dev, + int S,int H,int Q,int R,int V,int K,int T,float scale); +int coli_cuda_pipe_sync(int device); + #ifdef __cplusplus } #endif #endif + diff --git a/c/glm.c b/c/glm.c index 561d410..2780b79 100644 --- a/c/glm.c +++ b/c/glm.c @@ -113,6 +113,11 @@ typedef struct { float *in_ln, *post_ln; /* MLA (densa, quantizzata) */ QT q_a, q_b, kv_a, kv_b, o; float *q_a_ln, *kv_a_ln; +#ifdef COLI_CUDA + ColiCudaTensor *kv_b_shard[COLI_CUDA_MAX_DEVICES]; + int shard_h0[COLI_CUDA_MAX_DEVICES],shard_hn[COLI_CUDA_MAX_DEVICES],n_kv_b_shard; + int shared_w4a16_failed; +#endif int sparse; /* dense mlp (sparse==0) */ QT gate_proj, up_proj, down_proj; @@ -152,6 +157,7 @@ typedef struct { int *kv_start; /* prima pos valida nella KV del layer (MTP: parziale) */ 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) */ 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) */ @@ -1065,6 +1071,33 @@ static float *ld(Model *m, const char *name){ /* tensore 1D f32 residente (nor float *p=(float*)qalloc((size_t)n*sizeof(float)); /* registrato per la GPU sotto METAL */ st_read_f32(&m->S,name,p,0); return p; } +#ifdef COLI_CUDA +static void qt_cuda_colocate(QT *dst,const QT *src){ + if(!g_cuda_enabled||!g_cuda_dense||!dst->cuda_eligible||!src->cuda_eligible|| + dst->cuda_device==src->cuda_device)return; + int old=-1,now=-1;for(int i=0;icuda_device)old=i;if(g_cuda_devices[i]==src->cuda_device)now=i; + } + if(old>=0)g_cuda_dense_projected[old]-=qt_bytes(dst); + if(now>=0)g_cuda_dense_projected[now]+=qt_bytes(dst); + dst->cuda_device=src->cuda_device; +} +static void layer_cuda_shard_kvb(Layer *l,int H,int Q,int V){ + if(!g_cuda_enabled||!g_cuda_dense||g_cuda_ndev<2||l->kv_b.fmt==0)return; + int rb=l->kv_b.fmt==1?l->kv_b.I:(l->kv_b.fmt==2?(l->kv_b.I+1)/2:(l->kv_b.I+3)/4); + const uint8_t *weights=l->kv_b.fmt==1?(const uint8_t*)l->kv_b.q8:l->kv_b.q4; + for(int d=0,h0=0;dkv_b.s+(int64_t)h0*(Q+V); + if(!coli_cuda_tensor_upload(&l->kv_b_shard[d],part,scale,l->kv_b.fmt,l->kv_b.I,rows,g_cuda_devices[d]))return; + l->shard_h0[d]=h0;l->shard_hn[d]=hn;l->n_kv_b_shard++;h0+=hn; + } + int old=-1;for(int i=0;ikv_b.cuda_device)old=i; + if(old>=0)g_cuda_dense_projected[old]-=qt_bytes(&l->kv_b); + l->kv_b.cuda_eligible=0; +} +#endif static void model_init(Model *m, const char *snap, int cap, int ebits, int dbits){ memset(m,0,sizeof(*m)); m->ebits=ebits; m->dbits=dbits; @@ -1079,6 +1112,8 @@ static void model_init(Model *m, const char *snap, int cap, int ebits, int dbits m->L=calloc(c->n_layers,sizeof(Layer)); int NR=c->n_layers+1; /* +1: riga del layer MTP */ m->ecap=cap; m->ecache=calloc(NR,sizeof(ESlot*)); m->ecn=calloc(NR,sizeof(int)); + m->kv_dev_L=calloc(NR,sizeof(float*)); m->kv_dev_R=calloc(NR,sizeof(float*)); + m->kv_dev_valid=calloc(NR,sizeof(int)); m->eroute=calloc(NR,sizeof(int*)); m->enr=calloc(NR,sizeof(int)); m->pin=calloc(NR,sizeof(ESlot*)); m->npin=calloc(NR,sizeof(int)); m->eusage=calloc(NR,sizeof(uint32_t*)); m->eheat=calloc(NR,sizeof(uint32_t*)); @@ -1097,6 +1132,14 @@ static void model_init(Model *m, const char *snap, int cap, int ebits, int dbits l->kv_a_ln= ld(m,P("self_attn.kv_a_layernorm.weight")); l->kv_b = qt_load(m,P("self_attn.kv_b_proj.weight"), H*(c->qk_nope+c->v_head), c->kv_lora, dbits); l->o = qt_load(m,P("self_attn.o_proj.weight"), D, H*c->v_head, dbits); +#ifdef COLI_CUDA + qt_cuda_colocate(&l->o,&l->kv_b); + qt_cuda_colocate(&l->q_a,&l->kv_b); /* PIPE: intera catena attention sulla */ + qt_cuda_colocate(&l->q_b,&l->kv_b); /* stessa scheda / whole attention chain */ + qt_cuda_colocate(&l->kv_a,&l->kv_b); /* on the layer home device */ + if(getenv("COLI_CUDA_ATTN_SHARD")&&atoi(getenv("COLI_CUDA_ATTN_SHARD"))) + layer_cuda_shard_kvb(l,H,c->qk_nope,c->v_head); +#endif l->sparse = (i >= c->first_dense); if(!l->sparse){ l->gate_proj = qt_load(m,P("mlp.gate_proj.weight"), c->dense_inter, D, dbits); @@ -1109,6 +1152,11 @@ static void model_init(Model *m, const char *snap, int cap, int ebits, int dbits l->sh_gate = qt_load(m,P("mlp.shared_experts.gate_proj.weight"), sI, D, dbits); l->sh_up = qt_load(m,P("mlp.shared_experts.up_proj.weight"), sI, D, dbits); l->sh_down = qt_load(m,P("mlp.shared_experts.down_proj.weight"), D, sI, dbits); +#ifdef COLI_CUDA + qt_cuda_colocate(&l->sh_gate,&l->kv_b); /* PIPE2: shared chain on the layer home device */ + qt_cuda_colocate(&l->sh_up,&l->sh_gate); + qt_cuda_colocate(&l->sh_down,&l->sh_gate); +#endif m->ecache[i]=calloc(cap,sizeof(ESlot)); m->eroute[i]=calloc(c->topk,sizeof(int)); /* metodo C: ultimo routing del layer */ m->eusage[i]=calloc(c->n_experts,sizeof(uint32_t)); @@ -1573,7 +1621,10 @@ static void qt_matvec_rows(const QT *t, int r0, int n, const float *x, float *y) y[j]=(float)a; } } -static int g_absorb=-1; /* ABSORB: -1 auto (decode S<=4), 0 mai, 1 sempre (test) */ +static int g_absorb=-1; +#ifdef COLI_CUDA +static int g_cuda_pipe=0; /* COLI_CUDA_PIPE=1: prefill attention chain resident on the layer home device */ +#endif /* ABSORB: -1 auto (decode S<=4), 0 mai, 1 sempre (test) */ static int g_dsa_force=0; /* DSA_FORCE=1: selezione sempre attiva (test: top-min(k,T)=denso) */ static int cmp_fdesc(const void *a,const void *b){ float x=*(const float*)a, y=*(const float*)b; return xy?-1:0; } @@ -1581,6 +1632,141 @@ static int cmp_fdesc(const void *a,const void *b){ /* attenzione MLA con KV-cache compressa, su token nuovi x[S,hidden], pos_base = pos del primo */ /* kvs/pos describe a ragged decode batch: each row may belong to a different * sequence. NULL keeps the original contiguous, currently-bound KV path. */ +#ifdef COLI_CUDA +/* Ombra KV su device per il DECODE: righe [0,upto) valide sulla scheda di kv_b. + * L'host resta canonico; l'ombra si riallinea in blocco quando resta indietro e + * viene invalidata da kv_bind / dalla riscrittura di righe gia' specchiate. */ +static int kv_dev_sync(Model *m, Layer *l, int layer, int upto){ + Cfg *c=&m->c; int kvl=c->kv_lora, R=c->qk_rope, dev=l->kv_b.cuda_device; + if(upto>m->max_t) return 0; + if(!m->kv_dev_L[layer]){ + m->kv_dev_L[layer]=(float*)coli_cuda_pipe_alloc(dev,(size_t)m->max_t*kvl*4); + m->kv_dev_R[layer]=(float*)coli_cuda_pipe_alloc(dev,(size_t)m->max_t*R*4); + m->kv_dev_valid[layer]=0; + if(!m->kv_dev_L[layer]||!m->kv_dev_R[layer]) return 0; + } + int v=m->kv_dev_valid[layer]; + if(vkv_dev_L[layer]+(size_t)v*kvl, + coli_kv_row(m->kv->Lc[layer],v,kvl),(size_t)(upto-v)*kvl*4)|| + !coli_cuda_pipe_upload(dev,m->kv_dev_R[layer]+(size_t)v*R, + coli_kv_row(m->kv->Rc[layer],v,R),(size_t)(upto-v)*R*4)) return 0; + m->kv_dev_valid[layer]=upto; + } + return 1; +} + +/* Inc.1a — catena attention residente sul device del layer / attention chain + * resident on the layer home device. Proiezioni q/kv, norme, RoPE, batch + * attention e o_proj girano sulla scheda di kv_b; scaricano solo out [S,D], + * i nuovi record KV [S,kvl+R] e nulla altro. Ritorna 0 su qualsiasi errore: + * il chiamante riesegue il percorso CPU (idempotente). */ +static int attn_pipe_prefill(Model *m, Layer *l, int layer, const float *x, int x_is_dev, + int S, int pos_base, float *out, float *out_dev){ + Cfg *c=&m->c; int H=c->n_heads, D=c->hidden, qh=c->qk_head; + int kvl=c->kv_lora, R=c->qk_rope, ql=c->q_lora; + int dev=l->kv_b.cuda_device; + if(l->q_a.cuda_device!=dev||l->q_b.cuda_device!=dev|| + l->kv_a.cuda_device!=dev||l->o.cuda_device!=dev) return 0; + int st0=m->kv_start[layer], T=pos_base+S-st0, old=pos_base-st0; + if(T8192) return 0; + double t0=now_s(); + size_t xb=(size_t)S*D*4, qrb=(size_t)S*ql*4, qb=(size_t)S*H*qh*4; + size_t cb=(size_t)S*(kvl+R)*4, lb=(size_t)T*kvl*4, rb=(size_t)T*R*4; + float *chost=NULL; int ok=0; + /* scratch persistenti (slot fissi per device): zero churn di cudaMalloc */ + float *xd =x_is_dev?(float*)x:coli_cuda_pipe_scratch(dev,0,xb); + float *qrd=coli_cuda_pipe_scratch(dev,1,qrb); + float *qd =coli_cuda_pipe_scratch(dev,2,qb), *cd =coli_cuda_pipe_scratch(dev,3,cb); + float *ld_=coli_cuda_pipe_scratch(dev,4,lb), *rd =coli_cuda_pipe_scratch(dev,5,rb); + float *w1 =coli_cuda_pipe_scratch(dev,6,(size_t)ql*4); + float *w2 =coli_cuda_pipe_scratch(dev,7,(size_t)kvl*4); + chost=(float*)malloc(cb); + if(!xd||!qrd||!qd||!cd||!ld_||!rd||!w1||!w2||!chost) goto done; + if((!x_is_dev&&!coli_cuda_pipe_upload(dev,xd,x,xb))|| + !coli_cuda_pipe_upload(dev,w1,l->q_a_ln,(size_t)ql*4)|| + !coli_cuda_pipe_upload(dev,w2,l->kv_a_ln,(size_t)kvl*4)) goto done; + /* proiezioni + norme + rope, tutto sul device */ + if(!coli_cuda_pipe_gemm(l->q_a.cuda,qrd,xd,S)) goto done; + if(!coli_cuda_pipe_rmsnorm(dev,qrd,qrd,w1,S,ql,c->eps)) goto done; + if(!coli_cuda_pipe_gemm(l->q_b.cuda,qd,qrd,S)) goto done; + if(!coli_cuda_pipe_rope_base(dev,qd,pos_base,S*H,qh,c->qk_nope,R,H,c->theta)) goto done; + if(!coli_cuda_pipe_gemm(l->kv_a.cuda,cd,xd,S)) goto done; + if(!coli_cuda_pipe_rmsnorm_s(dev,cd,cd,w2,S,kvl,c->eps,kvl+R,kvl+R)) goto done; + if(!coli_cuda_pipe_rope_base(dev,cd,pos_base,S,kvl+R,kvl,R,1,c->theta)) goto done; + /* cache latente [T,kvl] + rot [T,R] contigue: righe vecchie da host, nuove da cd */ + if(old>0){ + if(!coli_cuda_pipe_upload(dev,ld_,coli_kv_row(m->Lc[layer],st0,kvl),(size_t)old*kvl*4)|| + !coli_cuda_pipe_upload(dev,rd,coli_kv_row(m->Rc[layer],st0,R),(size_t)old*R*4)) goto done; + } + if(!coli_cuda_pipe_copy2d(dev,ld_+(size_t)old*kvl,kvl,cd,kvl+R,kvl,S)) goto done; + if(!coli_cuda_pipe_copy2d(dev,rd+(size_t)old*R,R,cd+kvl,kvl+R,R,S)) goto done; + /* KV host resta canonica: scarica i record nuovi (gia' normati+ropati) */ + if(!coli_cuda_pipe_download(dev,cd,chost,cb)) goto done; + for(int s=0;sLc[layer],pos_base+s,kvl),chost+(size_t)s*(kvl+R),kvl*4); + memcpy(coli_kv_row(m->Rc[layer],pos_base+s,R),chost+(size_t)s*(kvl+R)+kvl,R*4); + } + if(m->kv_dev_valid[layer]>pos_base) m->kv_dev_valid[layer]=pos_base; + m->t_aproj+=now_s()-t0; t0=now_s(); +#ifdef COLI_CUDA + /* Negativo (2026-07-13): P2P a stella dal device di casa serializza ~95MB/layer + * sul suo link PCIe — attention 26->41-44s. Resta opt-in per topologie NVLink. */ + if(out_dev && l->n_kv_b_shard>1 && + getenv("COLI_CUDA_PIPE_SHARD") && atoi(getenv("COLI_CUDA_PIPE_SHARD"))){ + /* head-shard nel pipeline: q gia' sul device di casa. Per ogni scheda: + * slice di q (repack strided->contiguo), broadcast latent+rope via P2P, + * score parallelo sui rispettivi head, ctx slice riportata a casa e + * ricomposta, poi o_proj residente. */ + int n=l->n_kv_b_shard, vh=c->v_head; + size_t ctxb=(size_t)S*H*vh*4; + size_t stage_one=(size_t)S*H*(size_t)(c->qk_head>vh?c->qk_head:vh)*4; + float *ctx_full=coli_cuda_pipe_scratch(dev,16,ctxb); + float *stage=coli_cuda_pipe_scratch(dev,17,stage_one*n); + int ok_sh=(ctx_full&&stage)?1:0; + if(ok_sh){ + #pragma omp parallel for schedule(static) reduction(&:ok_sh) + for(int d2=0;d2shard_hn[d2], h0=l->shard_h0[d2]; + int sdev=coli_cuda_tensor_device(l->kv_b_shard[d2]); + float *st=stage+(size_t)d2*(stage_one/4); + size_t qsb=(size_t)S*hn*qh*4, csb=(size_t)S*hn*vh*4; + float *qs_r=coli_cuda_pipe_scratch(sdev,18,qsb); + float *ld_r=coli_cuda_pipe_scratch(sdev,19,(size_t)T*kvl*4); + float *rr_r=coli_cuda_pipe_scratch(sdev,20,(size_t)T*R*4); + float *cx_r=coli_cuda_pipe_scratch(sdev,21,csb); + int okd=qs_r&&ld_r&&rr_r&&cx_r; + /* slice di q: [S,H,qh] -> [S,hn,qh] contigua sul device di casa */ + okd=okd&&coli_cuda_pipe_copy2d(dev,st,hn*qh,qd+(size_t)h0*qh,H*qh,hn*qh,S); + okd=okd&&coli_cuda_pipe_peer_copy(sdev,qs_r,dev,st,qsb); + okd=okd&&coli_cuda_pipe_peer_copy(sdev,ld_r,dev,ld_,(size_t)T*kvl*4); + okd=okd&&coli_cuda_pipe_peer_copy(sdev,rr_r,dev,rd,(size_t)T*R*4); + okd=okd&&coli_cuda_attention_absorb_batch_dev(l->kv_b_shard[d2],cx_r,qs_r,ld_r,rr_r, + S,hn,c->qk_nope,R,vh,kvl,T,c->attn_scale); + okd=okd&&coli_cuda_pipe_peer_copy(dev,st,sdev,cx_r,csb); + okd=okd&&coli_cuda_pipe_copy2d(dev,ctx_full+(size_t)h0*vh,H*vh,st,hn*vh,hn*vh,S); + ok_sh&=okd; + } + } + if(ok_sh){ + ok=coli_cuda_pipe_gemm(l->o.cuda,out_dev,ctx_full,S)&&coli_cuda_pipe_sync(dev); + } else ok=0; + if(!ok) + ok=coli_cuda_attention_project_batch_dev_out(l->kv_b.cuda,l->o.cuda,out_dev,qd,ld_,rd, + S,H,c->qk_nope,R,c->v_head,kvl,T,c->attn_scale); + } else +#endif + ok=out_dev?coli_cuda_attention_project_batch_dev_out(l->kv_b.cuda,l->o.cuda,out_dev,qd,ld_,rd, + S,H,c->qk_nope,R,c->v_head,kvl,T,c->attn_scale) + :coli_cuda_attention_project_batch_dev(l->kv_b.cuda,l->o.cuda,out,qd,ld_,rd, + S,H,c->qk_nope,R,c->v_head,kvl,T,c->attn_scale); + m->t_acore+=now_s()-t0; +done: + free(chost); /* gli scratch device restano al contesto */ + return ok; +} +#endif + static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int pos_base, KVState *const *kvs, const int *positions, float *out){ Cfg *c=&m->c; int H=c->n_heads, D=c->hidden, qh=c->qk_head, vh=c->v_head; @@ -1631,12 +1817,24 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p * batch-union. matmul_qt_ex(...,0) keeps them on the EXACT int4 kernel: letting S>1 pull * them into IDOT is much faster but costs ~12% perplexity (measured). Batching alone is * bit-identical to upstream; the kernel switch is not. */ - matmul_qt_ex(QR, x, &l->q_a, S, 0); - for(int s=0;sq_lora; - rmsnorm(qr, qr, l->q_a_ln, c->q_lora, c->eps); } /* q_b legge il residuo NORMATO */ - matmul_qt_ex(Q, QR, &l->q_b, S, 0); - matmul_qt_ex(comp, x, &l->kv_a, S, 0); - for(int s=0;s=8&&layern_layers&&g_cuda_enabled&&c->kv_lora<=512&& + !(m->has_dsa&&pos_base+S>c->index_topk)&& + l->q_a.cuda_eligible&&l->q_b.cuda_eligible&&l->kv_a.cuda_eligible&& + l->kv_b.cuda_eligible&&l->o.cuda_eligible&& + qt_cuda_upload(&l->q_a)&&qt_cuda_upload(&l->q_b)&&qt_cuda_upload(&l->kv_a)&& + qt_cuda_upload(&l->kv_b)&&qt_cuda_upload(&l->o)) + pipe_done=attn_pipe_prefill(m,l,layer,x,0,S,pos_base,out,NULL); +#endif + if(!pipe_done){ + matmul_qt_ex(QR, x, &l->q_a, S, 0); + for(int s=0;sq_lora; + rmsnorm(qr, qr, l->q_a_ln, c->q_lora, c->eps); } /* q_b legge il residuo NORMATO */ + matmul_qt_ex(Q, QR, &l->q_b, S, 0); + matmul_qt_ex(comp, x, &l->kv_a, S, 0); + } + if(!pipe_done) for(int s=0;skv; int pos=positions?positions[s]:pos_base+s; float *qfull=Q+(int64_t)s*H*qh; @@ -1644,6 +1842,10 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p const float *cs=comp+(int64_t)s*cw; float *Ldst=coli_kv_row(ks->Lc[layer],pos,c->kv_lora); float *Rdst=coli_kv_row(ks->Rc[layer],pos,c->qk_rope); +#ifdef COLI_CUDA + if(ks==m->kv&&m->kv_dev_valid&&layer<=c->n_layers&&m->kv_dev_valid[layer]>pos) + m->kv_dev_valid[layer]=pos; /* riga riscritta: l'ombra si accorcia */ +#endif memcpy(Ldst, cs, c->kv_lora*sizeof(float)); rmsnorm(Ldst, Ldst, l->kv_a_ln, c->kv_lora, c->eps); /* latente normato */ memcpy(Rdst, cs+c->kv_lora, c->qk_rope*sizeof(float)); @@ -1720,7 +1922,17 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p * k/v per ogni token del contesto. Per linearita': * q·k_nope_t = (W_K^hT q_nope)·L_t ctx^h = W_V^h (Σ_t a_t L_t) * costo per step ~O(T·kv_lora) invece di O(T·H·(nope+vh)) del matmul kvb_all. */ - int absorb = kvs || g_absorb==1 || (g_absorb<0 && S<=4); + if(pipe_done){ + free(ctx); free(Q); free(QR); free(comp); + m->t_attn += now_s()-ta0; + return; + } + int cuda_absorb=0; +#ifdef COLI_CUDA + cuda_absorb=layern_layers&&!kvs&&g_cuda_enabled&&getenv("COLI_CUDA_ATTN")&& + atoi(getenv("COLI_CUDA_ATTN"))&&c->kv_lora<=512; +#endif + int absorb = kvs || g_absorb==1 || (g_absorb<0 && S<=4) || cuda_absorb; if(absorb && c->kv_lora<=512){ m->t_aproj+=now_s()-ta0; double tac=now_s(); int kvl=c->kv_lora, r0v=c->qk_nope; /* offset righe V dentro il blocco di testa */ @@ -1739,19 +1951,48 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p if(nt>sc_cap) sc_cap=nt; } float *sc_all = falloc((int64_t)omp_get_max_threads()*sc_cap); - int cuda_core=0; + int cuda_core=0,cuda_projected=0; #ifdef COLI_CUDA - if(S<=4&&g_cuda_enabled&&getenv("COLI_CUDA_ATTN")&&atoi(getenv("COLI_CUDA_ATTN"))&& + if(cuda_absorb&&l->n_kv_b_shard>1){ + int n=l->n_kv_b_shard,st0=m->kv_start[layer],nt=pos_base+S-st0,ok=1; + float *qs=falloc((int64_t)S*H*qh),*cs=falloc((int64_t)S*H*vh); + for(int d=0;dshard_h0[d]*S*qh+(int64_t)s*l->shard_hn[d]*qh, + Q+((int64_t)s*H+l->shard_h0[d])*qh,(size_t)l->shard_hn[d]*qh*sizeof(float)); + #pragma omp parallel for schedule(static) reduction(&:ok) + for(int d=0;dkv_b_shard[d], + cs+(int64_t)l->shard_h0[d]*S*vh,qs+(int64_t)l->shard_h0[d]*S*qh, + coli_kv_row(m->Lc[layer],st0,kvl),coli_kv_row(m->Rc[layer],st0,c->qk_rope), + S,l->shard_hn[d],c->qk_nope,c->qk_rope,vh,kvl,nt,c->attn_scale); + if(ok)for(int d=0;dshard_h0[d])*vh, + cs+(int64_t)l->shard_h0[d]*S*vh+(int64_t)s*l->shard_hn[d]*vh, + (size_t)l->shard_hn[d]*vh*sizeof(float)); + free(qs);free(cs);cuda_core=ok; + } else if(cuda_absorb&&l->kv_b.cuda_eligible&&l->o.cuda_eligible&& + qt_cuda_upload(&l->kv_b)&&qt_cuda_upload(&l->o)){ + int st0=m->kv_start[layer],nt=pos_base+S-st0; + cuda_core=cuda_projected=coli_cuda_attention_project_batch(l->kv_b.cuda,l->o.cuda,out,Q, + coli_kv_row(m->Lc[layer],st0,kvl),coli_kv_row(m->Rc[layer],st0,c->qk_rope), + S,H,c->qk_nope,c->qk_rope,vh,kvl,nt,c->attn_scale); + } else if(S<=4&&g_cuda_enabled&&getenv("COLI_CUDA_ATTN")&&atoi(getenv("COLI_CUDA_ATTN"))&& l->kv_b.cuda_eligible&&qt_cuda_upload(&l->kv_b)){ cuda_core=1; for(int s=0;skv;int pos=positions?positions[s]:pos_base+s; int st0=ks->kv_start[layer],nt=pos+1-st0; if(dnsel&&dnsel[s]>0){cuda_core=0;break;} - cuda_core=coli_cuda_attention_absorb(l->kv_b.cuda,ctx+(int64_t)s*H*vh, - Q+(int64_t)s*H*qh,coli_kv_row(ks->Lc[layer],st0,kvl), - coli_kv_row(ks->Rc[layer],st0,c->qk_rope),H,c->qk_nope,c->qk_rope, - vh,kvl,nt,c->attn_scale); + cuda_core=0; + if(g_cuda_pipe&&ks==m->kv&&layern_layers&&kv_dev_sync(m,l,layer,pos+1)) + cuda_core=coli_cuda_attention_absorb_kvdev(l->kv_b.cuda,ctx+(int64_t)s*H*vh, + Q+(int64_t)s*H*qh,m->kv_dev_L[layer]+(size_t)st0*kvl, + m->kv_dev_R[layer]+(size_t)st0*c->qk_rope,H,c->qk_nope,c->qk_rope, + vh,kvl,nt,c->attn_scale); + if(!cuda_core) + cuda_core=coli_cuda_attention_absorb(l->kv_b.cuda,ctx+(int64_t)s*H*vh, + Q+(int64_t)s*H*qh,coli_kv_row(ks->Lc[layer],st0,kvl), + coli_kv_row(ks->Rc[layer],st0,c->qk_rope),H,c->qk_nope,c->qk_rope, + vh,kvl,nt,c->attn_scale); } } #endif @@ -1786,7 +2027,7 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p } } m->t_acore+=now_s()-tac; double tao=now_s(); - matmul_qt(out, ctx, &l->o, S); m->t_aout+=now_s()-tao; + if(!cuda_projected){matmul_qt(out, ctx, &l->o, S);} m->t_aout+=now_s()-tao; free(ctx); free(Q); free(QR); free(comp); free(sc_all); m->t_attn += now_s()-ta0; return; @@ -1840,7 +2081,7 @@ static void attention(Model *m, Layer *l, int layer, float *x, int S, int pos_ba * una volta sola e moltiplicato per tutte le posizioni che lo usano (pesi letti 1 volta); * lo shared expert e' un unico matmul a S righe. Per posizione l'accumulo resta * nell'ordine (routed nel loro ordine di union, poi shared). */ -static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out){ +static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int with_shared){ if(g_pilot_real){ /* barriera cross-layer: prendi possesso di QUESTO layer e aspetta * l'eventuale load-pilota in volo sullo stesso layer (dopodiche' il * worker droppa ogni nuovo load <= layer -> ecache[layer] e' stabile @@ -1855,11 +2096,14 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out){ } } Cfg *c=&m->c; int D=c->hidden, E=c->n_experts, K=c->topk, I=c->moe_inter; - float *logit=falloc(E), *choice=falloc(E); + float *choice=falloc(E); int sI=c->moe_inter*c->n_shared; /* ---- FASE A: routing di tutte le S posizioni ---- */ int *idxs=malloc((size_t)S*K*sizeof(int)); float *ws=malloc((size_t)S*K*sizeof(float)); int *keff=malloc(S*sizeof(int)); + /* router in UN matmul batch: stessa matematica, via le S chiamate S=1 */ + float *logits_all=falloc((int64_t)S*E); + int pre_routed=0; (void)pre_routed; #ifdef COLI_METAL if(g_pre_idx){ /* routing gia' calcolata dal layer CB (GPU) */ memcpy(idxs,g_pre_idx,(size_t)S*K*sizeof(int)); @@ -1874,11 +2118,13 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out){ } for(int d=0;drouter, S, D, E); + if(!pre_routed) for(int s=0;srouter, 1, D, E); + float *logit=logits_all+(int64_t)s*E; for(int e=0;erouter_bias[e]; } int *idx=idxs+(int64_t)s*K; float *w=ws+(int64_t)s*K; int Ksel = g_topk>0 ? (g_topkws[q]; m->ws[q]=tmp; dst->used=(uint64_t)__atomic_add_fetch(&m->eclock,1,__ATOMIC_RELAXED); } } } - /* ---- FASE E: shared expert, un matmul a S righe (skipped se fuso nel blocco GPU) ---- */ - float *sg=falloc((int64_t)S*sI), *su=falloc((int64_t)S*sI); + /* ---- FASE E: shared expert (PIPE2: gia' sul device; Metal CB: gia' sommata) ---- */ + if(!with_shared) goto shared_done; + { + float *sg=NULL,*su=NULL;int shared_cuda=0; #ifdef COLI_METAL if(g_pre_sh){ for(int64_t z=0;z<(int64_t)S*D;z++) out[z]+=g_pre_sh[z]; shared_on_gpu=1; } + if(shared_on_gpu) shared_cuda=2; /* gia' sommato in out: salta calcolo e add */ #endif - if(!shared_on_gpu){ +#ifdef COLI_CUDA + int shared_min=getenv("COLI_CUDA_SHARED_W4A16_MIN_ROWS")? + atoi(getenv("COLI_CUDA_SHARED_W4A16_MIN_ROWS")):32; + if(shared_min<16)shared_min=16; + if(shared_cuda==0&&S>=shared_min&&!l->shared_w4a16_failed&&!omp_in_parallel()&&g_cuda_enabled&& + l->sh_gate.fmt==2&&l->sh_up.fmt==2&&l->sh_down.fmt==2&& + getenv("COLI_CUDA_SHARED_W4A16")&&atoi(getenv("COLI_CUDA_SHARED_W4A16"))&& + qt_cuda_upload(&l->sh_gate)&&qt_cuda_upload(&l->sh_up)&&qt_cuda_upload(&l->sh_down)){ + shared_cuda=coli_cuda_shared_mlp_w4a16(l->sh_gate.cuda,l->sh_up.cuda, + l->sh_down.cuda,hh,x,S); + if(!shared_cuda)l->shared_w4a16_failed=1; + } +#endif + if(!shared_cuda){ + sg=falloc((int64_t)S*sI);su=falloc((int64_t)S*sI); matmul_qt(sg, x, &l->sh_gate, S); matmul_qt(su, x, &l->sh_up, S); for(int64_t z=0;z<(int64_t)S*sI;z++) sg[z]=siluf(sg[z])*su[z]; matmul_qt(hh, sg, &l->sh_down, S); - for(int64_t z=0;z<(int64_t)S*D;z++) out[z]+=hh[z]; } - free(logit); free(choice); free(idxs); free(ws); free(keff); free(uniq); - free(xg); free(gg); free(uu); free(hh); free(rows); free(rw); free(sg); free(su); + if(shared_cuda!=2) for(int64_t z=0;z<(int64_t)S*D;z++) out[z]+=hh[z]; + free(sg); free(su); + } +shared_done: + free(logits_all); free(choice); free(idxs); free(ws); free(keff); free(uniq); + free(xg); free(gg); free(uu); free(hh); free(rows); free(rw); #ifdef COLI_CUDA - free(group_x); free(group_y); free(group_row); free(group_weight); + free(group_x);free(group_y); + free(group_row); free(group_weight); #endif } @@ -2393,6 +2663,65 @@ static void pilot_prefetch(Model *m, int lnext, const float *x, int S){ } /* forward di UN layer (usato dai 78 principali e dal layer MTP) */ +#ifdef COLI_CUDA +/* Inc.2a — intero layer SPARSO residente sul device del layer. x_dev entra e resta; + * lasciano il device solo: nrm post-attention (router + expert CPU + gather dei + * gruppi), i nuovi record KV, e la nrm pre-attention sui layer con indexer DSA. + * Ritorna 0 su errore: il chiamante ripristina lo snapshot e rifa' il layer su CPU. */ +static int pipe_layer_sparse(Model *m, Layer *l, int li, float *x_dev, int S, int pos_base, + float *nrm_host, float *out_host){ + Cfg *c=&m->c; int D=c->hidden, dev=l->kv_b.cuda_device; + int sI=c->moe_inter*c->n_shared; + size_t xb=(size_t)S*D*4; + 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); + 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(!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) */ + if(m->has_dsa && lin_layers && m->kv_start[li]==0 && c->idx_type[li]){ + if(!coli_cuda_pipe_download(dev,nrm_d,nrm_host,xb)) return 0; + int nh=c->index_nh, hd=c->index_hd; (void)nh; + for(int s=0;skv->Ic[li],pos,hd); + matmul_qt(kd, nrm_host+(int64_t)s*D, &m->ix_wk[li], 1); + layernorm(kd, m->ix_knw[li], m->ix_knb[li], hd, 1e-6f); + rope_interleave(kd, pos, c); + } + } + if(!attn_pipe_prefill(m,l,li,nrm_d,1,S,pos_base,NULL,y_d)) return 0; + if(!coli_cuda_pipe_add(dev,x_dev,y_d,(size_t)S*D)) return 0; /* prima mutazione */ + if(!coli_cuda_pipe_rmsnorm(dev,nrm_d,x_dev,w_post,S,D,c->eps)) return 0; + if(!coli_cuda_pipe_download(dev,nrm_d,nrm_host,xb)) return 0; + m->t_attn+=now_s()-ta; + /* expert routed su CPU/gruppi GPU come oggi (shared saltata: la fa il device) */ + moe(m,l,li,nrm_host,S,out_host,0); + double te=now_s(); + if(!coli_cuda_pipe_upload(dev,y_d,out_host,xb)) return 0; + if(!coli_cuda_pipe_add(dev,x_dev,y_d,(size_t)S*D)) return 0; + if(!coli_cuda_pipe_gemm(l->sh_gate.cuda,sg_d,nrm_d,S)) return 0; + if(!coli_cuda_pipe_gemm(l->sh_up.cuda,su_d,nrm_d,S)) return 0; + if(!coli_cuda_pipe_silu_mul(dev,sg_d,su_d,(size_t)S*sI)) return 0; + if(!coli_cuda_pipe_gemm(l->sh_down.cuda,y_d,sg_d,S)) return 0; + if(!coli_cuda_pipe_add(dev,x_dev,y_d,(size_t)S*D)) return 0; + if(!coli_cuda_pipe_sync(dev)) return 0; + m->t_emm+=now_s()-te; + return 1; +} +#endif + static void layer_forward_rows(Model *m, Layer *l, int li, float *x, int S, int pos_base, KVState *const *kvs, const int *positions, float *nrm, float *tmp){ Cfg *c=&m->c; int D=c->hidden; @@ -2459,7 +2788,7 @@ static void layer_forward_rows(Model *m, Layer *l, int li, float *x, int S, int if(g_pilot && S<=8 && li+1n_layers && m->L[li+1].sparse) pilot_prefetch(m,li+1,x,S); if(g_looka && S==1 && li+1n_layers && m->L[li+1].sparse) la_predict(m,li+1,x,1); for(int s=0;spost_ln, D, c->eps); - if(l->sparse) moe(m,l,li,nrm,S,tmp); else dense_mlp(l,nrm,S,D,c->dense_inter,tmp); + if(l->sparse) moe(m,l,li,nrm,S,tmp,1); else dense_mlp(l,nrm,S,D,c->dense_inter,tmp); for(int64_t j=0;j<(int64_t)S*D;j++) x[j]+=tmp[j]; } static void layer_forward(Model *m, Layer *l, int li, float *x, int S, int pos_base, float *nrm, float *tmp){ @@ -2474,13 +2803,53 @@ static void layers_forward_rows(Model *m, float *x, int S, int pos_base, pthread_mutex_unlock(&g_pilot_mx); } float *nrm=falloc((int64_t)S*D), *tmp=falloc((int64_t)S*D); +#ifdef COLI_CUDA + /* PIPE2 (Inc.2a): il residuo resta sul device del layer, saltando tra le schede + * ai confini di layer. x host diventa STALE finche' la residenza e' attiva. */ + float *x_dev=NULL; int x_dev_on=-1; + size_t xb=(size_t)S*(size_t)D*4; + int pipe2 = g_cuda_pipe>=2 && !kvs && S>=8 && g_cuda_enabled && c->kv_lora<=512 && + !(m->has_dsa && pos_base+S>c->index_topk); +#endif for(int i=0;in_layers;i++){ /* progresso su stderr per i batch grossi (prefill): il primo byte di risposta * puo' arrivare dopo MINUTI di streaming — al buio sembra un blocco. */ if(S>=8 && (i%4==0 || i==c->n_layers-1)) fprintf(stderr,"[prefill] layer %d/%d · %d token\n", i+1, c->n_layers, S); +#ifdef COLI_CUDA + Layer *l=&m->L[i]; + if(pipe2 && l->sparse && in_layers && + l->q_a.cuda_eligible&&l->q_b.cuda_eligible&&l->kv_a.cuda_eligible&& + l->kv_b.cuda_eligible&&l->o.cuda_eligible&& + qt_cuda_upload(&l->q_a)&&qt_cuda_upload(&l->q_b)&&qt_cuda_upload(&l->kv_a)&& + qt_cuda_upload(&l->kv_b)&&qt_cuda_upload(&l->o)&& + l->q_a.cuda_device==l->kv_b.cuda_device&&l->q_b.cuda_device==l->kv_b.cuda_device&& + l->kv_a.cuda_device==l->kv_b.cuda_device&&l->o.cuda_device==l->kv_b.cuda_device){ + int dev=l->kv_b.cuda_device, ok=1; + float *dst=coli_cuda_pipe_scratch(dev,15,xb); + if(dst){ + if(x_dev_on<0) ok=coli_cuda_pipe_upload(dev,dst,x,xb); + else if(x_dev_on!=dev){ ok=coli_cuda_pipe_peer_copy(dev,dst,x_dev_on,x_dev,xb); } + else dst=x_dev; + if(ok){ + x_dev=dst; x_dev_on=dev; + if(pipe_layer_sparse(m,l,i,x_dev,S,pos_base,nrm,tmp)) continue; + /* fallback: snapshot -> host, layer rifatto sul percorso CPU */ + coli_cuda_pipe_peer_copy(dev,x_dev,dev,coli_cuda_pipe_scratch(dev,14,xb),xb); + coli_cuda_pipe_download(dev,x_dev,x,xb); + x_dev_on=-1; + }else x_dev_on=-1; + } + } else if(x_dev_on>=0){ /* layer fuori pipe: il residuo torna a casa */ + coli_cuda_pipe_download(x_dev_on,x_dev,x,xb); + x_dev_on=-1; + } +#endif layer_forward_rows(m,&m->L[i],i,x,S,pos_base,kvs,positions,nrm,tmp); } +#ifdef COLI_CUDA + if(x_dev_on>=0) coli_cuda_pipe_download(x_dev_on,x_dev,x,xb); +#endif free(nrm); free(tmp); } static void layers_forward(Model *m, float *x, int S, int pos_base){ @@ -2490,6 +2859,14 @@ static void layers_forward(Model *m, float *x, int S, int pos_base){ static void kv_alloc(Model *m, int max_t){ Cfg *c=&m->c; KVState *k=m->kv; +#ifdef COLI_CUDA + if(m->kv_dev_L) for(int i=0;in_layers+1;i++){ /* dimensioni cambiate: ombra da rifare */ + if(m->kv_dev_L[i]){ coli_cuda_pipe_free(m->L[in_layers?i:0].kv_b.cuda_device,m->kv_dev_L[i]); m->kv_dev_L[i]=NULL; } + if(m->kv_dev_R[i]){ coli_cuda_pipe_free(m->L[in_layers?i:0].kv_b.cuda_device,m->kv_dev_R[i]); m->kv_dev_R[i]=NULL; } + m->kv_dev_valid[i]=0; + } +#endif + if(k->Lc){ for(int i=0;in_layers+1;i++){ free(k->Lc[i]); free(k->Rc[i]); } free(k->Lc); free(k->Rc); } if(k->Lc){ for(int i=0;in_layers+1;i++){ #ifdef COLI_METAL if(g_metal_enabled){ coli_metal_unregister(k->Lc[i]); coli_metal_unregister(k->Rc[i]); } @@ -2522,6 +2899,8 @@ static void kv_alloc(Model *m, int max_t){ } static void kv_bind(Model *m, KVState *k){ + if(m->kv!=k && m->kv_dev_valid) /* ombra legata al KVState corrente */ + for(int i=0;ic.n_layers+1;i++) m->kv_dev_valid[i]=0; m->kv=k; m->Lc=k->Lc; m->Rc=k->Rc; m->Ic=k->Ic; m->max_t=k->max_t; m->kv_start=k->kv_start; } @@ -3034,6 +3413,7 @@ static void run_text(Model *m, const char *snap, const char *prompt, int ngen){ repin_pass_limit(m,limit); /* prompt routing seeds every GPU layer */ } prefill_t=now_s()-prefill_t; + printf("PROFILO PREFILL (%.2fs):\n",prefill_t); profile_print(m,prefill_t); m->hits=m->miss=m->ereq=m->gpu_expert_calls=0; m->n_emit=m->n_fw=0; g_last_repin=0; @@ -3329,6 +3709,7 @@ static void mux_done(Model *m, ServeCtx *sc, ServeReq *r){ double dt=now_s()-r->started; if(dt<1e-6) dt=1e-6; double dh=(double)(m->hits-r->hits0), dm=(double)(m->miss-r->miss0); hwinfo_emit(m); + usage_save(m); /* la cache che impara non deve aspettare l'uscita */ tiers_emit(m); emap_emit(m); hits_emit(m); @@ -4204,6 +4585,7 @@ int main(int argc, char **argv){ if(!g_cuda_enabled){ fprintf(stderr,"[CUDA] requested backend is unavailable\n"); return 2; } } g_cuda_dense=getenv("CUDA_DENSE")?atoi(getenv("CUDA_DENSE")):0; + g_cuda_pipe=getenv("COLI_CUDA_PIPE")?atoi(getenv("COLI_CUDA_PIPE")):0; const char *cuda_expert=getenv("CUDA_EXPERT_GB"); g_cuda_expert_auto=cuda_expert&&!strcmp(cuda_expert,"auto"); g_cuda_expert_gb=cuda_expert&&!g_cuda_expert_auto?atof(cuda_expert):0; diff --git a/c/tests/test_i4_acc512 b/c/tests/test_compat_direct similarity index 72% rename from c/tests/test_i4_acc512 rename to c/tests/test_compat_direct index 76f68a3..074a7ba 100755 Binary files a/c/tests/test_i4_acc512 and b/c/tests/test_compat_direct differ diff --git a/c/tests/test_decode_batch b/c/tests/test_decode_batch deleted file mode 100755 index 989fe42..0000000 Binary files a/c/tests/test_decode_batch and /dev/null differ diff --git a/c/tests/test_idot b/c/tests/test_idot deleted file mode 100755 index bab981f..0000000 Binary files a/c/tests/test_idot and /dev/null differ diff --git a/c/tests/test_pipe_cuda.cu b/c/tests/test_pipe_cuda.cu new file mode 100644 index 0000000..f20a9e5 --- /dev/null +++ b/c/tests/test_pipe_cuda.cu @@ -0,0 +1,137 @@ +/* Unit tests for the resident-pipeline primitives (Inc.0). + * Each primitive is checked against the exact CPU math from glm.c. + * Build: nvcc backend_cuda.cu tests/test_pipe_cuda.cu -o pipe_test && ./pipe_test */ +#include +#include +#include +#include +#include +#include "../backend_cuda.h" + +static uint64_t rng=0x9E3779B97F4A7C15ULL; +static float rndf(void){ rng^=rng<<13; rng^=rng>>7; rng^=rng<<17; + return (float)((double)(rng>>11)/9007199254740992.0*2.0-1.0); } + +static void ref_rmsnorm(float *out,const float *x,const float *w,int D,float eps){ + double ms=0; for(int i=0;i1.f?fabsf(b[i]):1.f; + if(d/m>worst) worst=d/m; + } + printf(" %-14s max rel %.3e %s\n",what,worst,worst<=tol?"ok":"FAIL"); + return worst<=tol; +} + +int main(void){ + int dev0=0; + if(!coli_cuda_init(&dev0,1)){ puts("pipe tests: skipped (no CUDA device)"); return 0; } + int ok=1; + enum { S=7, D=6144, I=2048, R=64, H=4, QH=192 }; + + /* rmsnorm */ + { + float *x=(float*)malloc((size_t)S*D*4), *w=(float*)malloc(D*4); + float *got=(float*)malloc((size_t)S*D*4), *ref=(float*)malloc((size_t)S*D*4); + for(size_t i=0;i<(size_t)S*D;i++) x[i]=rndf()*3; + for(int i=0;i=3 reps, +# or snapshot/restore the usage file around the battery. +set -u +GLM="${1:-./glm}" +REPS="${REPS:-3}" +export TEMP=0 DRAFT=0 +[ -n "${PIPE:-}" ] && export COLI_CUDA_PIPE="$PIPE" + +SHORT_PROMPT="Explain why the sky is blue in simple terms." +LONGDOC_PROMPT=$(python3 - <<'PY' +base=("蜂鸟是世界上最小的鸟类之一，主要分布在美洲大陆。它们的翅膀每秒可以扇动五十到八十次，" +"使它们能够在空中悬停、倒飞，甚至垂直起降。蜂鸟的新陈代谢极快，心跳每分钟可达一千两百次，" +"因此它们必须不断进食来维持能量。它们的主要食物是花蜜，同时也会捕食小型昆虫和蜘蛛来补充蛋白质。" +"蜂鸟的喙细长，适合深入花朵内部吸食花蜜，舌头呈管状，可以快速伸缩。在授粉过程中，" +"蜂鸟扮演着重要的角色，许多美洲植物依赖蜂鸟传粉。") +print((base*12)+"请根据上文回答：蜂鸟的主要食物是什么？") +PY +) + +scenario(){ # name prompt ngen + local name=$1 prompt=$2 ngen=$3 + for r in $(seq 1 "$REPS"); do + local log; log=$(mktemp) + NGEN=$ngen PROMPT="$prompt" "$GLM" 64 4 4 >"$log" 2>&1 + local line; line=$(grep -aE "prefill .* decode .*tok/s" "$log" | tail -1) + local head; head=$(grep -av "^\[" "$log" | grep -avE "PROFILE|PROFILO|ATTENTION|expert|CUDA|prefill|specul|^---|TOPP|stop|Motore|caricato|prompt:|token" \ + | tail -1 | tr -d '\n' | cut -c1-48) + printf "%-10s r%d %s\n" "$name" "$r" "$line" + printf "%-10s r%d text: %s\n" "$name" "$r" "$head" + rm -f "$log" + done +} + +echo "== bench_ux: reps=$REPS pipe=${COLI_CUDA_PIPE:-unset} $(date +%F' '%T) ==" +scenario chat "$SHORT_PROMPT" 96 +scenario longdoc "$LONGDOC_PROMPT" 96 +echo "== bench_ux done $(date +%T) ==" diff --git a/c/tools/expert_atlas.py b/c/tools/expert_atlas.py new file mode 100644 index 0000000..dd0a2d6 --- /dev/null +++ b/c/tools/expert_atlas.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Expert Atlas (#175): measure per-expert topic affinity by diffing .coli_usage +across themed probe batches, served through a running colibri API server. + +Protocol per category: snapshot .coli_usage -> send probes -> snapshot again; +the delta is that category's expert-activation spectrum. One engine load total. + +Output: experts.json — for every (layer, expert): counts per category, +normalized affinity, entropy, and a "specialist" label when one topic dominates. + +Usage (server already running with the model): + python3 tools/expert_atlas.py --api http://127.0.0.1:8000 \ + --usage /path/to/model/.coli_usage --out experts.json --ngen 64 +""" +import argparse, json, math, time, urllib.request + +PROBES = { + "code": [ + "Write a Python function that parses a CSV file and returns a dict keyed by the first column.", + "Explain the difference between a mutex and a semaphore, with a C example.", + "Refactor this into idiomatic Rust: for i in range(len(xs)): total += xs[i] * 2", + ], + "math": [ + "Prove that the square root of 2 is irrational.", + "Compute the derivative of x^3 * ln(x) and explain each step.", + "A fair die is rolled 4 times. What is the probability of at least one six?", + ], + "chinese": [ + "请用中文解释一下什么是光合作用,以及它对地球生态系统的重要性。", + "把这句话翻译成中文并解释语法:The early bird catches the worm.", + "写一段关于秋天的短文,一百字左右。", + ], + "english_prose": [ + "Write a vivid paragraph describing an old lighthouse keeper watching a storm arrive.", + "Summarize the plot of Romeo and Juliet in three sentences.", + "Continue this story: The last train left the station, and Maria realized her mistake.", + ], + "science": [ + "Explain how mRNA vaccines work at the cellular level.", + "Why is the sky blue during the day but red at sunset?", + "Describe the life cycle of a massive star, from formation to supernova.", + ], + "law": [ + "Explain the difference between a patent, a trademark, and a copyright.", + "What are the key elements required to form a legally binding contract?", + "Summarize what 'due process' means in constitutional law.", + ], + "poetry": [ + "Write a short poem about a hummingbird in the style of Emily Dickinson.", + "Compose a haiku about winter rain, then explain its imagery.", + "Write four rhyming lines about the sea at night.", + ], + "structured": [ + 'Convert to JSON: name Alice, age 30, hobbies reading and chess, address 5 Oak St.', + "Write a SQL query returning the top 5 customers by total order value, with the schema you assume.", + "Write a regex that matches ISO-8601 dates and explain each part.", + ], + "translation": [ + "Translate into French, German and Spanish: 'Knowledge is the only treasure that grows when shared.'", + "Translate this Italian sentence to English and comment on nuance: 'In bocca al lupo per domani.'", + "Translate into Japanese: 'The meeting has been moved to next Tuesday afternoon.'", + ], + "casual": [ + "Hey! Any tips for staying awake during boring afternoon meetings?", + "What should I cook tonight? I have eggs, rice, tomatoes and some cheese.", + "My friend is always late. How do I tell them it bothers me without being rude?", + ], +} + + +def read_usage(path): + counts = {} + try: + with open(path) as f: + for line in f: + p = line.split() + if len(p) == 3: + counts[(int(p[0]), int(p[1]))] = int(p[2]) + except FileNotFoundError: + pass + return counts + + +def diff(after, before): + return {k: v - before.get(k, 0) for k, v in after.items() if v - before.get(k, 0) > 0} + + +def chat(api, prompt, ngen): + body = json.dumps({"model": "glm-5.2-colibri", "stream": False, "max_tokens": ngen, + "messages": [{"role": "user", "content": prompt}]}).encode() + req = urllib.request.Request(f"{api}/v1/chat/completions", data=body, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=600) as r: + json.load(r) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--api", default="http://127.0.0.1:8000") + ap.add_argument("--usage", required=True, help="path to the model's .coli_usage") + ap.add_argument("--out", default="experts.json") + ap.add_argument("--ngen", type=int, default=64) + a = ap.parse_args() + + spectra = {} + for cat, prompts in PROBES.items(): + before = read_usage(a.usage) + t0 = time.time() + for p in prompts: + chat(a.api, p, a.ngen) + time.sleep(2) # let the engine flush .coli_usage + spectra[cat] = diff(read_usage(a.usage), before) + total = sum(spectra[cat].values()) + print(f"[{cat}] {len(spectra[cat])} experts touched, {total} selections, {time.time()-t0:.0f}s", flush=True) + + cats = list(PROBES.keys()) + experts = {} + for cat, spec in spectra.items(): + for k, v in spec.items(): + experts.setdefault(k, {c: 0 for c in cats})[cat] = v + + atlas = {} + for (layer, eid), counts in experts.items(): + total = sum(counts.values()) + if total < 8: + continue # too few observations to characterise + aff = {c: v / total for c, v in counts.items() if v} + ent = -sum(p * math.log2(p) for p in aff.values()) + top = max(aff, key=aff.get) + label = f"specialist: {top}" if aff[top] >= 0.45 and ent < 2.2 else "generalist" + atlas[f"{layer}:{eid}"] = {"counts": counts, "affinity": {c: round(p, 3) for c, p in aff.items()}, + "entropy": round(ent, 2), "top": top, "label": label} + + spec_n = sum(1 for v in atlas.values() if v["label"].startswith("specialist")) + with open(a.out, "w") as f: + json.dump({"categories": cats, "ngen": a.ngen, "experts": atlas}, f) + print(f"\natlas: {len(atlas)} experts characterised, {spec_n} specialists -> {a.out}") + + +if __name__ == "__main__": + main() diff --git a/web/src/Brain.tsx b/web/src/Brain.tsx index ac20695..6811238 100644 --- a/web/src/Brain.tsx +++ b/web/src/Brain.tsx @@ -4,6 +4,7 @@ import { BrainCircuit, Flame, Layers } from "lucide-react" import { endpoint } from "@/lib/api" interface ExpertMap { rows: number; cols: number; map: string; hits: string; seq: number } +interface AtlasEntry { affinity: Record; entropy: number; top: string; label: string } const TIER_NAME = ["Disk", "RAM", "VRAM"] const TIER_RGB: [number, number, number][] = [[58, 71, 80], [90, 155, 216], [78, 214, 165]] @@ -26,11 +27,19 @@ export function Brain({ baseUrl, apiKey, connected }: { baseUrl: string; apiKey: const wrapRef = useRef(null) const [wrapSize, setWrapSize] = useState({ w: 1200, h: 700 }) const [data, setData] = useState(null) + const [atlas, setAtlas] = useState | null>(null) const [tip, setTip] = useState<{ x: number; y: number; row: number; col: number; tier: number; heat: number } | null>(null) const pulseRef = useRef(null) // per-expert pulse intensity 0..1 const lastSeq = useRef(0) const rafRef = useRef(0) + // load the expert atlas if published (measured topic affinity, #175) + useEffect(() => { + fetch("/experts.json").then(r => r.ok ? r.json() : null).then(d => { + if (d?.experts) setAtlas(d.experts) + }).catch(() => {}) + }, []) + // track container size for responsive cell sizing useEffect(() => { const el = wrapRef.current @@ -146,14 +155,26 @@ export function Brain({ baseUrl, apiKey, connected }: { baseUrl: string; apiKey: setTip(null)} /> {!connected &&

Connect to the engine to see the cortex.

} - {tip && data && ( + {tip && data && (() => { + const isMtp = tip.row === data.rows - 1 + const realLayer = isMtp ? 78 : tip.row + 3 + const entry = atlas?.[`${realLayer}:${tip.col}`] + return (
-
Layer row {tip.row}{tip.row === data.rows - 1 ? " (MTP)" : ""} · Expert {tip.col}
+
Layer {realLayer}{isMtp ? " (MTP)" : ""} · Expert {tip.col}
Tier: {TIER_NAME[tip.tier]}
Heat: {tip.heat === 0 ? "never routed" : `~2^${tip.heat} selections`}
-
{depthRole(tip.row, data.rows, tip.row === data.rows - 1)}
+ {entry ? <> +
+ {entry.label.startsWith("specialist") ? `⭐ Specialist: ${entry.top}` : "Generalist"} + (entropy {entry.entropy}) +
+
{Object.entries(entry.affinity).sort((a, b) => b[1] - a[1]).slice(0, 3) + .map(([c, p]) => `${c} ${Math.round(p * 100)}%`).join(" · ")}
+ :
{depthRole(tip.row, data.rows, isMtp)}
}
- )} + ) + })()} ) } diff --git a/web/src/index.css b/web/src/index.css index 59d5bc6..b428805 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -146,3 +146,8 @@ button:focus-visible, input:focus-visible, textarea:focus-visible, select:focus- .brain-legend { gap: 8px; font-size: 10px; } .brain-tip { max-width: 220px; font-size: 10px; } } + +/* atlas hover extras */ +.brain-tip-spec { color: var(--primary); font-weight: 700; } +.brain-tip-spec small, .brain-tip-aff { color: #8b9aa3; font-weight: 400; } +.brain-tip-aff { font-size: 10px; }