Merge pull request #232 from nbeerbower/profiling-upstream
Profiling: PROF=1 opt-in performance profile + live per-turn Profiling page in the web dashboard
This commit is contained in:
Vincenzo
@@ -545,9 +545,10 @@ What you get:
|
||||
|
||||
- **Chat** with live metrics: a flashing token counter while generating, then tok/s, time-to-first-token, prompt→completion counts and queue wait;
|
||||
- **Runtime panel**: your hardware (CPU, GPUs + VRAM, RAM, cores), the scheduler, and the live expert-tier bar — how many of the 19,456 experts sit in VRAM / RAM / disk right now;
|
||||
- **Brain**: the whole model as a 76×256 cortex, one cell per expert. Colour = tier, brightness = routing heat, and the experts routed in each turn flash white and decay — you watch the model think. Hover any cell for its tier, heat and [measured topic affinity](https://github.com/JustVugg/colibri/issues/175) (specialists for code, Chinese, math, law… live in layers 11–22).
|
||||
- **Brain**: the whole model as a 76×256 cortex, one cell per expert. Colour = tier, brightness = routing heat, and the experts routed in each turn flash white and decay — you watch the model think. Hover any cell for its tier, heat and [measured topic affinity](https://github.com/JustVugg/colibri/issues/175) (specialists for code, Chinese, math, law… live in layers 11–22);
|
||||
- **Profiling**: where each turn's wall time went — I/O wait vs expert matmul vs attention vs LM head — as stacked per-turn bars, plus throughput history, tokens-per-forward batching, and a table of the recent turns. The same phase timers behind the `PROFILE` line, streamed live.
|
||||
|
||||
The dashboard talks to the engine over two tiny protocol lines (`TIERS`, `EMAP`/`HITS`) and plain JSON endpoints — nothing heavier than the engine itself.
|
||||
The dashboard talks to the engine over a few tiny protocol lines (`TIERS`, `EMAP`/`HITS`, `PROF`) and plain JSON endpoints — nothing heavier than the engine itself.
|
||||
|
||||
## Got a better machine? Try it — here's what to expect
|
||||
|
||||
|
||||
@@ -198,7 +198,9 @@ 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_edisk, t_ewait, t_emm, t_attn, t_kvb, t_head;/* profiling: dove va il tempo */
|
||||
double t_ewait, t_emm, t_attn, t_kvb, t_head;/* 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 */
|
||||
int64_t resident_bytes;
|
||||
/* DISK_SPLIT=1: split dei DISK LOAD (miss LRU -> expert_load) per contesto e per tipo
|
||||
@@ -215,6 +217,7 @@ static void ehit_mark(Model *m, int layer, int eid);
|
||||
static void emap_emit(Model *m);
|
||||
static void hits_emit(Model *m);
|
||||
static void hwinfo_emit(Model *m);
|
||||
static int64_t expert_bytes_probe(Model *m, int ebits); /* PROF: tier sizes in the report */
|
||||
static int g_repin;
|
||||
static uint64_t g_last_repin;
|
||||
#ifdef COLI_CUDA
|
||||
@@ -280,6 +283,38 @@ static double rss_gb(void){ struct rusage r; getrusage(RUSAGE_SELF,&r);
|
||||
return r.ru_maxrss/(1024.0*1024.0); /* Linux: in KB */
|
||||
#endif
|
||||
}
|
||||
/* ---- PROF=1: opt-in performance profile ----------------------------------
|
||||
* Records per-forward decode latency and expert-file bytes fetched, then
|
||||
* reports percentiles, I/O totals, phase shares and a tuning verdict next to
|
||||
* the existing PROFILE line. Additive only: with PROF unset the output of
|
||||
* every mode stays byte-identical. */
|
||||
static int g_prof=0;
|
||||
static _Atomic int64_t g_prof_io; /* bytes pread()/faulted from expert files */
|
||||
/* Disk service: wall time inside expert_load on whichever thread runs the read
|
||||
* (PIPE I/O workers, OMP loaders, the speculative pilot). It overlaps compute,
|
||||
* so it is NOT a wall-time phase — the stall the compute thread actually felt
|
||||
* is m->t_ewait. Thread-seconds, so it can exceed wall time under parallel
|
||||
* reads; wait << service means overlap/parallelism is hiding the reads,
|
||||
* wait ~ service means the loads block the compute thread. */
|
||||
static _Atomic int64_t g_edisk_ns;
|
||||
static double edisk_s(void){ return atomic_load_explicit(&g_edisk_ns,memory_order_relaxed)*1e-9; }
|
||||
#define PROF_LAT_CAP 32768
|
||||
static double g_prof_lat[PROF_LAT_CAP]; /* per-forward decode wall clock (ring) */
|
||||
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;
|
||||
} ProfBase;
|
||||
static void prof_base(Model *m, ProfBase *b){
|
||||
b->edisk=edisk_s(); b->ewait=m->t_ewait; b->emm=m->t_emm;
|
||||
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;
|
||||
}
|
||||
|
||||
static float *falloc(int64_t n){
|
||||
/* guardia anti-wrap (report PR #25): n assurdo da file modello ostili non deve
|
||||
* diventare una malloc piccola. Niente calloc: il memset nel percorso caldo costa. */
|
||||
@@ -1705,7 +1740,7 @@ static int pread_full(int fd, void *buf, int64_t n, int64_t off, const char *tag
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
static int expert_load_impl(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
#ifdef COLI_CUDA
|
||||
/* A live REPIN may reuse a GPU-enabled pinned slot for a different expert.
|
||||
* Keep its tier assignment, but invalidate the old device weights. */
|
||||
@@ -1721,6 +1756,8 @@ static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
qt_from_disk(m,nm[0],I,D,b,g_drop,&s->g);
|
||||
qt_from_disk(m,nm[1],I,D,b,g_drop,&s->u);
|
||||
qt_from_disk(m,nm[2],D,I,b,g_drop,&s->d);
|
||||
atomic_fetch_add_explicit(&g_prof_io,
|
||||
st_nbytes(&m->S,nm[0])+st_nbytes(&m->S,nm[1])+st_nbytes(&m->S,nm[2]),memory_order_relaxed);
|
||||
s->eid=eid; return 0;
|
||||
}
|
||||
st_tensor *tw[3], *tq[3];
|
||||
@@ -1776,6 +1813,7 @@ static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
* leak locked pages for every GPU-tier expert). See pin_wire() below: it wires
|
||||
* the final resident set only, after GPU release has already nulled out the
|
||||
* pointers for anything that isn't genuinely RAM-tier. */
|
||||
atomic_fetch_add_explicit(&g_prof_io,(int64_t)(n+nq),memory_order_relaxed);
|
||||
}
|
||||
s->eid=eid; return 0;
|
||||
}
|
||||
@@ -1865,6 +1903,7 @@ static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
for(int k=0;k<3;k++){
|
||||
if(pread_full(tq[k]->fd, (char*)(s->fslab+fo), tq[k]->nbytes, tq[k]->off, "pread qs")){ if(fatal) exit(1); return -1; }
|
||||
fp[k]=s->fslab+fo; fo+=tq[k]->nbytes/4; }
|
||||
atomic_fetch_add_explicit(&g_prof_io,wtot+fo*4,memory_order_relaxed);
|
||||
if(g_drop){ /* scarta subito le pagine: evita che la page
|
||||
* cache in pressione strangoli il throughput */
|
||||
posix_fadvise(tw[ord[0]]->fd, tw[ord[0]]->off, wtot, POSIX_FADV_DONTNEED);
|
||||
@@ -1882,6 +1921,14 @@ static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
}
|
||||
s->eid=eid; return 0;
|
||||
}
|
||||
/* Every expert read goes through here: time the whole load (pread/fault +
|
||||
* bookkeeping) on the thread that runs it, into the disk-service counter. */
|
||||
static int expert_load(Model *m, int layer, int eid, ESlot *s, int fatal){
|
||||
double t0=now_s();
|
||||
int rc=expert_load_impl(m,layer,eid,s,fatal);
|
||||
atomic_fetch_add_explicit(&g_edisk_ns,(int64_t)((now_s()-t0)*1e9),memory_order_relaxed);
|
||||
return rc;
|
||||
}
|
||||
|
||||
#ifdef __linux__
|
||||
/* io_uring expert batches. One owner prepares all reads for a block, submits
|
||||
@@ -3080,11 +3127,12 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
|
||||
double t0=now_s();
|
||||
int eids[64]; for(int q=0;q<nmiss;q++) eids[q]=uniq[base+missk[q]];
|
||||
pipe_dispatch(m,layer,eids,nmiss);
|
||||
m->t_edisk += now_s()-t0; /* dispatch only; real reads hide behind matmul */
|
||||
m->t_ewait += now_s()-t0; /* dispatch only; the reads overlap matmul and
|
||||
* are timed as service inside expert_load */
|
||||
} else { double t0=now_s(); /* ORIGINALE: blocking parallel load */
|
||||
#pragma omp parallel for schedule(dynamic,1)
|
||||
for(int q=0;q<nmiss;q++) expert_load(m,layer,uniq[base+missk[q]],&m->ws[q],1);
|
||||
m->t_ewait += now_s()-t0; } /* blocking: whole load stalls compute */
|
||||
m->t_ewait += now_s()-t0; } /* compute thread blocked for the whole load */
|
||||
}
|
||||
/* I/O ASINCRONO: readahead (WILLNEED) del blocco SUCCESSIVO mentre calcoliamo
|
||||
* questo — il kernel legge in background, le pread dopo trovano cache calda */
|
||||
@@ -3113,7 +3161,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
|
||||
* correct (and free) when a subset falls back to the CPU. */
|
||||
if(g_pipe && nmiss){ double tw=now_s();
|
||||
for(int q=0;q<nmiss;q++) pipe_wait(q);
|
||||
m->t_ewait += now_s()-tw; } /* blocking: drain stalled compute */
|
||||
m->t_ewait += now_s()-tw; }
|
||||
MB_BUILD(1, 0); /* missed experts, now loaded */
|
||||
if(nbb>0){
|
||||
double t0=now_s();
|
||||
@@ -3137,7 +3185,7 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int
|
||||
* Stays ABOVE the METAL skip: a subset that fell back to the CPU still needs its
|
||||
* slot drained here, and under METAL the block-level drain above already ran (this
|
||||
* spin is then a no-op). */
|
||||
if(g_pipe && qof[j]>=0){ double tw=now_s(); pipe_wait(qof[j]); m->t_ewait += now_s()-tw; } /* blocking */
|
||||
if(g_pipe && qof[j]>=0){ double tw=now_s(); pipe_wait(qof[j]); m->t_ewait += now_s()-tw; }
|
||||
#ifdef COLI_METAL
|
||||
/* skip the subsets already computed on GPU */
|
||||
if(g_metal_enabled && ((is_miss[j] && !cpu_miss) || (!is_miss[j] && !cpu_res))) continue;
|
||||
@@ -4283,7 +4331,9 @@ static int spec_decode(Model *m, int *all, int kv, int n_new, int eos, float *lo
|
||||
if(g<0) g=0;
|
||||
if(gsrc==1) g_gr_prop+=(uint64_t)g;
|
||||
int S=1+g; int batch[64]; batch[0]=next; memcpy(batch+1,draft,g*sizeof(int));
|
||||
double tf0=g_prof?now_s():0;
|
||||
float *lo=step_all(m,batch,S,kv); m->n_fw++;
|
||||
if(g_prof) prof_lat(now_s()-tf0);
|
||||
int k=0; /* verifica: accetta finche' coincide */
|
||||
if(g>0 && getenv("MTP_DEBUG")){ int veri=argmax_v(lo,V);
|
||||
fprintf(stderr,"[mtpdbg] draft0=%d verified=%d %s\n", draft[0], veri, draft[0]==veri?"HIT":"miss"); }
|
||||
@@ -4432,10 +4482,10 @@ static void generate(Model *m, const int *prompt, int np, int n_new, int *out){
|
||||
}
|
||||
|
||||
static void profile_print(Model *m, double elapsed){
|
||||
double accounted=m->t_edisk+m->t_ewait+m->t_emm+m->t_attn+m->t_head;
|
||||
double accounted=m->t_ewait+m->t_emm+m->t_attn+m->t_head;
|
||||
printf("PROFILE: expert-disk %.3fs service / %.3fs wait | expert-matmul %.3fs | attention %.3fs "
|
||||
"(including kvb %.3fs) | lm_head %.3fs | other %.3fs\n",
|
||||
m->t_edisk,m->t_ewait,m->t_emm,m->t_attn,m->t_kvb,m->t_head,elapsed-accounted);
|
||||
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);
|
||||
#ifdef COLI_METAL
|
||||
@@ -4450,8 +4500,73 @@ static void profile_print(Model *m, double elapsed){
|
||||
}
|
||||
|
||||
static void profile_reset(Model *m){
|
||||
m->t_edisk=m->t_ewait=m->t_emm=m->t_attn=m->t_kvb=m->t_head=0;
|
||||
m->t_ewait=m->t_emm=m->t_attn=m->t_kvb=m->t_head=0;
|
||||
m->t_aproj=m->t_acore=m->t_aout=0;
|
||||
atomic_store_explicit(&g_edisk_ns,0,memory_order_relaxed);
|
||||
}
|
||||
|
||||
/* PROF=1 report: forward-latency percentiles, expert I/O totals, phase shares
|
||||
* of wall time, and a plain-language verdict naming the knob most likely to
|
||||
* move tok/s on THIS machine with THIS config. `b` marks the window start
|
||||
* (serve mode reports per turn; batch modes snapshot right after reset). */
|
||||
static int prof_cmp_d(const void *a,const void *b){
|
||||
double x=*(const double*)a, y=*(const double*)b; return (x>y)-(x<y); }
|
||||
static void prof_report(Model *m, const ProfBase *b, double elapsed, int tokens, FILE *f){
|
||||
Cfg *c=&m->c; if(elapsed<1e-9) elapsed=1e-9;
|
||||
uint64_t nw=g_prof_nlat-b->nlat; if(nw>PROF_LAT_CAP) nw=PROF_LAT_CAP; /* ring keeps the tail */
|
||||
uint64_t nfw=m->n_fw-b->n_fw, nem=m->n_emit-b->n_emit;
|
||||
if(nw){
|
||||
double *v=malloc((size_t)nw*sizeof(double));
|
||||
if(v){
|
||||
for(uint64_t i=0;i<nw;i++) v[i]=g_prof_lat[(g_prof_nlat-nw+i)%PROF_LAT_CAP];
|
||||
qsort(v,(size_t)nw,sizeof(double),prof_cmp_d);
|
||||
double p50=v[(nw-1)/2], p90=v[(uint64_t)((nw-1)*0.90)], p99=v[(uint64_t)((nw-1)*0.99)], mx=v[nw-1];
|
||||
fprintf(f,"[PROF] decode forwards: %llu | latency p50 %.1f ms | p90 %.1f ms | p99 %.1f ms | max %.1f ms | %.2f tok/forward\n",
|
||||
(unsigned long long)nfw,p50*1e3,p90*1e3,p99*1e3,mx*1e3,nfw?(double)nem/nfw:0.0);
|
||||
if(nw>=32 && p99>3*p50)
|
||||
fprintf(f,"[PROF] tail: p99 is %.1fx p50 — the slow forwards are cold-cache expert loads; "
|
||||
"a warm-up turn or a pinned hot-store (PIN / AUTOPIN history) shrinks them\n",p99/p50);
|
||||
free(v);
|
||||
}
|
||||
}
|
||||
int64_t io=atomic_load_explicit(&g_prof_io,memory_order_relaxed)-b->io;
|
||||
uint64_t dh=m->hits-b->hits, dm=m->miss-b->miss, dq=m->ereq-b->ereq;
|
||||
double hitp=(dh+dm)?100.0*dh/(dh+dm):100.0;
|
||||
double eb=(double)expert_bytes_probe(m,m->ebits);
|
||||
int pinned=0,lru=0;
|
||||
for(int i=0;i<=c->n_layers;i++){ if(m->npin)pinned+=m->npin[i]; if(m->ecn)lru+=m->ecn[i]; }
|
||||
double io_w=m->t_ewait-b->ewait; /* stall the compute thread felt */
|
||||
double io_svc=edisk_s()-b->edisk; /* read service on the loading threads (overlaps compute) */
|
||||
fprintf(f,"[PROF] expert I/O: %.3f GB fetched (%.1f MB/token, %.2f GB/s over the run%s) | "
|
||||
"hit %.1f%% (%llu hit / %llu load) | %.1f loads/token | %.1fs read service / %.1fs felt wait\n",
|
||||
io/1e9, tokens>0?io/1e6/tokens:0.0, io/1e9/elapsed,
|
||||
g_mmap?"; COLI_MMAP=1: page cache may serve part":"",
|
||||
hitp,(unsigned long long)dh,(unsigned long long)dm, tokens>0?(double)dq/tokens:0.0,
|
||||
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 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);
|
||||
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).");
|
||||
else fprintf(f," The cache is already warm — the routed working set streams from disk; a faster disk or a bigger pinned tier (PIN_GB) is the lever.");
|
||||
if(!g_pipe) fprintf(f," Try PIPE=1 (overlap reads with matmul).");
|
||||
if(!g_direct) fprintf(f," On NVMe try DIRECT=1.");
|
||||
fprintf(f,"\n");
|
||||
} else if(f_emm>=0.40){
|
||||
fprintf(f,"[PROF] verdict: compute-bound in expert matmuls (%.0f%%) — more cores/threads help; keep IDOT=1, or move hot experts to a GPU tier (COLI_CUDA / COLI_METAL).%s\n",
|
||||
100*f_emm, g_mmap?" Note: with COLI_MMAP=1 page-fault I/O is accounted inside matmul.":"");
|
||||
} else if(f_attn>=0.35){
|
||||
fprintf(f,"[PROF] verdict: attention-bound (%.0f%%) — context length is the cost (DSA %s). A lower CTX helps if the workload allows.\n",
|
||||
100*f_attn, m->has_dsa?"on":"not available for this model");
|
||||
} else {
|
||||
fprintf(f,"[PROF] verdict: balanced — no phase dominates (I/O %.0f%%, matmul %.0f%%, attention %.0f%%); this config is a reasonable fit for this machine.\n",
|
||||
100*f_io,100*f_emm,100*f_attn);
|
||||
}
|
||||
}
|
||||
|
||||
/* Fixed-token decode benchmark: prefill all but the prompt's last token, then
|
||||
@@ -4463,14 +4578,18 @@ static void run_replay(Model *m, const int *full, int nfull, int np){
|
||||
float *logit=step(m,full,np-1,0); free(logit);
|
||||
m->hits=m->miss=m->ereq=m->gpu_expert_calls=0;
|
||||
profile_reset(m);
|
||||
ProfBase pb; prof_base(m,&pb);
|
||||
double t0=now_s(); int steps=0;
|
||||
for(int i=np-1;i<nfull-1;i++){
|
||||
double tf0=g_prof?now_s():0;
|
||||
logit=step(m,full+i,1,i); free(logit); steps++;
|
||||
if(g_prof){ prof_lat(now_s()-tf0); m->n_fw++; m->n_emit++; }
|
||||
}
|
||||
double dt=now_s()-t0, tot=m->hits+m->miss;
|
||||
printf("REPLAY decode: %d tokens in %.3fs | %.2f tok/s | expert hit %.1f%%\n",
|
||||
steps,dt,steps/dt,tot?100.0*m->hits/tot:0.0);
|
||||
profile_print(m,dt);
|
||||
if(g_prof) prof_report(m,&pb,dt,steps,stdout);
|
||||
#ifdef COLI_CUDA
|
||||
if(m->gpu_expert_count) printf("CUDA expert tier: %d resident experts (%.2f GB) | %llu calls served from VRAM\n",
|
||||
m->gpu_expert_count,m->gpu_expert_bytes/1e9,(unsigned long long)m->gpu_expert_calls);
|
||||
@@ -4511,6 +4630,7 @@ static void run_text(Model *m, const char *snap, const char *prompt, int ngen){
|
||||
m->n_emit=m->n_fw=0;
|
||||
g_last_repin=0;
|
||||
profile_reset(m);
|
||||
ProfBase pb; prof_base(m,&pb);
|
||||
double t=now_s();
|
||||
EmitStream es={&T,m,t,0,0};
|
||||
grammar_reset();
|
||||
@@ -4554,6 +4674,7 @@ static void run_text(Model *m, const char *snap, const char *prompt, int ngen){
|
||||
if(g_cuda_enabled) cuda_stats_print();
|
||||
#endif
|
||||
profile_print(m,dt);
|
||||
if(g_prof) prof_report(m,&pb,dt,produced,stdout);
|
||||
if(g_pilot_real) printf("PILOT_REAL: %ld load cross-layer completati, %ld scartati (main gia' sul layer) | PILOT_K=%d\n",
|
||||
(long)atomic_load_explicit(&g_pilot_loads,memory_order_relaxed),
|
||||
(long)atomic_load_explicit(&g_pilot_drops,memory_order_relaxed), g_pilot_k);
|
||||
@@ -4860,6 +4981,8 @@ typedef struct {
|
||||
float temp, top_p;
|
||||
double started;
|
||||
uint64_t hits0, miss0;
|
||||
ProfBase pb; /* phase-time window start (same convention as hits0):
|
||||
feeds the PROF protocol line and the PROF=1 report */
|
||||
} ServeReq;
|
||||
|
||||
static void mux_data(Tok *T, unsigned long long id, int token){
|
||||
@@ -4876,10 +4999,26 @@ static void mux_done(Model *m, ServeCtx *sc, ServeReq *r){
|
||||
tiers_emit(m);
|
||||
emap_emit(m);
|
||||
hits_emit(m);
|
||||
/* PROF: per-turn phase timings for the dashboard profiling page —
|
||||
* "PROF <wall_s> <prompt> <completion> <edisk> <ewait> <emm> <attn> <head> <n_fw>".
|
||||
* edisk = disk service (expert_load wall on the reading threads, overlaps
|
||||
* compute); ewait = the stall the compute thread felt — only ewait belongs
|
||||
* in a wall-time breakdown. With KV_SLOTS>1 concurrent slots share the
|
||||
* batched forwards, so the shares describe the whole engine over the
|
||||
* window, not the single request (same convention as the STAT hit% below). */
|
||||
printf("PROF %.3f %d %d %.3f %.3f %.3f %.3f %.3f %llu\n",dt,
|
||||
r->prompt_tokens,r->emitted,
|
||||
edisk_s()-r->pb.edisk,m->t_ewait-r->pb.ewait,m->t_emm-r->pb.emm,
|
||||
m->t_attn-r->pb.attn,m->t_head-r->pb.head,
|
||||
(unsigned long long)(m->n_fw-r->pb.n_fw));
|
||||
printf("DONE %llu STAT %d %.2f %.1f %.2f %d %d\n",r->id,r->emitted,
|
||||
r->emitted/dt,(dh+dm)>0?100.0*dh/(dh+dm):0.0,rss_gb(),
|
||||
r->prompt_tokens,r->length_limited);
|
||||
fflush(stdout); kv_bind(m,&sc->kv); kv_disk_append(m,sc->hist,sc->len);
|
||||
/* PROF window = this request's lifetime; with KV_SLOTS>1 concurrent slots
|
||||
* share the batched forwards, so the shares describe the engine, not the
|
||||
* single request (same convention as the STAT hit%% above). */
|
||||
if(g_prof) prof_report(m,&r->pb,dt,r->emitted,stderr);
|
||||
r->active=0;
|
||||
}
|
||||
|
||||
@@ -4942,6 +5081,7 @@ static int mux_submit(Model *m, Tok *T, ServeCtx *ctx, ServeReq *req, int nctx,
|
||||
ServeReq *r=&req[sub.slot]; memset(r,0,sizeof(*r));
|
||||
r->id=sub.id; r->maximum=sub.max_tokens; r->temp=sub.temperature; r->top_p=sub.top_p;
|
||||
r->prompt_tokens=nt; r->started=now_s(); r->hits0=m->hits; r->miss0=m->miss;
|
||||
prof_base(m,&r->pb); /* a few loads: cheap enough to always track */
|
||||
int room=maxctx-sc->len-1; if(r->maximum>room){r->maximum=room; r->length_limited=1;}
|
||||
g_temp=r->temp; g_nuc=r->top_p;
|
||||
int next=pick_tok(logit,m->c.vocab,-1); free(logit);
|
||||
@@ -5018,8 +5158,10 @@ static void run_serve_mux(Model *m, const char *snap){
|
||||
for(int i=0;i<nctx;i++) if(req[i].active){
|
||||
rows[S]=(DecodeRow){&ctx[i].kv,req[i].pending,ctx[i].len}; slots[S++]=i;
|
||||
}
|
||||
double tf0=g_prof?now_s():0;
|
||||
float *lo=step_decode_batch(m,rows,S); if(!lo){fprintf(stderr,"decode batch failed\n");break;}
|
||||
m->n_fw++;
|
||||
if(g_prof) prof_lat(now_s()-tf0);
|
||||
for(int s=0;s<S;s++){
|
||||
int i=slots[s]; ServeCtx *sc=&ctx[i]; ServeReq *r=&req[i];
|
||||
sc->len++; g_temp=r->temp; g_nuc=r->top_p;
|
||||
@@ -5089,6 +5231,7 @@ static void run_serve(Model *m, const char *snap){
|
||||
if(len<1){ printf("\x01\x01" "END" "\x01\x01\n"); printf("STAT 0 0.00 0.0 %.2f\n", rss_gb()); fflush(stdout); continue; }
|
||||
int cur=ngen; if(len+cur+g_draft+2>=maxctx) cur=maxctx-len-g_draft-2;
|
||||
uint64_t h0=m->hits, ms0=m->miss; double tt0=now_s();
|
||||
ProfBase pb; if(g_prof) prof_base(m,&pb);
|
||||
float *logit=step(m,hist+len-1,1,len-1);
|
||||
EmitStream es={&T,m,now_s(),0,1};
|
||||
int prod=0;
|
||||
@@ -5098,7 +5241,9 @@ static void run_serve(Model *m, const char *snap){
|
||||
double dh=(double)(m->hits-h0), dm=(double)(m->miss-ms0);
|
||||
printf("\n\x01\x01" "END" "\x01\x01\n");
|
||||
printf("STAT %d %.2f %.1f %.2f\n", prod, prod/tdt, (dh+dm)>0?100.0*dh/(dh+dm):0.0, rss_gb());
|
||||
fflush(stdout); kv_disk_append(m,hist,len); repin_pass(m); continue; } /* RFC: re-pin a caldo tra i turni / live re-pin between turns */
|
||||
fflush(stdout);
|
||||
if(g_prof) prof_report(m,&pb,tdt,prod,stderr); /* per-turn window; stdout is the framed protocol */
|
||||
kv_disk_append(m,hist,len); repin_pass(m); continue; } /* RFC: re-pin a caldo tra i turni / live re-pin between turns */
|
||||
if(nr<1){ printf("\x01\x01" "END" "\x01\x01\n"); printf("STAT 0 0.00 0.0 %.2f\n", rss_gb()); fflush(stdout); continue; }
|
||||
/* API mode: an exact, length-prefixed prompt. Unlike the interactive
|
||||
* line protocol this accepts newlines. The tokenized prompt is matched
|
||||
@@ -5166,6 +5311,7 @@ static void run_serve(Model *m, const char *snap){
|
||||
uint64_t agh0=m->route_agree_hit, agt0=m->route_agree_tot;
|
||||
uint64_t kln0=m->route_kl_n; double kls0=m->route_kl_sum;
|
||||
double tt0=now_s();
|
||||
ProfBase pb; if(g_prof) prof_base(m,&pb);
|
||||
float *logit;
|
||||
if(k>0){ logit=step(m,hist+len,k,len); len+=k; }
|
||||
else logit=step(m,hist+len-1,1,len-1); /* prompt identico/prefisso: rigenera i logits */
|
||||
@@ -5192,6 +5338,7 @@ static void run_serve(Model *m, const char *snap){
|
||||
agree_pct,kl_mean);
|
||||
printf("\n");
|
||||
fflush(stdout);
|
||||
if(g_prof) prof_report(m,&pb,tdt,prod,stderr); /* per-turn window; stdout is the framed protocol */
|
||||
free(raw); g_temp=base_temp; g_nuc=base_nuc;
|
||||
usage_save(m); /* la cache che impara: storia aggiornata a ogni turno */
|
||||
kv_disk_append(m,hist,len); /* KV su disco: il prossimo avvio riparte da qui */
|
||||
@@ -5252,10 +5399,10 @@ static void ehit_mark(Model *m, int layer, int eid){
|
||||
}
|
||||
|
||||
/* HWINFO: hardware snapshot for the web dashboard — emitted once at READY. */
|
||||
static void hwinfo_emit(Model *m){
|
||||
Cfg *c=&m->c; (void)c; /* silence -Wunused on builds without /proc (#148 report) */
|
||||
/* CPU */
|
||||
char cpu[256]="";
|
||||
/* CPU model + cores + RAM (GB); empty/zero where unavailable.
|
||||
* Shared by the dashboard HWINFO line and the PROF=1 header. */
|
||||
static void hw_probe(char *cpu, size_t cn, int *cores, double *ram_total, double *ram_avail){
|
||||
cpu[0]=0;
|
||||
#ifdef _WIN32
|
||||
/* niente /proc su Windows: brand string via CPUID (0x80000002..4), zero
|
||||
* dipendenze extra. La dashboard mostrava "0 GB RAM / 0 cores" perche'
|
||||
@@ -5265,7 +5412,7 @@ static void hwinfo_emit(Model *m){
|
||||
for(unsigned int f=0x80000002u; f<=0x80000004u; f++,w+=4)
|
||||
__get_cpuid(f,&w[0],&w[1],&w[2],&w[3]);
|
||||
char *b=(char*)r; b[47]=0; while(*b==' ')b++;
|
||||
snprintf(cpu,sizeof(cpu),"%s",b); }
|
||||
snprintf(cpu,cn,"%s",b); }
|
||||
#endif
|
||||
#else
|
||||
FILE *ci=fopen("/proc/cpuinfo","r");
|
||||
@@ -5273,27 +5420,32 @@ static void hwinfo_emit(Model *m){
|
||||
while(fgets(ln,sizeof(ln),ci)) if(!strncmp(ln,"model name",10)){
|
||||
char *p=strchr(ln,':'); if(p){ p++; while(*p==' ')p++;
|
||||
int n=(int)strlen(p); if(n>0&&p[n-1]=='\n')p[--n]=0;
|
||||
snprintf(cpu,sizeof(cpu),"%s",p); } break; }
|
||||
snprintf(cpu,cn,"%s",p); } break; }
|
||||
fclose(ci); }
|
||||
#endif
|
||||
int cores=0;
|
||||
*cores=0;
|
||||
#ifdef _WIN32
|
||||
{ SYSTEM_INFO si; GetSystemInfo(&si); cores=(int)si.dwNumberOfProcessors; }
|
||||
{ SYSTEM_INFO si; GetSystemInfo(&si); *cores=(int)si.dwNumberOfProcessors; }
|
||||
#elif defined(_SC_NPROCESSORS_ONLN)
|
||||
cores=(int)sysconf(_SC_NPROCESSORS_ONLN);
|
||||
*cores=(int)sysconf(_SC_NPROCESSORS_ONLN);
|
||||
#endif
|
||||
/* RAM */
|
||||
double ram_total=0,ram_avail=0;
|
||||
*ram_total=*ram_avail=0;
|
||||
#ifdef _WIN32
|
||||
compat_meminfo(&ram_total,&ram_avail); /* GlobalMemoryStatusEx, gia' in compat.h */
|
||||
compat_meminfo(ram_total,ram_avail); /* GlobalMemoryStatusEx, gia' in compat.h */
|
||||
#else
|
||||
FILE *mi=fopen("/proc/meminfo","r");
|
||||
if(mi){ char ln[256]; double mt=0,ma=0;
|
||||
while(fgets(ln,sizeof(ln),mi)){
|
||||
if(sscanf(ln,"MemTotal: %lf",&mt)==1) ram_total=mt/1e6;
|
||||
if(sscanf(ln,"MemAvailable: %lf",&ma)==1) ram_avail=ma/1e6;
|
||||
if(sscanf(ln,"MemTotal: %lf",&mt)==1) *ram_total=mt/1e6;
|
||||
if(sscanf(ln,"MemAvailable: %lf",&ma)==1) *ram_avail=ma/1e6;
|
||||
} fclose(mi); }
|
||||
#endif
|
||||
}
|
||||
|
||||
static void hwinfo_emit(Model *m){
|
||||
Cfg *c=&m->c; (void)c; /* silence -Wunused on builds without /proc (#148 report) */
|
||||
char cpu[256]; int cores; double ram_total,ram_avail;
|
||||
hw_probe(cpu,sizeof(cpu),&cores,&ram_total,&ram_avail);
|
||||
/* GPU */
|
||||
int ngpu=0; double vram_total=0;
|
||||
char gpu_name[128]="";
|
||||
@@ -5796,6 +5948,34 @@ static const char *coli_user_prompt(void){
|
||||
return p;
|
||||
}
|
||||
|
||||
/* PROF=1 startup header: one self-describing block so a saved log answers
|
||||
* "what machine, what config" when comparing runs after changing RAM_GB,
|
||||
* knobs, or moving to another host. */
|
||||
static void prof_config(Model *m, double ram_env, int est_ctx){
|
||||
Cfg *c=&m->c;
|
||||
char cpu[256]; int cores; double rt,ra;
|
||||
hw_probe(cpu,sizeof(cpu),&cores,&rt,&ra);
|
||||
const char *backend="CPU";
|
||||
#ifdef COLI_CUDA
|
||||
if(g_cuda_enabled) backend="CUDA";
|
||||
#endif
|
||||
#ifdef COLI_METAL
|
||||
if(g_metal_enabled) backend="Metal";
|
||||
#endif
|
||||
int nsp=0; for(int i=0;i<c->n_layers;i++) if(m->L[i].sparse) nsp++;
|
||||
int rows=nsp+(m->has_mtp?2:0); /* stessa convenzione di cap_for_ram (MTP int8 = 2x) */
|
||||
int pinned=0; for(int i=0;i<=c->n_layers;i++) if(m->npin) pinned+=m->npin[i];
|
||||
double eb=(double)expert_bytes_probe(m,m->ebits);
|
||||
fprintf(stderr,"[PROF] machine: %s | %d cores (%d omp threads) | RAM %.1f GB total, %.1f GB available | backend %s\n",
|
||||
cpu[0]?cpu:"unknown CPU",cores,omp_get_max_threads(),rt,ra,backend);
|
||||
fprintf(stderr,"[PROF] config: RAM_GB=%s%.1f CTX=%d | expert cache cap %d/layer (up to %.1f GB) | pinned %d (%.1f GB) | "
|
||||
"DRAFT=%d PIPE=%d DIRECT=%d MMAP=%d IDOT=%d DSA=%s PILOT=%d CACHE_ROUTE=%d\n",
|
||||
ram_env<=0?"auto ":"",ram_env<=0?g_mem_avail_boot*0.88:ram_env,est_ctx,
|
||||
m->ecap,(double)m->ecap*rows*eb/1e9,pinned,pinned*eb/1e9,
|
||||
g_draft,g_pipe,g_direct,g_mmap,g_idot,
|
||||
(m->has_dsa&&c->index_topk)?"on":"off",g_pilot,g_cache_route);
|
||||
}
|
||||
|
||||
int main(int argc, char **argv){
|
||||
/* ---- Permanent OpenMP hot-thread tuning. The per-expert matmul regions are
|
||||
* tiny and back-to-back; with the default passive wait policy libgomp parks
|
||||
@@ -6104,7 +6284,9 @@ int main(int argc, char **argv){
|
||||
}
|
||||
/* SEMPRE: senza clamp la LRU cresce fino a cap*76 layer = decine di GB -> OOM-kill.
|
||||
* RAM_GB assente o <=0 = budget automatico da MemAvailable. */
|
||||
cap_for_ram(&m, ram_env, ebits, est_ctx); }
|
||||
cap_for_ram(&m, ram_env, ebits, est_ctx);
|
||||
g_prof = getenv("PROF")?atoi(getenv("PROF")):0; /* PROF=1: opt-in performance profile */
|
||||
if(g_prof) prof_config(&m, ram_env, est_ctx); }
|
||||
const char *stats=getenv("STATS"); /* STATS=<file> -> istogramma uso expert a fine run */
|
||||
|
||||
/* modo scoring per benchmark: SCORE=<requests.txt> -> log-likelihood per riga */
|
||||
@@ -6181,6 +6363,7 @@ int main(int argc, char **argv){
|
||||
return 0;
|
||||
}
|
||||
int *out=malloc((np+n_new)*sizeof(int));
|
||||
ProfBase pb; prof_base(&m,&pb);
|
||||
double t=now_s(); generate(&m,prompt,np,n_new,out); double dt=now_s()-t;
|
||||
int match=0;
|
||||
printf("\nReference (oracle): "); for(int i=np;i<nfull;i++) printf("%d ", full[i]);
|
||||
@@ -6192,6 +6375,7 @@ int main(int argc, char **argv){
|
||||
printf("Expert cache hit rate: %.1f%% (hit=%llu miss=%llu) | RSS: %.2f GB | %.1f tok/s\n",
|
||||
tot?100.0*m.hits/tot:0.0, (unsigned long long)m.hits, (unsigned long long)m.miss, rss_gb(), n_new/dt);
|
||||
profile_print(&m,dt);
|
||||
if(g_prof) prof_report(&m,&pb,dt,n_new,stdout);
|
||||
#ifdef COLI_CUDA
|
||||
if(m.gpu_expert_count) printf("CUDA expert tier: %d resident experts (%.2f GB) | %llu calls served from VRAM\n",
|
||||
m.gpu_expert_count,m.gpu_expert_bytes/1e9,(unsigned long long)m.gpu_expert_calls);
|
||||
|
||||
@@ -27,6 +27,7 @@ HERE = Path(__file__).resolve().parent
|
||||
END = b"\x01\x01END\x01\x01\n"
|
||||
READY = b"\x01\x01READY\x01\x01\n"
|
||||
MAX_BODY = 4 << 20
|
||||
PROFILE_TURNS = 120 # rolling window of per-turn PROF snapshots kept for /profile
|
||||
DEFAULT_CORS_ORIGINS = (
|
||||
"http://127.0.0.1:8000",
|
||||
"http://localhost:8000",
|
||||
@@ -473,6 +474,8 @@ class Engine:
|
||||
self.emap = None
|
||||
self.hits = None
|
||||
self.hits_seq = 0 # latest "TIERS" snapshot from the engine
|
||||
self.profile = collections.deque(maxlen=PROFILE_TURNS) # per-turn phase timings
|
||||
self.profile_seq = 0
|
||||
read_engine_turn(self.process.stdout, READY, lambda _: None)
|
||||
self.dispatcher = threading.Thread(target=self._dispatch_stdout,
|
||||
name="colibri-stdout", daemon=True)
|
||||
@@ -550,6 +553,20 @@ class Engine:
|
||||
elif kind == "HITS" and len(fields) == 4:
|
||||
self.hits = fields[3]
|
||||
self.hits_seq += 1
|
||||
elif kind == "PROF" and len(fields) >= 10:
|
||||
# per-turn phase timings: where the engine spent this turn's wall time
|
||||
self.profile.append({
|
||||
"wall_s": float(fields[1]),
|
||||
"prompt_tokens": int(fields[2]),
|
||||
"completion_tokens": int(fields[3]),
|
||||
"expert_disk_s": float(fields[4]),
|
||||
"expert_wait_s": float(fields[5]),
|
||||
"expert_matmul_s": float(fields[6]),
|
||||
"attention_s": float(fields[7]),
|
||||
"lm_head_s": float(fields[8]),
|
||||
"forwards": int(fields[9]),
|
||||
})
|
||||
self.profile_seq += 1
|
||||
elif kind == "TIERS" and len(fields) >= 6:
|
||||
self.tiers = {"vram": int(fields[1]), "ram": int(fields[2]),
|
||||
"disk": int(fields[3]), "vram_gb": float(fields[4]),
|
||||
@@ -782,6 +799,12 @@ class APIHandler(BaseHTTPRequestHandler):
|
||||
payload["seq"] = eng.hits_seq
|
||||
self.send_json(200, payload, request_id)
|
||||
return
|
||||
if path == "/profile":
|
||||
eng = self.server.engine
|
||||
payload = {"seq": getattr(eng, "profile_seq", 0) if eng else 0,
|
||||
"turns": list(getattr(eng, "profile", ()) or ()) if eng else []}
|
||||
self.send_json(200, payload, request_id)
|
||||
return
|
||||
if self.serve_static(path):
|
||||
return
|
||||
self.require_auth()
|
||||
|
||||
@@ -380,6 +380,25 @@ class DispatcherTest(unittest.TestCase):
|
||||
engine.close()
|
||||
self.assertEqual(chunks, ["é"])
|
||||
|
||||
def test_records_profile_snapshots_from_prof_lines(self):
|
||||
def respond(process, frame):
|
||||
request_id = frame.split()[1]
|
||||
process.stdout.feed(b"DATA " + request_id + b" 2\nok\n")
|
||||
process.stdout.feed(b"PROF 2.500 7 12 0.400 0.100 0.900 0.600 0.200 15\n")
|
||||
process.stdout.feed(b"DONE " + request_id + b" STAT 12 4.8 0 1.0 7 0\n")
|
||||
|
||||
process = FakeProcess(respond)
|
||||
with patch("openai_server.subprocess.Popen", return_value=process):
|
||||
engine = Engine("glm", "model")
|
||||
engine.generate("hello", 16, 0.7, 0.9, lambda _: None)
|
||||
engine.close()
|
||||
self.assertEqual(engine.profile_seq, 1)
|
||||
self.assertEqual(list(engine.profile), [{
|
||||
"wall_s": 2.5, "prompt_tokens": 7, "completion_tokens": 12,
|
||||
"expert_disk_s": 0.4, "expert_wait_s": 0.1, "expert_matmul_s": 0.9,
|
||||
"attention_s": 0.6, "lm_head_s": 0.2, "forwards": 15,
|
||||
}])
|
||||
|
||||
def test_cancels_generation_after_consumer_disconnects(self):
|
||||
request_id = None
|
||||
|
||||
@@ -443,6 +462,20 @@ class HTTPTest(unittest.TestCase):
|
||||
self.assertIn("queued", scheduler)
|
||||
self.assertEqual(health["kv_slots"], 2)
|
||||
|
||||
def test_profile_reports_recent_turns_without_auth(self):
|
||||
with urlopen(self.base + "/profile", timeout=2) as response:
|
||||
self.assertEqual(json.load(response), {"seq": 0, "turns": []})
|
||||
turn = {"wall_s": 2.5, "prompt_tokens": 7, "completion_tokens": 12,
|
||||
"expert_disk_s": 0.4, "expert_wait_s": 0.1, "expert_matmul_s": 0.9,
|
||||
"attention_s": 0.6, "lm_head_s": 0.2, "forwards": 15}
|
||||
self.engine.profile = [turn]
|
||||
self.engine.profile_seq = 1
|
||||
try:
|
||||
with urlopen(self.base + "/profile", timeout=2) as response:
|
||||
self.assertEqual(json.load(response), {"seq": 1, "turns": [turn]})
|
||||
finally:
|
||||
del self.engine.profile, self.engine.profile_seq
|
||||
|
||||
def test_browser_preflight(self):
|
||||
request = Request(self.base + "/v1/chat/completions", method="OPTIONS", headers={
|
||||
"Origin": "http://localhost:5173",
|
||||
|
||||
@@ -63,6 +63,7 @@ Format: `VAR` — default — effect.
|
||||
| `ABSORB` | `-1` (auto: absorbed for S≤4) | MLA attention absorption mode. |
|
||||
| `IDOT` | `1` | Integer dot-product kernel. `IDOT=0` uses exact f32 kernels (for A/B numerical checks). |
|
||||
| `COLI_POLICY` | `quality` | Resource policy: `quality`, `balanced`, or `experimental-fast`. |
|
||||
| `PROF` | `0` (off) | Performance profile: a startup header (machine + effective config), then per run — or per turn in serve mode, on stderr — forward-latency percentiles (p50/p90/p99/max), expert-I/O totals and cache-tier fill, phase shares of wall time, and a verdict naming the knob most likely to help on this machine. Output is additive; `PROF` unset changes nothing. |
|
||||
|
||||
---
|
||||
|
||||
|
||||
+6
-1
@@ -17,9 +17,14 @@ npm test
|
||||
npm run build
|
||||
```
|
||||
|
||||
Besides Chat and Brain, the **Profiling** tab charts where the engine spent
|
||||
each turn's wall time (I/O wait, expert matmul, attention, LM head) from the
|
||||
server's `/profile` endpoint — a rolling window of per-turn `PROF` snapshots
|
||||
emitted by the engine.
|
||||
|
||||
The test suite stays browser-light: API requests use a mocked `fetch`, while
|
||||
runtime capability and storage behavior are covered through pure helpers. It
|
||||
checks that `/health` is resolved next to (not below) the OpenAI `/v1` prefix,
|
||||
checks that `/health` and `/profile` are resolved next to (not below) the OpenAI `/v1` prefix,
|
||||
supports both boolean and numeric `scheduler.active` responses, and sends the
|
||||
colibrì-specific `cache_slot` field only when KV-slot support was advertised.
|
||||
|
||||
|
||||
+5
-2
@@ -31,6 +31,7 @@ import { Textarea } from "@/components/ui/textarea"
|
||||
import { getHealth, listModels, streamChat, type ChatMessage, type HealthResponse, type StreamChatResult } from "@/lib/api"
|
||||
import { activeRequests, supportsCacheSlots } from "@/lib/runtime"
|
||||
import { Brain } from "./Brain"
|
||||
import { Profiling } from "./Profiling"
|
||||
import { persistPublicSettings, stored } from "@/lib/storage"
|
||||
import { cn } from "@/lib/utils"
|
||||
|
||||
@@ -73,7 +74,7 @@ export default function App() {
|
||||
const [totalTokens, setTotalTokens] = useState({ prompt: 0, completion: 0 })
|
||||
const [connecting, setConnecting] = useState(false)
|
||||
const [connected, setConnected] = useState(false)
|
||||
const [view, setView] = useState<"chat" | "brain">("chat")
|
||||
const [view, setView] = useState<"chat" | "brain" | "profiling">("chat")
|
||||
const [error, setError] = useState("")
|
||||
const autoConnected = useRef(false)
|
||||
const abortRef = useRef<AbortController | null>(null)
|
||||
@@ -314,6 +315,7 @@ export default function App() {
|
||||
<div className="view-tabs">
|
||||
<button className={view === "chat" ? "active" : ""} onClick={() => setView("chat")}><MessageSquareText className="size-3.5" /> Chat</button>
|
||||
<button className={view === "brain" ? "active" : ""} onClick={() => setView("brain")}><BrainCircuit className="size-3.5" /> Brain</button>
|
||||
<button className={view === "profiling" ? "active" : ""} onClick={() => setView("profiling")}><Gauge className="size-3.5" /> Profiling</button>
|
||||
</div>
|
||||
<div className="top-actions">
|
||||
{loading && tokenCount > 0 ? <Badge className="badge-live"><Zap className="size-3 flash" /> {tokenCount} tokens</Badge> : null}
|
||||
@@ -326,7 +328,8 @@ export default function App() {
|
||||
</div>
|
||||
</header>
|
||||
|
||||
{view === "brain" ? <Brain baseUrl={baseUrl} apiKey={apiKey} connected={connected} /> : <>
|
||||
{view === "brain" ? <Brain baseUrl={baseUrl} apiKey={apiKey} connected={connected} />
|
||||
: view === "profiling" ? <Profiling baseUrl={baseUrl} apiKey={apiKey} connected={connected} /> : <>
|
||||
|
||||
<div className="conversation">
|
||||
{!messages.length ? (
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
import { useEffect, useState } from "react"
|
||||
import { Activity, Gauge, HardDrive, Timer } from "lucide-react"
|
||||
|
||||
import { getProfile, type ProfileTurn } from "@/lib/api"
|
||||
|
||||
/* Wall-time phases stacked per turn. The order is the palette's CVD-safe slot
|
||||
* order (validated as a set on this surface) — identity never leans on colour
|
||||
* alone: segments keep 2px gaps, the legend is always shown and the table
|
||||
* carries the exact numbers. Disk *service* time is reported separately: it
|
||||
* runs on I/O threads overlapped with compute, so only the stall the compute
|
||||
* thread actually felt (I/O wait) belongs inside the wall-time stack. */
|
||||
const PHASES = [
|
||||
{ key: "expert_wait_s", name: "I/O wait", color: "#3987e5" },
|
||||
{ key: "expert_matmul_s", name: "Expert matmul", color: "#199e70" },
|
||||
{ key: "attention_s", name: "Attention", color: "#c98500" },
|
||||
{ key: "lm_head_s", name: "LM head", color: "#008300" },
|
||||
{ key: "other_s", name: "Other", color: "#9085e9" },
|
||||
] as const
|
||||
|
||||
interface Turn extends ProfileTurn { other_s: number; toks: number }
|
||||
|
||||
const derive = (turn: ProfileTurn): Turn => ({
|
||||
...turn,
|
||||
other_s: Math.max(0, turn.wall_s - turn.expert_wait_s - turn.expert_matmul_s - turn.attention_s - turn.lm_head_s),
|
||||
toks: turn.wall_s > 0 ? turn.completion_tokens / turn.wall_s : 0,
|
||||
})
|
||||
|
||||
const seconds = (value: number) => (value >= 10 ? value.toFixed(1) : value.toFixed(2)) + "s"
|
||||
|
||||
function ShareBar({ label, turns }: { label: string; turns: Turn[] }) {
|
||||
const total = turns.reduce((sum, turn) => sum + turn.wall_s, 0)
|
||||
const parts = PHASES.map((phase) => ({ ...phase, value: turns.reduce((sum, turn) => sum + turn[phase.key], 0) }))
|
||||
return (
|
||||
<div className="prof-share">
|
||||
<div className="prof-share-head"><span>{label}</span><code>{seconds(total)}</code></div>
|
||||
<div className="prof-share-bar" role="img" aria-label={parts.map((part) => `${part.name} ${seconds(part.value)}`).join(", ")}>
|
||||
{parts.map((part) => {
|
||||
const share = total > 0 ? part.value / total : 0
|
||||
return share > 0.001 ? (
|
||||
<span key={part.key} style={{ width: `${100 * share}%`, background: part.color }} title={`${part.name} — ${seconds(part.value)} (${(100 * share).toFixed(1)}%)`}>
|
||||
{share >= 0.09 ? `${Math.round(100 * share)}%` : ""}
|
||||
</span>
|
||||
) : null
|
||||
})}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
/* Column chart over the recent turns; oldest on the left. Stacked mode draws the
|
||||
* wall-time composition, plain mode a single series (no legend — the title names it). */
|
||||
function TurnColumns({ turns, stacked, height, format }: { turns: Turn[]; stacked: boolean; height: number; format: (turn: Turn) => string }) {
|
||||
const [hover, setHover] = useState<number | null>(null)
|
||||
const peak = Math.max(...turns.map((turn) => (stacked ? turn.wall_s : turn.toks)), 1e-9)
|
||||
const gap = 2
|
||||
const width = Math.max(1, (100 - gap * (turns.length - 1)) / turns.length)
|
||||
return (
|
||||
<div className="prof-plot" onMouseLeave={() => setHover(null)}>
|
||||
<svg viewBox={`0 0 100 ${height}`} preserveAspectRatio="none" aria-hidden="true">
|
||||
{[0.25, 0.5, 0.75].map((line) => <line key={line} x1="0" x2="100" y1={height * line} y2={height * line} className="prof-grid" />)}
|
||||
{turns.map((turn, index) => {
|
||||
const x = index * (width + gap)
|
||||
if (!stacked) {
|
||||
const h = (height * turn.toks) / peak
|
||||
return <rect key={index} x={x} y={height - h} width={width} height={h} rx="1" fill="var(--primary)" opacity={hover === null || hover === index ? 1 : 0.45} />
|
||||
}
|
||||
let y = height
|
||||
return PHASES.map((phase) => {
|
||||
const h = (height * turn[phase.key]) / peak
|
||||
y -= h
|
||||
return h > 0.1 ? <rect key={`${index}-${phase.key}`} x={x} y={y + 0.35} width={width} height={Math.max(h - 0.7, 0.35)} fill={phase.color} opacity={hover === null || hover === index ? 1 : 0.45} /> : null
|
||||
})
|
||||
})}
|
||||
{/* hit targets bigger than the marks */}
|
||||
{turns.map((_, index) => <rect key={index} x={index * (width + gap) - gap / 2} y="0" width={width + gap} height={height} fill="transparent" onMouseEnter={() => setHover(index)} />)}
|
||||
</svg>
|
||||
<div className="prof-plot-foot">
|
||||
<span>{turns.length > 1 ? `${turns.length} turns · oldest → newest` : "1 turn"}</span>
|
||||
<code>{hover !== null && turns[hover] ? format(turns[hover]) : `peak ${stacked ? seconds(peak) : peak.toFixed(1) + " tok/s"}`}</code>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
export function Profiling({ baseUrl, apiKey, connected }: { baseUrl: string; apiKey: string; connected: boolean }) {
|
||||
const [turns, setTurns] = useState<Turn[]>([])
|
||||
|
||||
useEffect(() => {
|
||||
if (!connected) return
|
||||
let disposed = false
|
||||
const poll = async () => {
|
||||
if (document.visibilityState === "hidden") return
|
||||
try {
|
||||
const result = await getProfile(baseUrl, apiKey)
|
||||
if (!disposed) setTurns(result.turns.map(derive))
|
||||
} catch { /* engine busy or restarting — keep the last snapshot */ }
|
||||
}
|
||||
void poll()
|
||||
const timer = window.setInterval(() => void poll(), 2000)
|
||||
return () => { disposed = true; window.clearInterval(timer) }
|
||||
}, [baseUrl, apiKey, connected])
|
||||
|
||||
const latest = turns[turns.length - 1]
|
||||
const recent = turns.slice(-40)
|
||||
const diskService = turns.reduce((sum, turn) => sum + turn.expert_disk_s, 0)
|
||||
|
||||
return (
|
||||
<div className="prof-page">
|
||||
<div className="prof-head">
|
||||
<div className="section-title"><Gauge className="size-4" /> Profiling — where the engine spends each turn</div>
|
||||
<div className="prof-legend">
|
||||
{PHASES.map((phase) => <span key={phase.key}><i style={{ background: phase.color }} />{phase.name}</span>)}
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{!latest ? (
|
||||
<p className="runtime-unavailable">{connected ? "No profiled turns yet — send a chat message and the breakdown appears here." : "Connect to the engine to collect per-turn timings."}</p>
|
||||
) : (
|
||||
<>
|
||||
<div className="prof-tiles">
|
||||
<div><span><Gauge className="size-3" /> Last turn</span><strong>{latest.toks.toFixed(1)}</strong><small>tok/s</small></div>
|
||||
<div><span><Timer className="size-3" /> Wall time</span><strong>{seconds(latest.wall_s)}</strong><small>{latest.prompt_tokens} → {latest.completion_tokens} tokens</small></div>
|
||||
<div><span><Activity className="size-3" /> Batching</span><strong>{latest.forwards > 0 ? (latest.completion_tokens / latest.forwards).toFixed(2) : "—"}</strong><small>tokens / forward</small></div>
|
||||
<div><span><HardDrive className="size-3" /> Disk service</span><strong>{seconds(latest.expert_disk_s)}</strong><small>overlapped with compute</small></div>
|
||||
</div>
|
||||
|
||||
<div className="prof-shares">
|
||||
<ShareBar label="Last turn" turns={[latest]} />
|
||||
{turns.length > 1 ? <ShareBar label={`Window · last ${turns.length} turns`} turns={turns} /> : null}
|
||||
</div>
|
||||
|
||||
<div className="prof-charts">
|
||||
<div className="prof-chart">
|
||||
<div className="prof-chart-title">Throughput per turn (tok/s)</div>
|
||||
<TurnColumns turns={recent} stacked={false} height={36} format={(turn) => `${turn.toks.toFixed(1)} tok/s · ${turn.completion_tokens} tokens`} />
|
||||
</div>
|
||||
<div className="prof-chart">
|
||||
<div className="prof-chart-title">Turn wall time by phase (s)</div>
|
||||
<TurnColumns turns={recent} stacked height={36} format={(turn) => `${seconds(turn.wall_s)} · ${PHASES.map((phase) => `${phase.name} ${seconds(turn[phase.key])}`).join(" · ")}`} />
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="prof-table-wrap">
|
||||
<table className="prof-table">
|
||||
<thead><tr><th>Turn</th><th>Tokens</th><th>tok/s</th><th>Wall</th>{PHASES.map((phase) => <th key={phase.key}><i style={{ background: phase.color }} />{phase.name}</th>)}<th>Disk service</th></tr></thead>
|
||||
<tbody>
|
||||
{recent.slice().reverse().map((turn, index) => (
|
||||
<tr key={turns.length - index}>
|
||||
<td>{turns.length - index}</td>
|
||||
<td>{turn.prompt_tokens} → {turn.completion_tokens}</td>
|
||||
<td>{turn.toks.toFixed(1)}</td>
|
||||
<td>{seconds(turn.wall_s)}</td>
|
||||
{PHASES.map((phase) => <td key={phase.key}>{seconds(turn[phase.key])}</td>)}
|
||||
<td>{seconds(turn.expert_disk_s)}</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
{diskService > 0 ? <p className="prof-note">Disk service is time spent reading experts on I/O threads; it overlaps with compute, so only the <em>I/O wait</em> the compute thread felt counts inside the wall-time stack. With multiple KV sessions the shares describe the whole engine over the turn's window.</p> : null}
|
||||
</div>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -151,3 +151,39 @@ button:focus-visible, input:focus-visible, textarea:focus-visible, select:focus-
|
||||
.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; }
|
||||
|
||||
/* ---- Profiling page ---- */
|
||||
.prof-page { flex: 1; display: flex; flex-direction: column; gap: 16px; padding: 18px 22px; overflow: auto; }
|
||||
.prof-head { display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 10px; }
|
||||
.prof-legend { display: flex; flex-wrap: wrap; gap: 12px; font-size: 11px; color: #a5b0b4; align-items: center; }
|
||||
.prof-legend i { width: 9px; height: 9px; border-radius: 2px; display: inline-block; margin-right: 5px; vertical-align: -1px; }
|
||||
.prof-tiles { display: grid; grid-template-columns: repeat(4, 1fr); gap: 10px; }
|
||||
.prof-tiles > div { display: grid; gap: 4px; padding: 12px 14px; border: 1px solid var(--border); border-radius: 11px; background: var(--card); }
|
||||
.prof-tiles span { display: flex; align-items: center; gap: 5px; color: #7f8d92; font-size: 9px; font-weight: 700; letter-spacing: .08em; text-transform: uppercase; }
|
||||
.prof-tiles strong { font: 600 22px ui-monospace, SFMono-Regular, monospace; font-variant-numeric: tabular-nums; }
|
||||
.prof-tiles small { color: var(--muted-foreground); font-size: 10px; }
|
||||
.prof-shares { display: grid; gap: 10px; padding: 14px; border: 1px solid var(--border); border-radius: 11px; background: var(--card); }
|
||||
.prof-share { display: grid; gap: 6px; }
|
||||
.prof-share-head { display: flex; justify-content: space-between; color: #a5b0b4; font-size: 11px; font-weight: 600; }
|
||||
.prof-share-head code { color: var(--foreground); font-size: 11px; font-variant-numeric: tabular-nums; }
|
||||
.prof-share-bar { display: flex; gap: 2px; height: 22px; border-radius: 6px; overflow: hidden; }
|
||||
.prof-share-bar span { display: grid; place-items: center; min-width: 0; overflow: hidden; color: #06090b; font-size: 10px; font-weight: 700; font-variant-numeric: tabular-nums; transition: width .5s ease; }
|
||||
.prof-charts { display: grid; grid-template-columns: 1fr 1fr; gap: 10px; }
|
||||
.prof-chart { display: grid; gap: 8px; padding: 12px 14px; border: 1px solid var(--border); border-radius: 11px; background: var(--card); }
|
||||
.prof-chart-title { color: #a5b0b4; font-size: 11px; font-weight: 600; }
|
||||
.prof-plot svg { display: block; width: 100%; height: 120px; }
|
||||
.prof-grid { stroke: var(--border); stroke-width: .3; }
|
||||
.prof-plot-foot { display: flex; justify-content: space-between; gap: 8px; margin-top: 6px; color: #7f8d92; font-size: 10px; }
|
||||
.prof-plot-foot code { color: var(--foreground); font-variant-numeric: tabular-nums; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
|
||||
.prof-table-wrap { overflow-x: auto; border: 1px solid var(--border); border-radius: 11px; background: var(--card); }
|
||||
.prof-table { width: 100%; border-collapse: collapse; font-size: 11px; font-variant-numeric: tabular-nums; }
|
||||
.prof-table th { padding: 9px 12px; border-bottom: 1px solid var(--border); color: #7f8d92; font-size: 9px; font-weight: 700; letter-spacing: .07em; text-transform: uppercase; text-align: left; white-space: nowrap; }
|
||||
.prof-table th i { width: 8px; height: 8px; border-radius: 2px; display: inline-block; margin-right: 5px; vertical-align: -1px; }
|
||||
.prof-table td { padding: 7px 12px; border-bottom: 1px solid var(--border); color: #c3ccd0; white-space: nowrap; }
|
||||
.prof-table tbody tr:last-child td { border-bottom: 0; }
|
||||
.prof-note { margin: 0; padding: 10px 12px; color: var(--muted-foreground); font-size: 10px; line-height: 1.5; border-top: 1px solid var(--border); }
|
||||
@media (max-width: 900px) {
|
||||
.prof-page { padding: 10px; }
|
||||
.prof-tiles { grid-template-columns: 1fr 1fr; }
|
||||
.prof-charts { grid-template-columns: 1fr; }
|
||||
}
|
||||
|
||||
+14
-1
@@ -1,6 +1,6 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest"
|
||||
|
||||
import { extractSSE, getHealth, serverEndpoint, streamChat } from "./api"
|
||||
import { extractSSE, getHealth, getProfile, serverEndpoint, streamChat } from "./api"
|
||||
|
||||
afterEach(() => vi.unstubAllGlobals())
|
||||
|
||||
@@ -36,6 +36,19 @@ describe("runtime API", () => {
|
||||
headers: expect.objectContaining({ Authorization: "Bearer secret" }),
|
||||
}))
|
||||
})
|
||||
|
||||
it("requests the profiling history next to the OpenAI v1 prefix", async () => {
|
||||
const turn = {
|
||||
wall_s: 2.5, prompt_tokens: 7, completion_tokens: 12,
|
||||
expert_disk_s: 0.4, expert_wait_s: 0.1, expert_matmul_s: 0.9,
|
||||
attention_s: 0.6, lm_head_s: 0.2, forwards: 15,
|
||||
}
|
||||
const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ seq: 1, turns: [turn] })))
|
||||
vi.stubGlobal("fetch", fetchMock)
|
||||
|
||||
await expect(getProfile("http://localhost:8000/v1/")).resolves.toEqual({ seq: 1, turns: [turn] })
|
||||
expect(fetchMock).toHaveBeenCalledWith("http://localhost:8000/profile", expect.anything())
|
||||
})
|
||||
})
|
||||
|
||||
describe("chat request extensions", () => {
|
||||
|
||||
@@ -49,6 +49,23 @@ export interface HealthResponse {
|
||||
hwinfo?: HwinfoHealth
|
||||
}
|
||||
|
||||
export interface ProfileTurn {
|
||||
wall_s: number
|
||||
prompt_tokens: number
|
||||
completion_tokens: number
|
||||
expert_disk_s: number
|
||||
expert_wait_s: number
|
||||
expert_matmul_s: number
|
||||
attention_s: number
|
||||
lm_head_s: number
|
||||
forwards: number
|
||||
}
|
||||
|
||||
export interface ProfileResponse {
|
||||
seq: number
|
||||
turns: ProfileTurn[]
|
||||
}
|
||||
|
||||
export interface TokenUsage {
|
||||
prompt_tokens: number
|
||||
completion_tokens: number
|
||||
@@ -100,6 +117,12 @@ export async function getHealth(baseUrl: string, apiKey = "", signal?: AbortSign
|
||||
return (await response.json()) as HealthResponse
|
||||
}
|
||||
|
||||
export async function getProfile(baseUrl: string, apiKey = "", signal?: AbortSignal): Promise<ProfileResponse> {
|
||||
const response = await fetch(serverEndpoint(baseUrl, "profile"), { headers: headers(apiKey), signal })
|
||||
if (!response.ok) throw new Error(await responseError(response))
|
||||
return (await response.json()) as ProfileResponse
|
||||
}
|
||||
|
||||
export function extractSSE(buffer: string) {
|
||||
const frames = buffer.split(/\r?\n\r?\n/)
|
||||
const rest = frames.pop() || ""
|
||||
|
||||
Reference in New Issue
Block a user