url/colibri
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #374 from ZacharyZcR/perf/p0-execution-profile

Browse Source 
profile CPU/GPU tier execution and residual P2P
This commit is contained in:
Vincenzo
2026-07-18 00:57:35 +02:00
committed by GitHub
parent d5dac9bd6c 570d738ed5
commit 6ce6cc4380
3 changed files with 81 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
c/glm.c
+50 -10
View File
@@ -201,7 +201,10 @@ typedef struct {
uint64_t route_slots, route_swaps; /* CACHE_ROUTE: slots chosen / substituted vs true top-K */
uint64_t route_agree_hit, route_agree_tot; /* ROUTE_AGREE: |chosen ∩ true top-K| / K */
double route_kl_sum; uint64_t route_kl_n; /* mean KL(true||chosen) on gate mass */
double t_ewait, t_emm, t_attn, t_kvb, t_head;/* profiling: dove va il tempo (wall del
double t_ewait, t_emm, t_ecpu, t_egpu, t_route, t_p2p, t_attn, t_kvb, t_head;
uint64_t n_p2p; /* P0 execution profile: tier split + residual hops */
uint64_t cpu_expert_rows; int64_t cpu_expert_bytes;
/* profiling: dove va il tempo (wall del
* thread di compute; il servizio disco
* overlappato vive in g_edisk_ns) */
double t_aproj,t_acore,t_aout; /* attention breakdown */
@@ -307,15 +310,17 @@ static uint64_t g_prof_nlat; /* forwards recorded (monotonic
static void prof_lat(double s){ g_prof_lat[g_prof_nlat++ % PROF_LAT_CAP]=s; }
/* snapshot for windowed reports (serve mode: one report per turn) */
typedef struct {
double edisk,ewait,emm,attn,head;
int64_t io; uint64_t hits,miss,ereq,n_fw,n_emit,nlat;
double edisk,ewait,emm,ecpu,egpu,route,p2p,attn,head;
int64_t io,cpu_bytes; uint64_t hits,miss,ereq,n_fw,n_emit,nlat,n_p2p,cpu_rows;
} ProfBase;
static void prof_base(Model *m, ProfBase *b){
b->edisk=edisk_s(); b->ewait=m->t_ewait; b->emm=m->t_emm;
b->ecpu=m->t_ecpu; b->egpu=m->t_egpu; b->route=m->t_route; b->p2p=m->t_p2p;
b->attn=m->t_attn; b->head=m->t_head;
b->io=atomic_load_explicit(&g_prof_io,memory_order_relaxed);
b->hits=m->hits; b->miss=m->miss; b->ereq=m->ereq;
b->n_fw=m->n_fw; b->n_emit=m->n_emit; b->nlat=g_prof_nlat;
b->n_fw=m->n_fw; b->n_emit=m->n_emit; b->nlat=g_prof_nlat; b->n_p2p=m->n_p2p;
b->cpu_bytes=m->cpu_expert_bytes;b->cpu_rows=m->cpu_expert_rows;
}
static float *falloc(int64_t n){
@@ -2867,6 +2872,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
if(!rank_buf||!rank_w){ free(rank_buf); free(rank_w); rank_buf=NULL; rank_w=NULL; do_cache_route=0; }
}
/* ---- FASE A: routing di tutte le S posizioni ---- */
double route_t0=g_prof?now_s():0;
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 */
@@ -3016,6 +3022,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
for(int d=0;d<D;d++) out[(int64_t)s*D+d]=0;
}
free(rank_buf); free(rank_w);
if(g_prof)m->t_route+=now_s()-route_t0;
if(g_route_fp) g_route_call++;
if(g_couple && cp_pred && S<=8)
for(int s2=0;s2<S;s2++) couple_prefetch(m,layer,idxs+(int64_t)s2*K,keff[s2]);
@@ -3282,7 +3289,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
coli_cuda_expert_mlp(e->g.cuda,e->u.cuda,e->d.cuda,hh,xg,nr)){
for(int r=0;r<nr;r++){ float *os=out+(int64_t)rows[r]*D,wgt=rw[r],*hr=hh+(int64_t)r*D;
for(int d=0;d<D;d++) os[d]+=wgt*hr[d]; }
m->t_emm+=now_s()-t0; continue;
double dt=now_s()-t0;m->t_emm+=dt;if(g_prof)m->t_egpu+=dt;continue;
}
if(!e->slab) expert_host_ensure(m,layer,e);
#endif
@@ -3291,7 +3298,9 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
matmul_qt(hh, gg, &e->d, nr);
for(int r=0;r<nr;r++){ float *os=out+(int64_t)rows[r]*D, wgt=rw[r], *hr=hh+(int64_t)r*D;
for(int d=0;d<D;d++) os[d]+=wgt*hr[d]; }
m->t_emm += now_s()-t0;
double dt=now_s()-t0;m->t_emm+=dt;if(g_prof){m->t_ecpu+=dt;
m->cpu_expert_bytes+=qt_bytes(&e->g)+qt_bytes(&e->u)+qt_bytes(&e->d);
m->cpu_expert_rows+=(uint64_t)nr;}
}
#ifdef COLI_CUDA
ColiCudaTensor *dev_g[COLI_CUDA_MAX_DEVICES][64],*dev_u[COLI_CUDA_MAX_DEVICES][64];
@@ -3299,6 +3308,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
int dev_rows[COLI_CUDA_MAX_DEVICES][64],dev_which[COLI_CUDA_MAX_DEVICES][64];
int dev_nc[COLI_CUDA_MAX_DEVICES]={0},dev_total[COLI_CUDA_MAX_DEVICES]={0};
int dev_off[COLI_CUDA_MAX_DEVICES]={0},dev_ok[COLI_CUDA_MAX_DEVICES]={0};
double dev_time[COLI_CUDA_MAX_DEVICES]={0};
for(int di=0;di<g_cuda_ndev;di++) for(int q=0;q<ngroup;q++)
if(group_e[q]->g.cuda_device==g_cuda_devices[di]) dev_total[di]+=group_n[q];
for(int di=1;di<g_cuda_ndev;di++) dev_off[di]=dev_off[di-1]+dev_total[di-1];
@@ -3315,21 +3325,28 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
}
double tg=now_s();
#pragma omp parallel for if(g_cuda_ndev>1) schedule(static)
for(int di=0;di<g_cuda_ndev;di++) if(dev_nc[di])
for(int di=0;di<g_cuda_ndev;di++) if(dev_nc[di]){
double td=g_prof?now_s():0;
dev_ok[di]=coli_cuda_expert_group(dev_g[di],dev_u[di],dev_d[di],dev_rows[di],dev_nc[di],
group_y+(int64_t)dev_off[di]*D,group_x+(int64_t)dev_off[di]*D);
if(g_prof)dev_time[di]=now_s()-td;
}
for(int di=0;di<g_cuda_ndev;di++){
int off=dev_off[di];
for(int q=0;q<dev_nc[di];q++){
int gi=dev_which[di][q],nr=group_n[gi]; ESlot *e=group_e[gi];
if(!dev_ok[di]){
for(int r=0;r<nr;r++) memcpy(xg+(int64_t)r*D,x+(int64_t)group_row[(int64_t)gi*S+r]*D,D*sizeof(float));
double tc=g_prof?now_s():0;
if(!coli_cuda_expert_mlp(e->g.cuda,e->u.cuda,e->d.cuda,hh,xg,nr)){
expert_host_ensure(m,layer,e);
expert_gate_up(gg,uu,xg,&e->g,&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);
if(g_prof){m->cpu_expert_bytes+=qt_bytes(&e->g)+qt_bytes(&e->u)+qt_bytes(&e->d);
m->cpu_expert_rows+=(uint64_t)nr;}
}
if(g_prof)m->t_ecpu+=now_s()-tc;
}
float *src=dev_ok[di]?group_y+(int64_t)off*D:hh;
for(int r=0;r<nr;r++){ float *os=out+(int64_t)group_row[(int64_t)gi*S+r]*D,wgt=group_weight[(int64_t)gi*S+r];
@@ -3337,6 +3354,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
off+=nr;
}
}
if(g_prof){double mx=0;for(int di=0;di<g_cuda_ndev;di++)if(dev_time[di]>mx)mx=dev_time[di];m->t_egpu+=mx;}
m->t_emm+=now_s()-tg;
#endif
/* No drain barrier: the per-expert pipe_wait(qof[j]) above (issued for every
@@ -3944,7 +3962,11 @@ static void layers_forward_rows(Model *m, float *x, int S, int pos_base,
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 if(x_dev_on!=dev){
double tp=g_prof?now_s():0;
ok=coli_cuda_pipe_peer_copy(dev,dst,x_dev_on,x_dev,xb);
if(g_prof){m->t_p2p+=now_s()-tp;m->n_p2p++;}
}
else dst=x_dev;
if(ok){
x_dev=dst; x_dev_on=dev;
@@ -4577,6 +4599,11 @@ static void profile_print(Model *m, double elapsed){
edisk_s(),m->t_ewait,m->t_emm,m->t_attn,m->t_kvb,m->t_head,elapsed-accounted);
printf("ATTENTION: projection/RoPE %.3fs | score-softmax-value %.3fs | output projection %.3fs\n",
m->t_aproj,m->t_acore,m->t_aout);
if(g_prof)printf("P0-EXEC: routed CPU %.3fs / %.2f GB/s (%llu row) | routed GPU critical %.3fs | router %.3fs | residual P2P %.3fs / %llu hop | orchestration %.3fs\n",
m->t_ecpu,m->t_ecpu>0?m->cpu_expert_bytes/1e9/m->t_ecpu:0.0,
(unsigned long long)m->cpu_expert_rows,m->t_egpu,m->t_route,m->t_p2p,(unsigned long long)m->n_p2p,
elapsed-m->t_ewait-m->t_emm-m->t_attn-m->t_head-m->t_route-m->t_p2p>0?
elapsed-m->t_ewait-m->t_emm-m->t_attn-m->t_head-m->t_route-m->t_p2p:0);
#ifdef COLI_METAL
if(g_metal_enabled){ uint64_t ok=0,fb=0,ex=0; double su=0,gp=0,sc=0;
coli_metal_moe_counts(&ok,&fb,&ex); coli_metal_moe_times(&su,&gp,&sc);
@@ -4590,6 +4617,8 @@ static void profile_print(Model *m, double elapsed){
static void profile_reset(Model *m){
m->t_ewait=m->t_emm=m->t_attn=m->t_kvb=m->t_head=0;
m->t_ecpu=m->t_egpu=m->t_route=m->t_p2p=0;m->n_p2p=0;
m->cpu_expert_bytes=0;m->cpu_expert_rows=0;
m->t_aproj=m->t_acore=m->t_aout=0;
atomic_store_explicit(&g_edisk_ns,0,memory_order_relaxed);
}
@@ -4634,11 +4663,22 @@ static void prof_report(Model *m, const ProfBase *b, double elapsed, int tokens,
io_svc,io_w);
fprintf(f,"[PROF] resident experts: %d pinned (%.1f GB) + %d in LRU (%.1f GB, cap %d/layer)\n",
pinned,pinned*eb/1e9,lru,lru*eb/1e9,m->ecap);
double emm=m->t_emm-b->emm, attn=m->t_attn-b->attn, head=m->t_head-b->head;
double other=elapsed-io_w-emm-attn-head; if(other<0) other=0;
double emm=m->t_emm-b->emm, ecpu=m->t_ecpu-b->ecpu, egpu=m->t_egpu-b->egpu;
double route=m->t_route-b->route,p2p=m->t_p2p-b->p2p;
uint64_t np2p=m->n_p2p-b->n_p2p;
int64_t cpu_bytes=m->cpu_expert_bytes-b->cpu_bytes;
uint64_t cpu_rows=m->cpu_expert_rows-b->cpu_rows;
double attn=m->t_attn-b->attn, head=m->t_head-b->head;
double other=elapsed-io_w-emm-attn-head-route-p2p; if(other<0) other=0;
double f_io=io_w/elapsed, f_emm=emm/elapsed, f_attn=attn/elapsed;
fprintf(f,"[PROF] time shares: expert-I/O %.0f%% | expert-matmul %.0f%% | attention %.0f%% | lm_head %.0f%% | other %.0f%%\n",
100*f_io,100*f_emm,100*f_attn,100*head/elapsed,100*other/elapsed);
double slow=ecpu>egpu?ecpu:egpu,fast=ecpu<egpu?ecpu:egpu;
fprintf(f,"[PROF] P0 execution: routed CPU %.3fs / %.2f GB/s (%llu row) | routed GPU critical %.3fs | tier straggler %.2fx | "
"router %.3fs | residual P2P %.3fs (%llu hop, %.3f ms/hop) | orchestration %.3fs\n",
ecpu,ecpu>0?cpu_bytes/1e9/ecpu:0.0,(unsigned long long)cpu_rows,
egpu,fast>1e-9?slow/fast:0.0,route,p2p,(unsigned long long)np2p,
np2p?p2p*1e3/np2p:0.0,other);
if(f_io>=0.30){
fprintf(f,"[PROF] verdict: I/O-bound — %.0f%% of the time waits on expert reads (hit %.0f%%).",100*f_io,hitp);
if(hitp<90) fprintf(f," More cache is the lever: raise RAM_GB (or add RAM).");
c/tests/test_benchmark_cuda_fixture.py
+5 -1
View File
@@ -1,11 +1,12 @@
import unittest
from tools.benchmark_cuda_fixture import parse_output
from tools.benchmark_cuda_fixture import parse_output, parse_p0
SAMPLE = """
REPLAY decode: 4 tokens | 12.34 tok/s
PROFILE: expert-disk 1.25s | expert-matmul 2.50s | attention 0.75s | lm_head 0.10s | other -0.05s
P0-EXEC: routed CPU 1.200s / 123.40 GB/s (456 row) | routed GPU critical 0.150s | router 0.200s | residual P2P 0.030s / 75 hop | orchestration 0.100s
"""
@@ -19,6 +20,9 @@ class ParseOutputTest(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "benchmark output missing"):
parse_output("REPLAY decode: 4 tokens | 12.34 tok/s", "engine failed")
def test_extracts_p0_profile(self):
self.assertEqual(parse_p0(SAMPLE), [1.2, 123.4, 456.0, 0.15, 0.2, 0.03, 75.0, 0.1])
if __name__ == "__main__":
unittest.main()
c/tools/benchmark_cuda_fixture.py
+26 -3
View File
@@ -17,6 +17,12 @@ PROFILE_RE = re.compile(
r"\| attention ([0-9.]+)s .* lm_head ([0-9.]+)s \| other ([0-9.-]+)s"
)
PROFILE_KEYS = ("disk", "expert_matmul", "attention", "lm_head", "other")
P0_RE = re.compile(
r"P0-EXEC: routed CPU ([0-9.]+)s / ([0-9.]+) GB/s \(([0-9]+) row\) \| routed GPU critical ([0-9.]+)s \| "
r"router ([0-9.]+)s \| residual P2P ([0-9.]+)s / ([0-9]+) hop \| orchestration ([0-9.]+)s"
)
P0_KEYS = ("routed_cpu", "routed_cpu_gb_s", "routed_cpu_rows", "routed_gpu_critical",
"router", "p2p", "p2p_hops", "orchestration")
def parse_output(stdout: str, stderr: str = "") -> tuple[float, list[float]]:
@@ -30,11 +36,20 @@ def parse_output(stdout: str, stderr: str = "") -> tuple[float, list[float]]:
return float(speed.group(1)), [disk] + [float(value) for value in rest]
def execute(engine: str, env: dict[str, str]) -> tuple[float, list[float]]:
def parse_p0(stdout: str) -> list[float]:
"""Extract the optional PROF=1 execution-layer breakdown."""
row = P0_RE.search(stdout)
if not row:
raise RuntimeError("benchmark output missing P0-EXEC profile")
return [float(value) for value in row.groups()]
def execute(engine: str, env: dict[str, str]) -> tuple[float, list[float], list[float]]:
run = subprocess.run(
[engine, "4", "4", "4"], env=env, text=True, capture_output=True, check=True
)
return parse_output(run.stdout, run.stderr)
speed, profile = parse_output(run.stdout, run.stderr)
return speed, profile, parse_p0(run.stdout)
def main() -> None:
@@ -63,6 +78,8 @@ def main() -> None:
OMP_NUM_THREADS=str(args.threads),
OMP_PROC_BIND="spread",
OMP_PLACES="cores",
DRAFT="0",
PROF="1",
)
execute(args.engine, base | {"STATS": str(stats)})
@@ -86,13 +103,15 @@ def main() -> None:
execute(args.engine, base | extra) # warm-up
speeds = {name: [] for name in modes}
profiles = {name: [] for name in modes}
p0_profiles = {name: [] for name in modes}
names = list(modes)
for run_index in range(args.runs):
order = names[run_index % len(names):] + names[:run_index % len(names)]
for name in order:
speed, profile = execute(args.engine, base | modes[name])
speed, profile, p0 = execute(args.engine, base | modes[name])
speeds[name].append(speed)
profiles[name].append(profile)
p0_profiles[name].append(p0)
result = {}
for name in names:
@@ -103,6 +122,10 @@ def main() -> None:
key: statistics.median(row[index] for row in profiles[name])
for index, key in enumerate(PROFILE_KEYS)
},
"median_p0": {
key: statistics.median(row[index] for row in p0_profiles[name])
for index, key in enumerate(P0_KEYS)
},
}
print(json.dumps(result, indent=2))