Merge remote-tracking branch 'origin/dev' into pr330
# Conflicts: # c/Makefile
This commit is contained in:
JustVugg
@@ -0,0 +1,98 @@
|
||||
# Efficiency suite — regression tests + optimization dossier
|
||||
|
||||
Two layers:
|
||||
|
||||
1. **`test_inefficiency.py`** — tiny-model *asserted* regression tests. Fast
|
||||
(~0.15s/run), gate CI, catch breakage. Run as part of `make test`.
|
||||
2. **`test_efficiency_report.py`** — an *opt-in optimization dossier* for a real
|
||||
model. Runs every instrumentation flag, prints a 9-section report answering
|
||||
*what is doing what, when, with what, is it inefficient, how to improve*.
|
||||
Never fails CI (it's a report, not a gate).
|
||||
|
||||
## The dossier (what you run when optimizing)
|
||||
|
||||
```bash
|
||||
# CPU-only (safe, fast to validate):
|
||||
COLI_EFFICIENCY_MODEL=../glm52_i4_g64 make efficiency-report
|
||||
|
||||
# CUDA (dense + expert tiers — needs a CUDA build, see below):
|
||||
COLI_EFFICIENCY_MODEL=../glm52_i4_g64 COLI_EFFICIENCY_CUDA=1 make efficiency-report
|
||||
```
|
||||
|
||||
It turns ON every observability flag the engine supports — `PROF=1`,
|
||||
`COLI_CUDA_PROFILE=1`, `CACHE_ROUTE=1` (auto-unlocks `route_agree`/`route_kl`),
|
||||
`DISK_SPLIT=1`, `LOOKA=1` — so nothing the engine can tell you is left dark.
|
||||
None of these change the computed output; they only add telemetry.
|
||||
|
||||
The 9 sections, and the question each answers:
|
||||
|
||||
| § | section | answers |
|
||||
|---|---|---|
|
||||
| 1 | PROVENANCE | what is running, on what CPU/backend, with what effective config |
|
||||
| 2 | THROUGHPUT | tok/s + forward-latency p50/p90/p99/max (is the tail healthy?) |
|
||||
| 3 | WHERE TIME GOES | the 5 PROFILE phases as % of decode + absolute seconds + verdict |
|
||||
| 3a | ATTENTION BREAKDOWN | attention split into projection/RoPE, score-softmax-value, output |
|
||||
| 4 | EXPERT CACHE | hit %, experts-loaded/token vs baseline topk |
|
||||
| 5 | DISK I/O | GB fetched, MB/token, GB/s, read-service vs felt-wait, phase split |
|
||||
| 5a | DISK-LOAD SPLIT | loads by decode phase (draft/absorb/verify) + MTP-vs-main bytes |
|
||||
| 6 | ROUTING QUALITY | route_agree %, route_kl, cache swaps |
|
||||
| 6a | ROUTING PREDICTABILITY | LOOKAHEAD recall per predictor (which prefetch wins) |
|
||||
| 7 | SPECULATION | tokens/forward, MTP acceptance % |
|
||||
| 8 | GPU TIERS | resident tensors, expert tier (count/GB/calls), H2D/kernel/D2H ms |
|
||||
|
||||
Every line that crosses an advisory threshold is marked `[FLAG]` with the
|
||||
concrete lever to pull (raise RAM_GB, add PIN_GB, try DIRECT=1, lower CTX, …),
|
||||
and all flags repeat in a summary at the end.
|
||||
|
||||
## Tunable thresholds
|
||||
|
||||
The `IS IT INEFFICIENT?` lines are advisory constants at the top of
|
||||
`test_efficiency_report.py`:
|
||||
|
||||
| constant | default | meaning |
|
||||
|---|---|---|
|
||||
| `DISK_WAIT_DOMINANT` | 0.40 | >40% decode waiting on expert reads → I/O-bound |
|
||||
| `LOW_HIT_RATE` | 0.30 | <30% cache hit → thrashing |
|
||||
| `LOW_ROUTE_AGREE` | 0.80 | <80% routing overlap → prefetch guessing wrong |
|
||||
| `HIGH_TAIL_RATIO` | 3.0 | p99 > 3× p50 → decode stalls |
|
||||
| `LOW_MTP_ACCEPT` | 0.20 | <20% MTP acceptance → draft decoder is dead weight |
|
||||
|
||||
The tiny-model asserted floors live in `tools/efficiency.py` (`TINY_TOK_S_FLOOR`,
|
||||
`MAX_DISK_WAIT_SHARE`, `MIN_CPU_CUDA_AGREEMENT`).
|
||||
|
||||
## The regression tests (what gates CI)
|
||||
|
||||
`test_inefficiency.py` runs on the bundled `glm_tiny` model and asserts:
|
||||
|
||||
- telemetry parses (no format drift)
|
||||
- tiny tok/s ≥ floor (throughput regression)
|
||||
- PROFILE phases present and non-negative (accounting sanity)
|
||||
- disk-wait not dominant on a resident model (I/O-path regression)
|
||||
- CPU determinism (two greedy runs agree)
|
||||
- **CUDA** (skip unless CUDA built): init path, dense uploads VRAM, CPU-vs-CUDA
|
||||
argmax agreement ≥ 70% (kernel-correctness guard)
|
||||
|
||||
```bash
|
||||
make efficiency # tiny CPU tests
|
||||
make efficiency-cuda # tiny CUDA tests (needs CUDA build)
|
||||
```
|
||||
|
||||
## CUDA build prerequisite
|
||||
|
||||
The default `make glm.exe` builds **without** CUDA. The CUDA tests and the CUDA
|
||||
dossier need a host built with `-DCOLI_CUDA` plus the runtime DLL:
|
||||
|
||||
```bash
|
||||
make clean && make glm.exe CUDA_DLL=1 && make cuda-dll
|
||||
```
|
||||
|
||||
`make efficiency-cuda` auto-skips with a clear message if the host is CPU-only
|
||||
(it scans the binary for the "CPU-only" marker the engine embeds).
|
||||
|
||||
## Files
|
||||
|
||||
- `tools/efficiency.py` — shared harness: `parse_run()` (captures every
|
||||
telemetry signal), `run_engine()`, thresholds. Reuses `PROFILE_RE`/`SPEED_RE`
|
||||
from `tools/benchmark_cuda_fixture.py`.
|
||||
- `tests/test_inefficiency.py` — tiny-model asserted tests (CPU + CUDA).
|
||||
- `tests/test_efficiency_report.py` — the opt-in optimization dossier.
|
||||
@@ -0,0 +1,165 @@
|
||||
/* Microbenchmark: old (full-qsort) vs new (quickselect partial-select) DSA top-keep.
|
||||
*
|
||||
* This is NOT a unit test -- test_dsa_select.c proves correctness. This measures the
|
||||
* headline claim of #356: that replacing the O(nk log nk) qsort over all nk context
|
||||
* scores with an O(nk) partial_select_desc is materially faster per call, which is
|
||||
* the win the issue was opened for -- and that the win GROWS with context length
|
||||
* (because quickselect is linear average, qsort is n-log-n).
|
||||
*
|
||||
* It re-implements the OLD top-keep inline (qsort + threshold + scans) on a private
|
||||
* buffer so the A/B runs in one process, same inputs, same warm caches -- a controlled
|
||||
* comparison. It calls the REAL (new) partial_select_desc via the include-glm.c
|
||||
* pattern, replicating the production threshold derivation + scans.
|
||||
*
|
||||
* Methodology (chosen to be honest, not to flatter the change):
|
||||
* - keep = 2048 (the real GLM-5.2 index_topk), nk swept across context lengths from
|
||||
* the 2049 activation boundary up to 65536 (a long conversation).
|
||||
* - Three score shapes: (a) realistic peaked -- a few hot keys, long tail, the shape
|
||||
* real DSA attention scores take; (b) uniform random -- no structure; (c) a plateau
|
||||
* of ties, to exercise the boundary-membership path.
|
||||
* - Each (shape, nk) is timed over N_REPEAT=2000 iterations, with the scores frozen
|
||||
* so both algorithms do IDENTICAL work. We report median ns/call and the new/old
|
||||
* ratio. A warmup pass primes caches before timing.
|
||||
*
|
||||
* Run: make tests/bench_dsa_select && ./tests/bench_dsa_select (not in TEST_BINS)
|
||||
*/
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
#include <math.h>
|
||||
#include <stdint.h>
|
||||
|
||||
/* ---- the OLD algorithm, verbatim from dev before #356, on a private buffer ---- */
|
||||
static int cmp_pdesc_old(const void *a, const void *b){
|
||||
float x=*(const float*)a, y=*(const float*)b; return x<y?1:x>y?-1:0; }
|
||||
static void keep_old(const float *isc, int nk, int keep, int *dst, int *nd_out){
|
||||
float *tmp=malloc((size_t)nk*sizeof(float));
|
||||
memcpy(tmp,isc,(size_t)nk*sizeof(float));
|
||||
qsort(tmp,(size_t)nk,sizeof(float),cmp_pdesc_old);
|
||||
float thr=tmp[keep-1]; int nd=0;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]>thr) dst[nd++]=t;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]==thr) dst[nd++]=t;
|
||||
free(tmp); *nd_out=nd;
|
||||
}
|
||||
|
||||
/* ---- timing: median of N_REPEAT runs in ns/call, sorted ascending ---- */
|
||||
#define N_REPEAT 2000
|
||||
static double bench_ns(void (*fn)(const float*,int,int,int*,int*),
|
||||
const float *isc, int nk, int keep){
|
||||
static double ts[N_REPEAT]; int *dst=malloc((size_t)nk*sizeof(int)); int nd;
|
||||
for(int r=0;r<N_REPEAT;r++){
|
||||
double t0=now_s();
|
||||
fn(isc,nk,keep,dst,&nd);
|
||||
ts[r]=(now_s()-t0)*1e9;
|
||||
}
|
||||
for(int a=1;a<N_REPEAT;a++){ double k=ts[a]; int b=a-1;
|
||||
while(b>=0 && ts[b]>k){ ts[b+1]=ts[b]; b--; } ts[b+1]=k; }
|
||||
free(dst);
|
||||
return ts[N_REPEAT/2];
|
||||
}
|
||||
|
||||
/* Sort an array of doubles ascending (median-of-medians aggregation below). */
|
||||
static void dsort(double *a, int n){
|
||||
for(int s=1;s<n;s++){ double k=a[s]; int b=s-1;
|
||||
while(b>=0 && a[b]>k){ a[b+1]=a[b]; b--; } a[b+1]=k; }
|
||||
}
|
||||
|
||||
/* the NEW algorithm calls the real partial_select_desc + replicates the production
|
||||
* threshold derivation and position scans. */
|
||||
static void keep_new(const float *isc, int nk, int keep, int *dst, int *nd_out){
|
||||
float *tmp=malloc((size_t)nk*sizeof(float));
|
||||
memcpy(tmp,isc,(size_t)nk*sizeof(float));
|
||||
partial_select_desc(tmp,nk,keep);
|
||||
float thr=tmp[0]; for(int t=1;t<keep;t++) if(tmp[t]<thr) thr=tmp[t];
|
||||
int nd=0;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]>thr) dst[nd++]=t;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]==thr) dst[nd++]=t;
|
||||
free(tmp); *nd_out=nd;
|
||||
}
|
||||
|
||||
/* deterministic score fill for three shapes */
|
||||
static uint32_t brng = 0xA5A5A5A5u;
|
||||
static void brng_seed(uint32_t s){ brng = s; }
|
||||
static double brand(void){ brng ^= brng << 13; brng ^= brng >> 17; brng ^= brng << 5;
|
||||
return (double)(brng >> 8) * (1.0 / 16777216.0); }
|
||||
static void fill_realistic(float *isc, int nk){ /* few hot, long distinct tail */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(-1.0 - brand()*4.0);
|
||||
isc[0]=3.f; isc[nk/50<nk?nk/50:nk-1]=1.f; isc[nk/200<nk?nk/200:nk-1]=0.5f;
|
||||
}
|
||||
static void fill_uniform(float *isc, int nk){ /* no structure */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(brand()*1000.0);
|
||||
}
|
||||
static void fill_plateau(float *isc, int nk){ /* tie blocks -> boundary path */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(-(double)(i/7));
|
||||
}
|
||||
|
||||
/* Multi-seed aggregation. Per @KingIcyCreamProjects (#357 thread): with a single frozen
|
||||
* input per cell, quickselect's deterministic median-of-three pivot means one lucky input
|
||||
* can spike a single nk row (an observed ~75x at nk=8192 on a 9950X3D that was really a
|
||||
* ~13-40x algorithm). Two bugs compounded it: (1) the old bench drew ONE input per cell;
|
||||
* (2) brng was never reset, so each cell's input depended on every prior cell's draws --
|
||||
* reordering nks[] silently shifted all later inputs. Both fixed here: brng is reseeded
|
||||
* per (shape, nk, seed), and we take the MEDIAN of N_SEEDS independent inputs, each itself
|
||||
* a median over N_REPEAT timing reps. A lucky pivot now moves one of the N_SEEDS samples,
|
||||
* not the reported number. */
|
||||
|
||||
int main(void){
|
||||
int keep = 2048; /* GLM-5.2 index_topk */
|
||||
int nks[] = {2049, 4096, 8192, 16384, 32768, 65536};
|
||||
const int N_SEEDS = 11; /* odd so the median is a real sample, not interpolated */
|
||||
float *isc = malloc((size_t)65536*sizeof(float));
|
||||
int *dst = malloc((size_t)65536*sizeof(int)); int nd;
|
||||
double *old_seeds = malloc((size_t)N_SEEDS*sizeof(double));
|
||||
double *new_seeds = malloc((size_t)N_SEEDS*sizeof(double));
|
||||
|
||||
struct { const char *name; void (*fill)(float*,int); } shapes[] = {
|
||||
{ "realistic", fill_realistic },
|
||||
{ "uniform", fill_uniform },
|
||||
{ "plateau", fill_plateau },
|
||||
};
|
||||
|
||||
printf("bench_dsa_select: DSA top-keep, old (qsort) vs new (partial-select) keep=%d (median of %d seeds x %d reps)\n",
|
||||
keep, N_SEEDS, N_REPEAT);
|
||||
printf("%-12s %7s %14s %14s %9s\n", "shape", "nk", "old ns/call", "new ns/call", "speedup");
|
||||
printf("------------------------------------------------------------------------\n");
|
||||
|
||||
for(size_t sh=0; sh<sizeof(shapes)/sizeof(shapes[0]); sh++){
|
||||
for(size_t ni=0; ni<sizeof(nks)/sizeof(nks[0]); ni++){
|
||||
int nk=nks[ni];
|
||||
int bad = 0;
|
||||
for(int sd=0; sd<N_SEEDS; sd++){
|
||||
/* reseed per (shape,nk,seed) so each cell's input is reproducible and
|
||||
* independent of cell ordering, and so lucky pivots are sampled, not fixed. */
|
||||
brng_seed(0xA5A5A5A5u + (uint32_t)(sd*0x9E3779B9u));
|
||||
shapes[sh].fill(isc,nk);
|
||||
/* warmup both paths so caches/branch predictors are primed */
|
||||
for(int w=0; w<50; w++){ keep_old(isc,nk,keep,dst,&nd); keep_new(isc,nk,keep,dst,&nd); }
|
||||
/* sanity: both must keep exactly `keep` (correctness is test_dsa_select's
|
||||
* job, but a count divergence here would make the timing meaningless) */
|
||||
keep_old(isc,nk,keep,dst,&nd); int na=nd;
|
||||
keep_new(isc,nk,keep,dst,&nd); int nb=nd;
|
||||
if(na!=keep || nb!=keep){ bad++; continue; }
|
||||
old_seeds[sd] = bench_ns(keep_old,isc,nk,keep);
|
||||
new_seeds[sd] = bench_ns(keep_new,isc,nk,keep);
|
||||
}
|
||||
if(bad == N_SEEDS){
|
||||
printf("%-12s %7d (BAD COUNTS on all %d seeds, skipped)\n",
|
||||
shapes[sh].name, nk, bad);
|
||||
continue;
|
||||
}
|
||||
/* report median-of-seed-medians: robust to a single lucky/unlucky pivot */
|
||||
dsort(old_seeds, N_SEEDS);
|
||||
dsort(new_seeds, N_SEEDS);
|
||||
double t_old = old_seeds[N_SEEDS/2];
|
||||
double t_new = new_seeds[N_SEEDS/2];
|
||||
printf("%-12s %7d %14.0f %14.0f %8.2fx\n",
|
||||
shapes[sh].name, nk, t_old, t_new, t_old/t_new);
|
||||
}
|
||||
printf("\n");
|
||||
}
|
||||
|
||||
printf("bench_dsa_select: done (lower ns is better; speedup = old/new)\n");
|
||||
free(isc); free(dst); free(old_seeds); free(new_seeds);
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
/* Microbenchmark: old (full-vocab qsort) vs new (heap partial-select) top-p truncation.
|
||||
*
|
||||
* This is NOT a unit test -- test_topp.c proves correctness. This measures the headline
|
||||
* claim of #335: that replacing the O(V log V) qsort over all 151936 vocab entries with
|
||||
* an O(V) heapify + k*log-V pops is materially faster per call, which is the win the issue
|
||||
* was opened for.
|
||||
*
|
||||
* It re-implements the OLD dist_build inline (qsort + scan) on a private buffer so the A/B
|
||||
* runs in one process, same inputs, same warm caches -- a controlled comparison. It calls
|
||||
* the REAL (new) dist_build via the include-glm.c pattern on the global g_pbuf.
|
||||
*
|
||||
* Methodology (chosen to be honest, not to flatter the change):
|
||||
* - V = 151936 (the actual GLM-5.2 vocab), g_nuc swept across the values that matter
|
||||
* for serving: 0.5 / 0.9 (serve default) / 0.95 / 0.99.
|
||||
* - Three logit shapes: (a) realistic peaked -- one hot token, long exponential tail,
|
||||
* the shape real language-model logits take; (b) uniform -- worst case for the heap,
|
||||
* maximum pop count; (c) a plateau of ties, to exercise the tie path.
|
||||
* - Each (shape, nuc) is timed over N_REPEAT=2000 iterations, with the RNG/logits frozen
|
||||
* so both algorithms do IDENTICAL work. We report median ns/call and the new/old ratio.
|
||||
* - A warmup pass primes caches before timing.
|
||||
*
|
||||
* Run: make tests/bench_topp && ./tests/bench_topp (not in TEST_BINS -- not a gate)
|
||||
*/
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
#include <math.h>
|
||||
#include <stdint.h>
|
||||
|
||||
/* ---- the OLD algorithm, verbatim from dev before #354, on a private buffer ---- */
|
||||
static float *s_pbuf; static int *s_pidx; static double *s_ref;
|
||||
static int cmp_pdesc_old(const void *a, const void *b){
|
||||
double pa = s_ref[*(const int*)a], pb = s_ref[*(const int*)b];
|
||||
return pa < pb ? 1 : pa > pb ? -1 : 0; }
|
||||
static void dist_build_old(const float *lo, int V, double temp, double nuc){
|
||||
double mx = lo[0]; for (int i = 1; i < V; i++) if (lo[i] > mx) mx = lo[i];
|
||||
double s = 0, invt = 1.0 / (temp > 1e-4 ? temp : 1e-4);
|
||||
for (int i = 0; i < V; i++){ s_ref[i] = exp((lo[i]-mx)*invt); s += s_ref[i]; }
|
||||
for (int i = 0; i < V; i++) s_ref[i] /= s;
|
||||
if (nuc > 0 && nuc < 1.0){
|
||||
for (int i = 0; i < V; i++) s_pidx[i] = i;
|
||||
qsort(s_pidx, V, sizeof(int), cmp_pdesc_old);
|
||||
double cum = 0; int keep = V;
|
||||
for (int i = 0; i < V; i++){ cum += s_ref[s_pidx[i]]; if (cum >= nuc){ keep = i+1; break; } }
|
||||
double s2 = 0;
|
||||
for (int i = keep; i < V; i++) s_ref[s_pidx[i]] = 0;
|
||||
for (int i = 0; i < keep; i++) s2 += s_ref[s_pidx[i]];
|
||||
for (int i = 0; i < keep; i++) s_ref[s_pidx[i]] /= s2;
|
||||
}
|
||||
(void)s_pbuf;
|
||||
}
|
||||
|
||||
/* ---- timing: median of N_REPEAT runs in ns/call, sorted ascending ---- */
|
||||
#define N_REPEAT 2000
|
||||
static double bench_ns(void (*fn)(const float*,int,double,double),
|
||||
const float *lo, int V, double temp, double nuc){
|
||||
static double ts[N_REPEAT];
|
||||
for (int r = 0; r < N_REPEAT; r++){
|
||||
double t0 = now_s();
|
||||
fn(lo, V, temp, nuc);
|
||||
ts[r] = (now_s() - t0) * 1e9;
|
||||
}
|
||||
/* insertion sort the N_REPEAT samples (small), take median */
|
||||
for (int a = 1; a < N_REPEAT; a++){ double k = ts[a]; int b = a-1;
|
||||
while (b >= 0 && ts[b] > k){ ts[b+1] = ts[b]; b--; } ts[b+1] = k; }
|
||||
return ts[N_REPEAT/2];
|
||||
}
|
||||
|
||||
/* the NEW algorithm is the real dist_build, but it writes g_pbuf (not a private buf).
|
||||
* Wrap it so the bench signature matches, and set the globals it reads. */
|
||||
static void dist_build_new(const float *lo, int V, double temp, double nuc){
|
||||
g_temp = (float)temp; g_nuc = (float)nuc;
|
||||
dist_build(lo, V);
|
||||
}
|
||||
|
||||
/* deterministic logit fill for three shapes */
|
||||
static uint32_t brng = 0xA5A5A5A5u;
|
||||
static double brand(void){ brng ^= brng << 13; brng ^= brng >> 17; brng ^= brng << 5;
|
||||
return (double)(brng >> 8) * (1.0 / 16777216.0); }
|
||||
static void fill_realistic(float *lo, int V){ /* one hot, exponential tail -- like real logits */
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(-4.0 * brand() - (double)i * 0.0001);
|
||||
lo[0] = 6.f; lo[V/50] = 4.f; lo[V/200] = 3.f;
|
||||
}
|
||||
static void fill_uniform(float *lo, int V){ /* worst case for the heap: max pop count */
|
||||
for (int i = 0; i < V; i++) lo[i] = 0.f;
|
||||
}
|
||||
static void fill_plateau(float *lo, int V){ /* ties: blocks of equal value */
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(-(double)(i / 50));
|
||||
}
|
||||
|
||||
int main(void){
|
||||
int V = 151936;
|
||||
float *lo = malloc((size_t)V * sizeof(float));
|
||||
s_ref = malloc((size_t)V * sizeof(double));
|
||||
s_pidx = malloc((size_t)V * sizeof(int));
|
||||
/* force the new dist_build to allocate g_pbuf/g_pidx at full V once */
|
||||
g_temp = 0.7f; g_nuc = 0.9f; dist_build(lo, V);
|
||||
|
||||
double temp = 0.7;
|
||||
struct { const char *name; void (*fill)(float*,int); } shapes[] = {
|
||||
{ "realistic", fill_realistic },
|
||||
{ "uniform", fill_uniform },
|
||||
{ "plateau", fill_plateau },
|
||||
};
|
||||
double nucs[] = { 0.5, 0.9, 0.95, 0.99 };
|
||||
|
||||
printf("bench_topp: top-p truncation, old (qsort) vs new (heap) V=%d temp=%.2f\n", V, temp);
|
||||
printf("%-12s %6s %14s %14s %9s %9s\n", "shape", "nuc", "old ns/call", "new ns/call", "speedup", "keep");
|
||||
printf("-----------------------------------------------------------------------------\n");
|
||||
|
||||
for (size_t sh = 0; sh < sizeof(shapes)/sizeof(shapes[0]); sh++){
|
||||
shapes[sh].fill(lo, V);
|
||||
for (size_t ni = 0; ni < sizeof(nucs)/sizeof(nucs[0]); ni++){
|
||||
double nuc = nucs[ni];
|
||||
/* warmup both paths so caches/branch predictors are primed */
|
||||
for (int w = 0; w < 50; w++){ dist_build_old(lo, V, temp, nuc); dist_build_new(lo, V, temp, nuc); }
|
||||
double t_old = bench_ns(dist_build_old, lo, V, temp, nuc);
|
||||
double t_new = bench_ns(dist_build_new, lo, V, temp, nuc);
|
||||
/* keep count = non-zero entries the new path leaves (== old's keep) */
|
||||
int keep = 0; for (int i = 0; i < V; i++) if (g_pbuf[i] != 0.f) keep++;
|
||||
printf("%-12s %6.2f %14.0f %14.0f %8.2fx %9d\n",
|
||||
shapes[sh].name, nuc, t_old, t_new, t_old / t_new, keep);
|
||||
}
|
||||
printf("\n");
|
||||
}
|
||||
|
||||
printf("bench_topp: done (lower ns is better; speedup = old/new)\n");
|
||||
free(lo); free(s_ref); free(s_pidx);
|
||||
return 0;
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
/* DSA top-keep partial-select: the quickselect rewrite (#356) must produce a
|
||||
* BIT-IDENTICAL kept-position set to the old full-vocab qsort, for every score
|
||||
* shape the attention indexer can see.
|
||||
*
|
||||
* Why this test exists (#356): attention_rows() selects the top-`keep` context
|
||||
* keys (index_topk=2048 on GLM-5.2) to attend to. It previously did this by
|
||||
* full-qsorting all `nk` scores (O(nk log nk)) and reading tmp[keep-1] as the
|
||||
* threshold. It now does a partial_select_desc (quickselect, O(nk) average) and
|
||||
* takes the threshold as the min of the selected top-keep block. The contract is
|
||||
* subtle but STRONGER than test_topp's:
|
||||
*
|
||||
* The two position-order scans that build dst[] --
|
||||
* for t: if isc[t] > thr -> keep (strictly above threshold)
|
||||
* for t: if isc[t] == thr -> keep (ties, in position order)
|
||||
* -- are UNCHANGED by the rewrite. So if the threshold value is identical,
|
||||
* the kept-position set is identical element-by-element (not just as a
|
||||
* multiset, which is all the unstable sampling heap in #335 could promise).
|
||||
*
|
||||
* Strategy: drive the REAL partial_select_desc (via the include-glm.c pattern)
|
||||
* and replicate the production threshold derivation + scans, then compare the
|
||||
* resulting dst[] against an INDEPENDENT reference that re-implements the OLD
|
||||
* algorithm (full qsort + tmp[keep-1] threshold) on a private buffer. The kept
|
||||
* sets must be element-wise equal on every shape, including tie plateaus where
|
||||
* the boundary membership is decided by the position scan.
|
||||
*
|
||||
* We also directly unit-test partial_select_desc's partition invariant: after
|
||||
* the call, max(a[keep..n)) <= min(a[0..keep)) -- i.e. the keep largest really
|
||||
* did land in the prefix. This catches a broken quickselect even before the
|
||||
* end-to-end comparison.
|
||||
*
|
||||
* In-memory only (no scratch files), so it builds clean on the Windows MinGW CI
|
||||
* job without the unmerged compat shim. */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
#include <math.h>
|
||||
|
||||
static int g_nfails = 0;
|
||||
|
||||
#define FAIL(fmt, ...) do { \
|
||||
fprintf(stderr, " FAIL [%s nk=%d keep=%d shape=%s]: " fmt "\n", \
|
||||
label, nk, keep, shape_name, ##__VA_ARGS__); \
|
||||
g_nfails++; \
|
||||
return; \
|
||||
} while (0)
|
||||
|
||||
/* ---- independent reference: the OLD algorithm (full qsort + tmp[keep-1]) ---- */
|
||||
/* qsort comparator matching the production cmp_fdesc exactly (desc, unstable). */
|
||||
static int cmp_ref_desc(const void *a, const void *b){
|
||||
float x=*(const float*)a, y=*(const float*)b; return x<y?1:x>y?-1:0; }
|
||||
|
||||
/* Reproduce the OLD glm.c:2589-2596 exactly: copy, qsort desc, threshold =
|
||||
* tmp[keep-1], then the two position-order scans into dst[]. Returns nd. */
|
||||
static int keep_old(const float *isc, int nk, int keep, int *dst){
|
||||
float *tmp=malloc((size_t)nk*sizeof(float));
|
||||
memcpy(tmp,isc,(size_t)nk*sizeof(float));
|
||||
qsort(tmp,(size_t)nk,sizeof(float),cmp_ref_desc);
|
||||
float thr=tmp[keep-1];
|
||||
int nd=0;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]>thr) dst[nd++]=t;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]==thr) dst[nd++]=t;
|
||||
free(tmp);
|
||||
return nd;
|
||||
}
|
||||
|
||||
/* Reproduce the NEW glm.c path: partial_select desc, threshold = min of the
|
||||
* selected block, same two scans. Uses the REAL partial_select_desc from glm.c. */
|
||||
static int keep_new(const float *isc, int nk, int keep, int *dst){
|
||||
float *tmp=malloc((size_t)nk*sizeof(float));
|
||||
memcpy(tmp,isc,(size_t)nk*sizeof(float));
|
||||
partial_select_desc(tmp,nk,keep);
|
||||
float thr=tmp[0]; for(int t=1;t<keep;t++) if(tmp[t]<thr) thr=tmp[t];
|
||||
int nd=0;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]>thr) dst[nd++]=t;
|
||||
for(int t=0;t<nk && nd<keep;t++) if(isc[t]==thr) dst[nd++]=t;
|
||||
free(tmp);
|
||||
return nd;
|
||||
}
|
||||
|
||||
/* ---- direct unit test of the partition invariant ---- */
|
||||
static void check_partition(const char *label, const float *isc, int nk, int keep,
|
||||
const char *shape_name){
|
||||
float *tmp=malloc((size_t)nk*sizeof(float));
|
||||
memcpy(tmp,isc,(size_t)nk*sizeof(float));
|
||||
partial_select_desc(tmp,nk,keep);
|
||||
/* invariant: every element of tmp[0..keep) is >= every element of tmp[keep..n).
|
||||
* (>=, not >: equal values may sit on either side of the partition boundary,
|
||||
* which is fine -- the threshold is the MIN of the prefix, and the position
|
||||
* scan handles ties.) */
|
||||
float top_min=INFINITY, tail_max=-INFINITY;
|
||||
for(int i=0;i<keep;i++) if(tmp[i]<top_min) top_min=tmp[i];
|
||||
for(int i=keep;i<nk;i++) if(tmp[i]>tail_max) tail_max=tmp[i];
|
||||
if(!(top_min >= tail_max))
|
||||
FAIL("partition invariant violated: top_min=%.9g < tail_max=%.9g", top_min, tail_max);
|
||||
free(tmp);
|
||||
}
|
||||
|
||||
/* ---- end-to-end: old vs new kept-set must be element-wise identical ---- */
|
||||
static void check_case(const char *label, int nk, int keep, const char *shape_name,
|
||||
const float *isc){
|
||||
int *da=malloc((size_t)nk*sizeof(int));
|
||||
int *db=malloc((size_t)nk*sizeof(int));
|
||||
int na=keep_old(isc,nk,keep,da);
|
||||
int nb=keep_new(isc,nk,keep,db);
|
||||
|
||||
/* 1. both keep exactly `keep` positions (the contract: keep the top-keep by
|
||||
* count). A count mismatch is a real bug, not a tie artifact. */
|
||||
if(na!=keep) FAIL("old kept %d, expected %d (old path is the reference)", na, keep);
|
||||
if(nb!=keep) FAIL("new kept %d, expected %d", nb, keep);
|
||||
if(na!=nb) FAIL("keep-count mismatch: old=%d new=%d", na, nb);
|
||||
|
||||
/* 2. element-wise identical dst[]. This is the strong contract: because the
|
||||
* threshold is derived identically and the position-order scans are byte-
|
||||
* for-byte the same, the kept SET and its ORDER must match exactly. (This
|
||||
* is what makes #356 cleaner than #335, which was multiset-only.) */
|
||||
int first_diff=-1;
|
||||
for(int i=0;i<na;i++){ if(da[i]!=db[i]){ first_diff=i; break; } }
|
||||
if(first_diff>=0)
|
||||
FAIL("kept-set differs at index %d: old dst[%d]=%d new dst[%d]=%d",
|
||||
first_diff, first_diff, da[first_diff], first_diff, db[first_diff]);
|
||||
|
||||
/* 3. also check the partition invariant directly (catches a subtly broken
|
||||
* quickselect even if the threshold happened to come out right). */
|
||||
check_partition(label,isc,nk,keep,shape_name);
|
||||
|
||||
free(da); free(db);
|
||||
printf(" ok [nk=%d keep=%d shape=%s]\n", nk, keep, shape_name);
|
||||
}
|
||||
#undef FAIL
|
||||
|
||||
/* deterministic xorshift32 RNG (matches the test_i4_grouped.c / test_topp.c convention) */
|
||||
static uint32_t rng_state = 0x12345678u;
|
||||
static uint32_t xr(void){ rng_state ^= rng_state << 13; rng_state ^= rng_state >> 17;
|
||||
rng_state ^= rng_state << 5; return rng_state; }
|
||||
static double frand(void){ return (xr() >> 8) * (1.0 / 16777216.0); } /* [0,1) */
|
||||
|
||||
/* fill scores for a given shape. Shapes stress the threshold boundary and the
|
||||
* quickselect's median-of-three pivot differently. */
|
||||
static void fill_shape(float *isc, int nk, int shape){
|
||||
switch(shape){
|
||||
case 0: /* uniform random distinct (no ties): the clean contract case */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(frand()*1000.0); break;
|
||||
case 1: /* peaked: a few hot, long distinct tail (realistic attention shape) */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(-1.0 - frand()*4.0);
|
||||
isc[0]=3.f; if(nk>3) isc[nk/3]=1.f; if(nk>2) isc[nk/2]=0.5f; break;
|
||||
case 2: /* strictly decreasing geometric (no ties): sorted input -- worst case
|
||||
* for a naive quickselect; median-of-three must handle it */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(-0.001*(double)i); break;
|
||||
case 3: /* strictly increasing (reverse-sorted): the other quickselect worst case */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(0.001*(double)i); break;
|
||||
case 4: /* plateau ties: blocks of equal value -> boundary membership decided
|
||||
* entirely by the position scan (exercises the ==thr path) */
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)(-(double)(i/7)); break;
|
||||
case 5: /* all-equal: every value identical -> degenerate threshold, all kept
|
||||
* via the ==thr scan; quickselect must not infinite-loop or corrupt */
|
||||
for(int i=0;i<nk;i++) isc[i]=5.f; break;
|
||||
}
|
||||
}
|
||||
|
||||
int main(void){
|
||||
/* Sizes around the real index_topk=2048 boundary, plus small cases that
|
||||
* exercise the k>=n / k==1 / k==n edges. */
|
||||
int nks[] = {1, 2, 8, 64, 2049, 4097, 8193};
|
||||
int keeps[] = {1, 8, 256, 1024, 2048};
|
||||
int n_shapes = 6;
|
||||
|
||||
int cases = 0;
|
||||
for(size_t ni=0; ni<sizeof(nks)/sizeof(nks[0]); ni++){
|
||||
int nk=nks[ni];
|
||||
float *isc=malloc((size_t)nk*sizeof(float));
|
||||
for(int shape=0; shape<n_shapes; shape++){
|
||||
/* skip shapes that write out of bounds on tiny nk (fill_shape guards
|
||||
* the hot-spots with nk>n, but skip the plateau/geometric edge if nk<7) */
|
||||
fill_shape(isc,nk,shape);
|
||||
for(size_t ki=0; ki<sizeof(keeps)/sizeof(keeps[0]); ki++){
|
||||
int keep=keeps[ki];
|
||||
if(keep>nk) continue; /* keep<=nk invariant of the production code */
|
||||
if(keep<=0) continue;
|
||||
char label[40]; snprintf(label,sizeof(label),"nk[%zu]/keep[%zu]/shape[%d]",ni,ki,shape);
|
||||
const char *sn=(const char*[]){"random","peaked","decreasing","increasing","plateau","all-equal"}[shape];
|
||||
check_case(label,nk,keep,sn,isc);
|
||||
cases++;
|
||||
}
|
||||
}
|
||||
free(isc);
|
||||
}
|
||||
|
||||
/* edge: keep == nk (nothing to partition; both paths keep everything) */
|
||||
{
|
||||
int nk=100, keep=100; float isc[100];
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)frand();
|
||||
int *db=malloc(sizeof(int)*nk); int nb=keep_new(isc,nk,keep,db);
|
||||
if(nb!=nk){ fprintf(stderr," FAIL [keep==nk]: kept %d expected %d\n",nb,nk); g_nfails++; }
|
||||
else printf(" ok [keep==nk nk=%d]\n",nk);
|
||||
free(db); cases++;
|
||||
}
|
||||
/* edge: keep == 1 (threshold = the single max; quickselect must find it) */
|
||||
{
|
||||
int nk=500, keep=1; float isc[500];
|
||||
for(int i=0;i<nk;i++) isc[i]=(float)frand();
|
||||
int da[1],db[1]; int na=keep_old(isc,nk,keep,da), nb=keep_new(isc,nk,keep,db);
|
||||
if(na!=1||nb!=1||da[0]!=db[0]){
|
||||
fprintf(stderr," FAIL [keep==1]: old={%d (n=%d)} new={%d (n=%d)}\n",da[0],na,db[0],nb); g_nfails++; }
|
||||
else printf(" ok [keep==1 argmax=%d]\n",db[0]); cases++;
|
||||
}
|
||||
/* edge: all-equal scores, keep in the middle -> every kept slot is a tie;
|
||||
* the position scan must pick positions 0..keep-1 deterministically */
|
||||
{
|
||||
int nk=1000, keep=500; float isc[1000];
|
||||
for(int i=0;i<nk;i++) isc[i]=3.14f;
|
||||
int *db=malloc(sizeof(int)*nk); int nb=keep_new(isc,nk,keep,db);
|
||||
int bad=0; for(int i=0;i<nb;i++) if(db[i]!=i) bad=1;
|
||||
if(nb!=keep||bad){ fprintf(stderr," FAIL [all-equal keep=%d]: nb=%d bad=%d\n",keep,nb,bad); g_nfails++; }
|
||||
else printf(" ok [all-equal keep=%d -> positions 0..%d]\n",keep,keep-1); cases++;
|
||||
free(db);
|
||||
}
|
||||
|
||||
printf("\ntest_dsa_select: %d cases run, %d failure(s)\n", cases, g_nfails);
|
||||
if(g_nfails){ printf("test_dsa_select: FAIL\n"); return 1; }
|
||||
printf("test_dsa_select: ok\n");
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,315 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Exhaustive optimization dossier for a colibri engine run.
|
||||
|
||||
This is NOT a pass/fail test. It runs the engine with every instrumentation flag
|
||||
on (PROF, COLI_CUDA_PROFILE, CACHE_ROUTE, DISK_SPLIT, LOOKA) and prints a section-
|
||||
by-section report answering, for each subsystem:
|
||||
|
||||
WHAT is doing it — which phase/kernel/tier
|
||||
WHEN it is doing it — how much of decode wall-time it owns
|
||||
WITH WHAT — the config/weights/tier it used
|
||||
IS IT INEFFICIENT? — a verdict, with the threshold
|
||||
HOW TO IMPROVE — the concrete knob, named
|
||||
|
||||
Activation (opt-in only — NOT in `make test`):
|
||||
COLI_EFFICIENCY_MODEL=<model_dir> python tests/test_efficiency_report.py
|
||||
|
||||
Optional env:
|
||||
COLI_EFFICIENCY_CUDA=1 also exercise the CUDA dense/expert tiers
|
||||
COLI_EFFICIENCY_NGEN=N decode tokens (default 24)
|
||||
COLI_EFFICIENCY_RAM_GB=N RAM budget (default 28)
|
||||
COLI_EFFICIENCY_VRAM_GB=N CUDA expert-tier budget GB (default 4)
|
||||
COLI_EFFICIENCY_PROMPT=... prompt (default: a code-gen prompt)
|
||||
|
||||
Exit code is always 0 (it's a dossier, not a gate). Lines marked FLAG point at
|
||||
the most likely lever to move tok/s for the observed bottleneck.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from tools.efficiency import run_engine, disk_wait_share # noqa: E402
|
||||
|
||||
|
||||
# --- advisory thresholds (the "IS IT INEFFICIENT?" lines) ---
|
||||
DISK_WAIT_DOMINANT = 0.40 # >40% decode waiting on expert reads -> I/O-bound
|
||||
LOW_HIT_RATE = 0.30 # <30% cache hit -> thrashing (cap too small)
|
||||
LOW_ROUTE_AGREE = 0.80 # <80% routing overlap -> prefetch is guessing wrong
|
||||
HIGH_TAIL_RATIO = 3.0 # p99 > 3x p50 -> decode stalls (I/O hiccups / KV grow)
|
||||
LOW_MTP_ACCEPT = 0.20 # <20% MTP acceptance -> draft decoder is dead weight
|
||||
VRAM_WASTE_CALLS = 0 # experts pinned in VRAM but 0 calls served
|
||||
|
||||
|
||||
def _flag(ok): return "OK " if ok else "FLAG"
|
||||
|
||||
|
||||
def _bar(frac, width=24):
|
||||
"""A simple ASCII bar for share visualization."""
|
||||
n = max(0, min(width, round(frac * width)))
|
||||
return "#" * n + "." * (width - n)
|
||||
|
||||
|
||||
def _line(label, value, flag=None, note=""):
|
||||
tag = f" [{flag}]" if flag else ""
|
||||
print(f" {label:<22} {value}{tag} {note}" if note else f" {label:<22} {value}{tag}")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
model = os.environ.get("COLI_EFFICIENCY_MODEL")
|
||||
if not model:
|
||||
print(__doc__)
|
||||
print("\nNot activated: set COLI_EFFICIENCY_MODEL=<model_dir> to run.")
|
||||
return 0
|
||||
model = str(Path(model).resolve())
|
||||
if not Path(model).is_dir():
|
||||
print(f"ERROR: {model} is not a directory", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
ngen = int(os.environ.get("COLI_EFFICIENCY_NGEN", "24"))
|
||||
ram_gb = os.environ.get("COLI_EFFICIENCY_RAM_GB", "28")
|
||||
vram_gb = os.environ.get("COLI_EFFICIENCY_VRAM_GB", "4")
|
||||
prompt = os.environ.get(
|
||||
"COLI_EFFICIENCY_PROMPT",
|
||||
"Write a Python function that computes the factorial of a number. "
|
||||
"Include error handling and a docstring.")
|
||||
use_cuda = os.environ.get("COLI_EFFICIENCY_CUDA") == "1"
|
||||
|
||||
# Turn ON every instrumentation flag so the dossier has maximum detail.
|
||||
# These are all observability toggles (PROF/COLI_CUDA_PROFILE/CACHE_ROUTE/
|
||||
# DISK_SPLIT/LOOKA); none change the computed output.
|
||||
overlay = dict(
|
||||
NGEN=str(ngen), TEMP="0", RAM_GB=ram_gb, PROMPT=prompt,
|
||||
PROF="1", CACHE_ROUTE="1", DISK_SPLIT="1", LOOKA="1", ROUTE_AGREE="1",
|
||||
)
|
||||
if use_cuda:
|
||||
overlay.update(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1",
|
||||
COLI_CUDA_PROFILE="1", CUDA_EXPERT_GB=vram_gb)
|
||||
|
||||
print("=" * 78)
|
||||
print(f"OPTIMIZATION DOSSIER — {Path(model).name}")
|
||||
print(f" mode : {'CUDA (dense+expert tiers)' if use_cuda else 'CPU-only'} "
|
||||
f"ngen : {ngen} ram : {ram_gb} GB" +
|
||||
(f" vram : {vram_gb} GB" if use_cuda else ""))
|
||||
print("=" * 78)
|
||||
|
||||
t0 = time.time()
|
||||
t, proc = run_engine(overlay, snap=model, timeout=3600.0)
|
||||
wall = time.time() - t0
|
||||
flags = [] # collected FLAG lines for the summary
|
||||
|
||||
print(f"\n[0] RUN")
|
||||
_line("wall clock", f"{wall:.0f}s")
|
||||
_line("exit code", proc.returncode,
|
||||
None if proc.returncode == 0 else "FLAG",
|
||||
"" if proc.returncode == 0 else "non-zero exit")
|
||||
if proc.returncode != 0:
|
||||
print(" stderr tail:")
|
||||
for ln in proc.stderr.strip().splitlines()[-8:]:
|
||||
print(f" {ln}")
|
||||
return 0
|
||||
|
||||
# ---------------------------------------------------------------- [1] WHO ----
|
||||
print(f"\n[1] PROVENANCE — what is running, on what, with what config")
|
||||
if t.get("machine"):
|
||||
m = t["machine"]
|
||||
_line("CPU", m["cpu"])
|
||||
_line("cores / omp", f"{m['cores']} cores")
|
||||
_line("backend", m["backend"])
|
||||
if t.get("load"):
|
||||
ld = t["load"]
|
||||
_line("model load time", f"{ld['load_s']:.2f}s")
|
||||
_line("resident dense", f"{ld['resident_dense_mb']:.1f} MB")
|
||||
_line("layers / experts", f"{ld['layers']} layers, {ld['experts']} experts")
|
||||
_line("MTP", f"{ld['mtp_status']} (draft={ld['draft']})")
|
||||
if t.get("config_str"):
|
||||
_line("resolved config", t["config_str"])
|
||||
print(" (this is the EFFECTIVE config after auto-budgeting — not your env verbatim)")
|
||||
|
||||
# ---------------------------------------------------------------- [2] SPEED --
|
||||
print(f"\n[2] THROUGHPUT — is it fast, is the tail healthy")
|
||||
if t.get("tok_s") is not None:
|
||||
_line("tok/s", f"{t['tok_s']:.3f}")
|
||||
else:
|
||||
flags.append("throughput line missing — engine output format may have changed")
|
||||
if t.get("latency"):
|
||||
la = t["latency"]
|
||||
_line("decode forwards", f"{int(la['forwards'])}")
|
||||
_line("p50 / p90", f"{la['p50_ms']:.1f} / {la['p90_ms']:.1f} ms")
|
||||
_line("p99 / max", f"{la['p99_ms']:.1f} / {la['max_ms']:.1f} ms")
|
||||
tail_ok = la["p99_ms"] <= HIGH_TAIL_RATIO * la["p50_ms"]
|
||||
_line("tail ratio (p99/p50)", f"{la['p99_ms']/max(la['p50_ms'],1e-9):.2f}x",
|
||||
_flag(tail_ok),
|
||||
"high tail = decode stalls (I/O hiccups, KV growth, re-pin)")
|
||||
if not tail_ok:
|
||||
flags.append(f"tail latency p99={la['p99_ms']:.1f}ms >> p50={la['p50_ms']:.1f}ms "
|
||||
"(look for REPIN swaps or disk stalls)")
|
||||
|
||||
# ---------------------------------------------------------------- [3] TIME ---
|
||||
print(f"\n[3] WHERE TIME GOES — what is doing it, when (share of decode)")
|
||||
ts = t.get("time_shares")
|
||||
prof = t.get("profile")
|
||||
if ts:
|
||||
order = [("io", "expert-disk I/O", DISK_WAIT_DOMINANT, "the cache is too small / disk is slow"),
|
||||
("matmul", "expert matmul", 0.40, "compute-bound; more cores or a GPU expert tier"),
|
||||
("attention", "attention", 0.35, "context length is the cost; lower CTX"),
|
||||
("head", "lm_head", 0.10, "vocab projection; unusual to dominate"),
|
||||
("other", "other", 0.30, "scheduling / KV bookkeeping overhead")]
|
||||
for key, name, thresh, lever in order:
|
||||
f = ts[key]
|
||||
ok = f < thresh
|
||||
_line(name, f"{f:5.0%} {_bar(f)}", _flag(ok),
|
||||
"" if ok else f"->{lever}")
|
||||
if not ok:
|
||||
flags.append(f"{name} dominates ({f:.0%}) -> {lever}")
|
||||
if t.get("verdict"):
|
||||
print(f" engine verdict : {t['verdict']}")
|
||||
elif prof:
|
||||
print(" (no [PROF] time shares — set PROF=1 for phase percentages)")
|
||||
if prof:
|
||||
print(" absolute seconds :")
|
||||
for k in ("disk", "expert_matmul", "attention", "lm_head", "other"):
|
||||
_line(k, f"{prof[k]:.3f}s")
|
||||
|
||||
# attention sub-breakdown: how is attention being read
|
||||
ab = t.get("attn_breakdown")
|
||||
if ab:
|
||||
print(f"\n[3a] ATTENTION BREAKDOWN — how the attention phase is spent")
|
||||
atot = sum(ab.values()) or 1.0
|
||||
for k, label in (("proj_rope", "projection + RoPE"),
|
||||
("score_sm_value", "score-softmax-value"),
|
||||
("out_proj", "output projection")):
|
||||
_line(label, f"{ab[k]:.3f}s ({ab[k]/atot:.0%} of attn)")
|
||||
|
||||
# ---------------------------------------------------------------- [4] CACHE --
|
||||
print(f"\n[4] EXPERT CACHE — is the cache efficient")
|
||||
hit = t.get("hit_pct")
|
||||
if hit is not None:
|
||||
ok = hit >= LOW_HIT_RATE * 100
|
||||
_line("hit rate", f"{hit:.1f}%", _flag(ok),
|
||||
"" if ok else "<30% = thrashing; raise RAM_GB or cap")
|
||||
if not ok:
|
||||
flags.append(f"cache hit {hit:.1f}% is low -> raise RAM_GB (or cap), add PIN_GB")
|
||||
el = t.get("experts_loaded")
|
||||
if el:
|
||||
per_tok = el["per_tok"]
|
||||
_line("experts loaded/token", f"{per_tok:.1f}")
|
||||
_line(" per-layer", f"{el['per_layer']:.2f} across {el['n_sparse_layers']} sparse layers")
|
||||
_line(" baseline", f"topk={el['baseline_topk']} active experts/token")
|
||||
base_topk = el["baseline_topk"]
|
||||
if base_topk > 0 and per_tok > 2 * base_topk:
|
||||
flags.append(f"loading {per_tok:.0f} experts/token vs topk={base_topk} "
|
||||
"-> redundant I/O; cache is re-fetching evicted experts")
|
||||
|
||||
# ---------------------------------------------------------------- [5] DISK ---
|
||||
print(f"\n[5] DISK I/O — is I/O the bottleneck, and where")
|
||||
eio = t.get("expert_io")
|
||||
if eio:
|
||||
_line("total fetched", f"{eio['gb_fetched']:.3f} GB")
|
||||
_line("per token", f"{eio['mb_per_tok']:.1f} MB/token")
|
||||
_line("disk throughput", f"{eio['gb_per_s']:.2f} GB/s over the run")
|
||||
_line("read service", f"{eio['read_service_s']:.2f}s (on I/O threads)")
|
||||
_line("felt wait", f"{eio['felt_wait_s']:.2f}s (stall compute felt)")
|
||||
if eio["felt_wait_s"] > eio["read_service_s"] * 0.5 and eio["read_service_s"] > 0:
|
||||
flags.append("felt wait is a large fraction of read service -> PIPE=1 may not be "
|
||||
"overlapping fully, or DIRECT=1 on NVMe")
|
||||
ds = t.get("disk_split")
|
||||
if ds:
|
||||
print(f"\n[5a] DISK-LOAD SPLIT — which decode phase reads the bytes")
|
||||
_line("draft phase", f"{ds['draft']} loads")
|
||||
_line("absorb phase", f"{ds['absorb']} loads")
|
||||
_line("verify/main", f"{ds['verify_main']} loads")
|
||||
_line("MTP-layer bytes", f"{ds['mtp_loads']} loads, {ds['mtp_gb']:.2f} GB")
|
||||
_line("main-layer bytes", f"{ds['main_loads']} loads, {ds['main_gb']:.2f} GB")
|
||||
if ds.get("mtp_bytes_pct") is not None:
|
||||
_line("MTP share of bytes", f"{ds['mtp_bytes_pct']:.1f}%")
|
||||
share = disk_wait_share(t)
|
||||
if share is not None:
|
||||
ok = share < DISK_WAIT_DOMINANT
|
||||
_line("disk-wait share", f"{share:.0%}", _flag(ok),
|
||||
"" if ok else "I/O-bound (see levers in [3])")
|
||||
|
||||
# ---------------------------------------------------------------- [6] ROUTE --
|
||||
print(f"\n[6] ROUTING QUALITY — is the router / prefetch accurate")
|
||||
ra = t.get("route_agree")
|
||||
if ra:
|
||||
ok = ra["agree_pct"] >= LOW_ROUTE_AGREE * 100
|
||||
_line("route_agree", f"{ra['agree_pct']:.1f}% overlap with true top-K",
|
||||
_flag(ok),
|
||||
"" if ok else "prefetch is guessing wrong; CACHE_ROUTE params may need tuning")
|
||||
_line("route_kl", f"{ra['kl']:.4f} mean KL (lower = closer to true routing)")
|
||||
if not ok:
|
||||
flags.append(f"route_agree {ra['agree_pct']:.1f}% low -> tune ROUTE_J/M/P, "
|
||||
"or prefetch is hurting more than helping")
|
||||
sw = t.get("swap")
|
||||
if sw:
|
||||
_line("cache swaps", f"{sw['swaps']}/{sw['slots']} ({sw['pct']:.1f}%)",
|
||||
None, "high swap = churn between turns")
|
||||
la = t.get("lookahead")
|
||||
if la:
|
||||
print(f"\n[6a] ROUTING PREDICTABILITY — recall of true experts in predicted top-8")
|
||||
print(" (which predictor should drive prefetch? highest recall wins)")
|
||||
for row in la:
|
||||
_line(row["predictor"][:34], f"{row['pct']:5.1f}% ({row['hit']}/{row['tot']})")
|
||||
|
||||
# ---------------------------------------------------------------- [7] SPEC ---
|
||||
print(f"\n[7] SPECULATION — is the draft decoder pulling weight")
|
||||
sp = t.get("speculation")
|
||||
if sp:
|
||||
_line("tokens/forward", f"{sp['tok_per_fw']:.2f} (>1.0 means speculation helps)")
|
||||
_line("forwards/tokens", f"{sp['forwards']} forwards for {sp['tokens']} tokens")
|
||||
# Speculation helps only if acceptance is high enough that tok/forward > 1.
|
||||
# tok_per_fw already == 1.0 when nothing verifies, so judge by acceptance.
|
||||
ok = sp["mtp_accept_pct"] >= LOW_MTP_ACCEPT * 100
|
||||
_line("MTP acceptance", f"{sp['mtp_accept_pct']:.0f}%", _flag(ok),
|
||||
"" if ok else "<20% -> drafts rarely verify; DRAFT=0 may be faster")
|
||||
if not ok:
|
||||
flags.append(f"MTP acceptance {sp['mtp_accept_pct']:.0f}% low -> "
|
||||
"drafts cost more I/O than they save; try DRAFT=0")
|
||||
|
||||
# ---------------------------------------------------------------- [8] GPU ----
|
||||
print(f"\n[8] GPU TIERS — is the GPU actually used")
|
||||
cuda = t.get("cuda") or {}
|
||||
if not cuda.get("enabled"):
|
||||
print(" (CUDA not enabled — CPU-only run)")
|
||||
else:
|
||||
if cuda.get("resident_tensors") is not None:
|
||||
_line("resident dense tensors", f"{cuda['resident_tensors']} tensors, "
|
||||
f"{cuda['resident_gb']:.2f} GB")
|
||||
if cuda.get("expert_count") is not None:
|
||||
waste = cuda["calls_served"] <= VRAM_WASTE_CALLS
|
||||
_line("expert tier", f"{cuda['expert_count']} experts pinned "
|
||||
f"({cuda['expert_gb']:.2f} GB)", _flag(not waste))
|
||||
_line(" calls served", f"{cuda['calls_served']} from VRAM",
|
||||
_flag(not waste),
|
||||
"" if not waste else "WASTE: pinned but never routed -> lower CUDA_EXPERT_GB")
|
||||
if waste:
|
||||
flags.append("VRAM expert tier has 0 calls served -> experts pinned but unused; "
|
||||
"PIN stats may not match this workload")
|
||||
if cuda.get("groups"):
|
||||
g = cuda["groups"]
|
||||
_line("expert groups", f"{g['calls']} calls, {g['experts']} experts, "
|
||||
f"{g['rows']} rows ({g['experts_per_call']:.1f} experts/call)")
|
||||
if cuda.get("groups_timing"):
|
||||
gt = cuda["groups_timing"]
|
||||
_line("GPU timing", f"H2D {gt['h2d_ms']:.1f} ms | kernel {gt['kernel_ms']:.1f} ms | "
|
||||
f"D2H {gt['d2h_ms']:.1f} ms")
|
||||
if gt["h2d_ms"] + gt["d2h_ms"] > gt["kernel_ms"]:
|
||||
flags.append("CUDA H2D+D2H > kernel time -> transfer-bound; "
|
||||
"consider larger expert tier to keep weights resident")
|
||||
|
||||
# ---------------------------------------------------------------- summary ----
|
||||
print("\n" + "=" * 78)
|
||||
if flags:
|
||||
print(f" {len(flags)} FLAG(s) — the most likely levers to move tok/s:")
|
||||
for f in flags:
|
||||
print(f" - {f}")
|
||||
else:
|
||||
print(" no flags — every measured subsystem is within advisory thresholds.")
|
||||
print("=" * 78)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -18,7 +18,7 @@
|
||||
* index, a wrong group boundary or a swapped nibble cannot hide under it —
|
||||
* those are O(1) relative errors, not O(1e-6). */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../glm.c"
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
static uint32_t rng_state=0xC0FFEEu;
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
* (sign-trick kernels must treat |−128| as 128 unsigned, not saturate to 127),
|
||||
* and random data at qrow_i8's contract (|x| <= 127, w full int8 range). */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../glm.c"
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
static uint32_t rng_state=0x12345678u;
|
||||
|
||||
@@ -0,0 +1,257 @@
|
||||
"""Inefficiency / regression tests for the colibri engine (tiny model, asserted).
|
||||
|
||||
These run against the bundled glm_tiny model (~0.6 MB resident, ~0.1s/run) and
|
||||
gate CI: a regression here means something broke. They run on a plain CPU-only
|
||||
`glm.exe` build; the CUDA_* tests auto-skip if the engine wasn't built with
|
||||
CUDA_DLL=1 (see tests/README_efficiency.md for the build command).
|
||||
|
||||
The signals under test, and the inefficiency each catches:
|
||||
- tok/s floor : a throughput regression (broken build / bad config)
|
||||
- profile phases sum : telemetry accounting bug (other balloons)
|
||||
- disk-wait not dominant: a tiny resident model should never be I/O-bound
|
||||
- CPU determinism : greedy decode is reproducible (no stray RNG/threading)
|
||||
- CUDA init path : COLI_CUDA=1 initializes and does not silently exit 2
|
||||
- CUDA dense uses VRAM : CUDA_DENSE=1 actually uploads tensors (no silent CPU fallback)
|
||||
- CPU vs CUDA TF-match : identical weights+inputs → identical argmax (kernel bug guard)
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from tools.efficiency import (
|
||||
parse_run, run_engine, disk_wait_share, tf_agreement,
|
||||
TINY_TOK_S_FLOOR, MAX_DISK_WAIT_SHARE, MIN_CPU_CUDA_AGREEMENT,
|
||||
)
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
C_DIR = HERE.parent
|
||||
ENGINE = C_DIR / "glm.exe"
|
||||
TINY = C_DIR / "glm_tiny"
|
||||
|
||||
|
||||
def _engine_present() -> bool:
|
||||
"""True iff BOTH the built engine AND the tiny fixture are available.
|
||||
|
||||
These tests need glm.exe (a build artifact) AND glm_tiny/ (a generated
|
||||
fixture, gitignored). CI runs `make check` = "dependency-free tests, no
|
||||
model downloads" (workflow .github/workflows/check.yml, by design #140), so
|
||||
neither is present there and these tests must SKIP rather than fail. They
|
||||
run locally after `make glm.exe` (glm_tiny ships alongside the source, or
|
||||
is regenerated by tools/make_glm_oracle.py).
|
||||
"""
|
||||
return ENGINE.exists() and (TINY / "config.json").exists()
|
||||
|
||||
|
||||
def _skip_reason() -> str:
|
||||
"""Name exactly which prerequisite is missing, so the skip is actionable."""
|
||||
if not ENGINE.exists():
|
||||
return "glm.exe not built (run: make glm.exe)"
|
||||
if not (TINY / "config.json").exists():
|
||||
return "glm_tiny fixture absent (gitignored; ship it locally or run tools/make_glm_oracle.py)"
|
||||
return ""
|
||||
|
||||
|
||||
def _cuda_available() -> bool:
|
||||
"""True iff the engine binary has the CUDA loader compiled in AND the DLL is present.
|
||||
|
||||
The host is built with -DCOLI_CUDA only when CUDA_DLL=1 (Makefile). A binary
|
||||
built without it embeds the string "this binary is CPU-only; rebuild" and
|
||||
exits 2 on any CUDA env var — so we detect the CPU-only build by scanning
|
||||
the binary for that marker (avoids a slow ldd/strings on every import; we
|
||||
only read enough to find it). On Windows the DLL is also required
|
||||
(backend_loader.c loads it at runtime); on Linux it's direct-linked.
|
||||
"""
|
||||
if not _engine_present():
|
||||
return False
|
||||
try:
|
||||
# Read once; the marker is near the read-only string table. 256 KB is
|
||||
# plenty for this string and avoids loading a 1 MB binary into memory.
|
||||
with open(ENGINE, "rb") as f:
|
||||
blob = f.read(2 * 1024 * 1024)
|
||||
if b"this binary is CPU-only" in blob:
|
||||
return False # CPU-only build: COLI_CUDA=1 would exit 2.
|
||||
except OSError:
|
||||
pass
|
||||
# Windows: DLL required at runtime. Linux: direct-linked (no DLL).
|
||||
if os.name == "nt":
|
||||
return (C_DIR / "coli_cuda.dll").exists()
|
||||
import subprocess
|
||||
try:
|
||||
out = subprocess.run(["ldd", str(ENGINE)], capture_output=True, text=True)
|
||||
return "libcudart" in out.stdout
|
||||
except (FileNotFoundError, OSError):
|
||||
return False
|
||||
|
||||
|
||||
@unittest.skipUnless(_engine_present(), _skip_reason() or "glm.exe + glm_tiny required")
|
||||
class TinyEfficiencyTest(unittest.TestCase):
|
||||
"""Asserted regression tests on the resident tiny model. Gates CI."""
|
||||
|
||||
def _run(self, **overlay):
|
||||
return run_engine(overlay, engine=str(ENGINE), snap=str(TINY))[0]
|
||||
|
||||
# -- telemetry contract ---------------------------------------------------
|
||||
|
||||
def test_telemetry_parses(self):
|
||||
"""A REPLAY run must emit the throughput + PROFILE lines the suite keys on.
|
||||
|
||||
If this fails, either the engine changed its output format (update the
|
||||
parsers in tools/efficiency.py) or the run crashed early."""
|
||||
t = self._run(REPLAY="1", TEMP="0", NGEN="4")
|
||||
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
|
||||
self.assertIn("tok_s", t["parsed"], f"missing tok/s line:\n{t['stderr']}")
|
||||
self.assertIn("profile", t["parsed"], f"missing PROFILE line:\n{t['stderr']}")
|
||||
self.assertIn("hit_pct", t["parsed"], f"missing expert hit line:\n{t['stderr']}")
|
||||
self.assertIsNotNone(t["tok_s"])
|
||||
self.assertIsNotNone(t["profile"])
|
||||
|
||||
# -- throughput floor -----------------------------------------------------
|
||||
|
||||
def test_tiny_tok_s_floor(self):
|
||||
"""Tiny decode must beat TINY_TOK_S_FLOOR.
|
||||
|
||||
The tiny model is fully resident and runs ~200 tok/s; the 20 tok/s
|
||||
default floor is a 10x margin that catches broken builds or a pathological
|
||||
config cascade (the cap=1 trap from ISSUE_new_model_resource_regression.md)
|
||||
without flapping on machine noise."""
|
||||
t = self._run(REPLAY="1", TEMP="0", NGEN="8")
|
||||
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
|
||||
self.assertGreaterEqual(
|
||||
t["tok_s"], TINY_TOK_S_FLOOR,
|
||||
f"tok/s {t['tok_s']:.1f} below floor {TINY_TOK_S_FLOOR} "
|
||||
f"(regression, or a config cascade starving the cache)",
|
||||
)
|
||||
|
||||
# -- accounting sanity ----------------------------------------------------
|
||||
|
||||
def test_profile_phases_present_and_nonneg(self):
|
||||
"""Every PROFILE phase must be present and non-negative.
|
||||
|
||||
`other` can go slightly negative from timer overhead (the engine allows
|
||||
it), but a large negative means the timers are double-counting."""
|
||||
t = self._run(REPLAY="1", TEMP="0", NGEN="4")
|
||||
p = t["profile"]
|
||||
for phase in ("disk", "expert_matmul", "attention", "lm_head"):
|
||||
self.assertGreaterEqual(p[phase], -0.01, f"{phase} went negative: {p}")
|
||||
# 'other' is a residual; allow a small negative from timer overlap.
|
||||
self.assertGreaterEqual(p["other"], -0.05, f"other too negative (double-count): {p}")
|
||||
|
||||
# -- disk-wait not dominant on a resident model ---------------------------
|
||||
|
||||
def test_disk_wait_not_dominant(self):
|
||||
"""A fully-resident tiny model must NOT be I/O-bound.
|
||||
|
||||
Everything fits in RAM; the expert-disk wait share should be ~0. If it
|
||||
exceeds MAX_DISK_WAIT_SHARE, the cache/I/O path regressed — on a real
|
||||
model this same regression would make decode I/O-bound (the exact
|
||||
failure mode the [PROF] verdict flags)."""
|
||||
t = self._run(REPLAY="1", TEMP="0", NGEN="8", PROF="1")
|
||||
self.assertIn("time_shares", t["parsed"], f"missing [PROF] time shares:\n{t['stderr']}")
|
||||
share = disk_wait_share(t)
|
||||
self.assertIsNotNone(share)
|
||||
self.assertLess(
|
||||
share, MAX_DISK_WAIT_SHARE,
|
||||
f"expert-I/O share {share:.0%} on a resident model — I/O path regressed",
|
||||
)
|
||||
|
||||
# -- determinism ----------------------------------------------------------
|
||||
|
||||
def test_cpu_vs_cpu_determinism(self):
|
||||
"""Two greedy REPLAY runs with the same seed produce identical telemetry.
|
||||
|
||||
TEMP=0 = greedy (no sampling), so tok/s and hit-rate must be reproducible.
|
||||
A drift here means non-determinism crept into the decode path (stray
|
||||
threading, uninitialized state) — which on a real model would make A/B
|
||||
comparisons meaningless."""
|
||||
a = self._run(REPLAY="1", TEMP="0", NGEN="8", SEED="1")
|
||||
b = self._run(REPLAY="1", TEMP="0", NGEN="8", SEED="1")
|
||||
self.assertEqual(a["returncode"], 0)
|
||||
self.assertEqual(b["returncode"], 0)
|
||||
self.assertEqual(a["hit_pct"], b["hit_pct"], "greedy hit-rate drifted between runs")
|
||||
# tok/s within 25% — exact equality is too strict across scheduler noise.
|
||||
self.assertLess(abs(a["tok_s"] - b["tok_s"]) / max(a["tok_s"], b["tok_s"]), 0.25)
|
||||
|
||||
|
||||
@unittest.skipUnless(_cuda_available(),
|
||||
_skip_reason() or "CUDA build not present (run: make clean && make glm.exe CUDA_DLL=1 && make cuda-dll)")
|
||||
class TinyCudaEfficiencyTest(unittest.TestCase):
|
||||
"""CUDA-path regression tests on the tiny model. Skip unless CUDA built.
|
||||
|
||||
glm_tiny is small and fully resident, so CUDA here is fast and exercises the
|
||||
real GPU code path (init, dense upload, kernel correctness) without the
|
||||
long load time or memory pressure of the full model. These guard the
|
||||
silent-failure modes that are otherwise invisible:
|
||||
- COLI_CUDA=1 silently falling back to CPU (loader/DLL missing)
|
||||
- CUDA_DENSE=1 uploading nothing
|
||||
- a CUDA kernel producing different argmax than CPU on identical inputs
|
||||
"""
|
||||
|
||||
def _run(self, **overlay):
|
||||
return run_engine(overlay, engine=str(ENGINE), snap=str(TINY))[0]
|
||||
|
||||
def test_cuda_init_path(self):
|
||||
"""COLI_CUDA=1 must initialize the device and NOT exit 2.
|
||||
|
||||
Exit 2 is the engine's "requested backend is unavailable" path
|
||||
(glm.c: g_cuda_enabled check). A clean init prints the [CUDA] device
|
||||
banner to stderr. If this fails, the DLL is broken or the loader can't
|
||||
resolve symbols (ABI drift between backend_cuda.h and the dll)."""
|
||||
t = self._run(COLI_CUDA="1", COLI_GPU="0", REPLAY="1", TEMP="0", NGEN="4")
|
||||
self.assertNotEqual(t["returncode"], 2,
|
||||
f"engine refused CUDA backend:\n{t['stderr']}")
|
||||
self.assertTrue(t["cuda"]["enabled"],
|
||||
f"no [CUDA] device banner on stderr:\n{t['stderr']}")
|
||||
|
||||
def test_cuda_dense_uses_vram(self):
|
||||
"""CUDA_DENSE=1 must actually upload dense tensors to VRAM.
|
||||
|
||||
The minimal GPU-exercising config (per backend_loader.c analysis):
|
||||
COLI_CUDA=1 + CUDA_DENSE=1. Without CUDA_DENSE the dense path stays on
|
||||
CPU and [CUDA] resident set reports 0 tensors — a silent no-op. This
|
||||
catches that regression: after the run, resident_tensors > 0."""
|
||||
t = self._run(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1",
|
||||
REPLAY="1", TEMP="0", NGEN="4")
|
||||
self.assertEqual(t["returncode"], 0, f"engine exited non-zero:\n{t['stderr']}")
|
||||
rt = t["cuda"]["resident_tensors"]
|
||||
self.assertIsNotNone(rt, f"no [CUDA] resident set line:\n{t['stderr']}")
|
||||
self.assertGreater(rt, 0,
|
||||
f"CUDA_DENSE=1 but {rt} tensors resident — silent CPU fallback")
|
||||
|
||||
def test_cpu_vs_cuda_tf_match(self):
|
||||
"""CPU and CUDA teacher-forcing must agree on most positions (DIRECTLY).
|
||||
|
||||
Both paths prefill the SAME oracle sequence on the SAME weights, so their
|
||||
argmaxes should match position-for-position — but not exactly: the two
|
||||
backends accumulate dot-products in different orders (x86 SIMD vs CUDA
|
||||
kernel), so a few near-tied logits flip. That divergence is expected
|
||||
numeric behavior, not a kernel bug. A *catastrophic* kernel regression
|
||||
(wrong GEMM, wrong scale, wrong fmt) would collapse agreement toward
|
||||
random (~1/vocab = ~4%); the MIN_CPU_CUDA_AGREEMENT floor (default 70%)
|
||||
catches that while tolerating harmless drift.
|
||||
|
||||
We compare CPU-vs-CUDA *directly* (not via the oracle match-counts),
|
||||
because both backends differ from the oracle at different positions and
|
||||
the summary line can't tell "CPU≠CUDA" from "CPU≠oracle"."""
|
||||
import json
|
||||
ref = json.loads((C_DIR / "ref_glm.json").read_text())
|
||||
oracle = ref["tf_pred"]
|
||||
|
||||
cpu = self._run(TF="1", TEMP="0")
|
||||
cuda = self._run(COLI_CUDA="1", COLI_GPU="0", CUDA_DENSE="1", TF="1", TEMP="0")
|
||||
self.assertEqual(cpu["returncode"], 0, f"CPU TF run failed:\n{cpu['stderr']}")
|
||||
self.assertEqual(cuda["returncode"], 0, f"CUDA TF run failed:\n{cuda['stderr']}")
|
||||
self.assertIn("tf_mismatches", cpu["parsed"], "CPU run missing per-position mismatches")
|
||||
self.assertIn("tf_mismatches", cuda["parsed"], "CUDA run missing per-position mismatches")
|
||||
|
||||
agree, diff = tf_agreement(cpu, cuda, oracle)
|
||||
self.assertGreaterEqual(
|
||||
agree, MIN_CPU_CUDA_AGREEMENT,
|
||||
f"CPU-vs-CUDA argmax agreement {agree:.0%} below floor "
|
||||
f"{MIN_CPU_CUDA_AGREEMENT:.0%} — CUDA kernel likely regressed. "
|
||||
f"Differing positions: {diff[:10]}{'...' if len(diff)>10 else ''}",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -4,7 +4,7 @@
|
||||
* arrays twice on the second call -> allocator abort. No model file needed:
|
||||
* the CPU path of kv_alloc only reads c->n_layers/kv_lora/qk_rope. */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../glm.c"
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
int main(void){
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
/* Regression for non-finite-logit poisoning of sampling.
|
||||
*
|
||||
* A single NaN or +Inf in the logits (a bad streamed expert tile, or an fp
|
||||
* overflow in the matmul at a low-RAM eviction boundary) used to make softmax
|
||||
* produce an all-NaN g_pbuf; dist_sample then never satisfied cum>=u and fell
|
||||
* through to return token 0 — so the engine silently emitted an unbroken run of
|
||||
* token 0 with no error, on the DEFAULT serve path (TEMP>0, 0<NUCLEUS<1).
|
||||
*
|
||||
* Fix under test: argmax_v() skips NaN (picks the max finite/+Inf entry instead
|
||||
* of being pinned to index 0), and dist_build() detects a non-finite softmax sum
|
||||
* and collapses to a one-hot over the finite argmax (warning once) rather than
|
||||
* dividing every entry into NaN. Degrade + diagnose, never silently corrupt.
|
||||
*
|
||||
* No model file needed: exercises argmax_v / dist_build / dist_sample directly. */
|
||||
#include <assert.h>
|
||||
#include <math.h>
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
static int approx1(double x){ return x > 0.999 && x < 1.001; }
|
||||
|
||||
int main(void){
|
||||
/* --- argmax_v must skip NaN (greedy decode + speculative-verify paths) --- */
|
||||
{ float lo[8]={NAN,1.f,5.f,2.f,NAN,-3.f,4.f,0.f};
|
||||
assert(argmax_v(lo,8)==2 && "pick max finite (idx2=5.0), not NaN-pinned idx0"); }
|
||||
{ float lo[8]={3.f,INFINITY,1.f,2.f,0.f,-1.f,2.5f,1.5f};
|
||||
assert(argmax_v(lo,8)==1 && "pick the +Inf position"); }
|
||||
{ float lo[8]; for(int i=0;i<8;i++) lo[i]=NAN;
|
||||
assert(argmax_v(lo,8)==0 && "all-NaN: no crash, defined fallback"); }
|
||||
|
||||
g_temp=0.7f; g_nuc=0.9f; /* the default serve/chat sampling path */
|
||||
|
||||
/* --- dist_build: a NaN logit must yield a finite one-hot, not all-NaN --- */
|
||||
{ float lo[8]={0.5f,1.f,NAN,8.f,0.2f,-1.f,0.f,0.3f}; /* max finite = idx3 (8.0) */
|
||||
dist_build(lo,8);
|
||||
double sum=0; int nan=0;
|
||||
for(int i=0;i<8;i++){ if(!(g_pbuf[i]==g_pbuf[i])) nan=1; sum+=g_pbuf[i]; }
|
||||
assert(!nan && "g_pbuf must be finite after a NaN logit");
|
||||
assert(approx1(sum) && "g_pbuf must normalize to 1");
|
||||
assert(approx1(g_pbuf[3]) && "mass must land on the max finite logit (idx3)");
|
||||
assert(dist_sample(8,-1)==3 && "sampler emits the finite argmax, not token 0"); }
|
||||
|
||||
/* --- NaN at index 0: poisons the max scan itself (the old mx=lo[0] seed),
|
||||
* the failure mode that starts before the sum (review note on #369) --- */
|
||||
{ float lo[8]={NAN,1.f,0.5f,6.f,0.2f,-1.f,0.f,0.3f}; /* max finite = idx3 (6.0) */
|
||||
dist_build(lo,8);
|
||||
double sum=0; int nan=0;
|
||||
for(int i=0;i<8;i++){ if(!(g_pbuf[i]==g_pbuf[i])) nan=1; sum+=g_pbuf[i]; }
|
||||
assert(!nan && "NaN at lo[0] must not poison via the mx seed");
|
||||
assert(approx1(sum) && "still normalizes to 1");
|
||||
assert(dist_sample(8,-1)==3 && "emits the max finite logit, not token 0"); }
|
||||
|
||||
/* --- all-NaN logits: worst case — must stay finite, no crash --- */
|
||||
{ float lo[8]; for(int i=0;i<8;i++) lo[i]=NAN;
|
||||
dist_build(lo,8);
|
||||
for(int i=0;i<8;i++) assert(g_pbuf[i]==g_pbuf[i] && "all-NaN: g_pbuf stays finite"); }
|
||||
|
||||
/* --- regression: clean logits still produce a valid distribution --- */
|
||||
{ float lo[8]={0.1f,0.2f,3.0f,0.4f,0.5f,0.6f,0.7f,0.8f}; /* peak = idx2 */
|
||||
dist_build(lo,8);
|
||||
double sum=0; int nan=0;
|
||||
for(int i=0;i<8;i++){ if(!(g_pbuf[i]==g_pbuf[i])) nan=1; sum+=g_pbuf[i]; }
|
||||
assert(!nan && "clean softmax stays finite");
|
||||
assert(approx1(sum) && "clean softmax must sum to 1");
|
||||
assert(g_pbuf[2]>=g_pbuf[0] && "peak token keeps the most mass"); }
|
||||
|
||||
printf("OK test_logit_nan: argmax_v NaN-skip + dist_build finite-collapse\n");
|
||||
return 0;
|
||||
}
|
||||
@@ -35,7 +35,7 @@ class MakefilePlatformTests(unittest.TestCase):
|
||||
env["PATH"] = ""
|
||||
|
||||
result = subprocess.run(
|
||||
[MAKE, "--no-print-directory", "-B", "-n", "glm"],
|
||||
[MAKE, "--no-print-directory", "-B", "-n", "colibri"],
|
||||
cwd=C_DIR,
|
||||
env=env,
|
||||
text=True,
|
||||
@@ -43,7 +43,7 @@ class MakefilePlatformTests(unittest.TestCase):
|
||||
check=True,
|
||||
)
|
||||
|
||||
self.assertIn("-o glm.exe", result.stdout)
|
||||
self.assertIn("-o colibri.exe", result.stdout)
|
||||
self.assertIn("-fopenmp", result.stdout)
|
||||
self.assertIn("-static", result.stdout)
|
||||
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
"""End-to-end tool-calling test for the OpenAI gateway (#401).
|
||||
|
||||
Unlike the unit tests in test_openai_server.py (which call parse_tool_calls /
|
||||
render_chat directly), this suite runs openai_server.py as a real subprocess
|
||||
against a mock engine that speaks the actual SERVE wire protocol
|
||||
(READY / SUBMIT / DATA / DONE), then talks to it over real HTTP. It pins down
|
||||
the full path a coding client exercises: tool declaration rendering, marker
|
||||
suppression in streamed deltas (across chunk boundaries), tool_calls in both
|
||||
response shapes, and the <|observation|><tool_response> round trip.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
SERVER = Path(__file__).resolve().parent.parent / "openai_server.py"
|
||||
MODEL_ID = "glm-5.2-colibri"
|
||||
|
||||
# Mock engine: replies are keyed on the prompt so one process covers every case.
|
||||
# Prompts received are appended to MOCK_LOG for assertions on the rendering.
|
||||
MOCK_ENGINE = r'''#!/usr/bin/env python3
|
||||
import sys, os
|
||||
out, inp = sys.stdout.buffer, sys.stdin.buffer
|
||||
out.write(b"\x01\x01READY\x01\x01\n" + b"STAT 0 0 0 0 0\n"); out.flush()
|
||||
|
||||
def reply(rid, text, chunks=1):
|
||||
data = text.encode("utf-8")
|
||||
n = max(1, len(data) // chunks)
|
||||
for i in range(0, len(data), n):
|
||||
part = data[i:i+n]
|
||||
out.write(("DATA %s %d\n" % (rid, len(part))).encode() + part + b"\n"); out.flush()
|
||||
out.write(("DONE %s STAT %d 1.0 50.0 10.0 42 0\n" % (rid, len(text.split()))).encode())
|
||||
out.flush()
|
||||
|
||||
while True:
|
||||
line = inp.readline()
|
||||
if not line: break
|
||||
f = line.decode().strip().split()
|
||||
if not f or f[0] != "SUBMIT": continue
|
||||
rid, plen = f[1], int(f[3])
|
||||
prompt = inp.read(plen).decode("utf-8", "replace"); inp.read(1)
|
||||
with open(os.environ["MOCK_LOG"], "a") as log:
|
||||
log.write(prompt + "\n\x00\n")
|
||||
if "<tool_response>" in prompt:
|
||||
reply(rid, "25 degrees and sunny in Rome.")
|
||||
elif "weather in Rome" in prompt:
|
||||
reply(rid, "<tool_call>get_weather<arg_key>city</arg_key>"
|
||||
"<arg_value>Rome</arg_value></tool_call>")
|
||||
elif "weather in Milan" in prompt:
|
||||
# split across many tiny DATA chunks: streamed marker suppression must
|
||||
# hold even when a marker straddles a chunk boundary
|
||||
reply(rid, "Checking. <tool_call>get_weather<arg_key>city</arg_key>"
|
||||
"<arg_value>Milan</arg_value></tool_call>", chunks=20)
|
||||
else:
|
||||
reply(rid, "Hello from the mock engine.")
|
||||
'''
|
||||
|
||||
TOOLS = [{"type": "function", "function": {
|
||||
"name": "get_weather",
|
||||
"description": "Current weather for a city",
|
||||
"parameters": {"type": "object",
|
||||
"properties": {"city": {"type": "string"}},
|
||||
"required": ["city"]}}}]
|
||||
|
||||
|
||||
@unittest.skipUnless(os.name == "posix",
|
||||
"the mock engine is a shebang script the gateway execs directly; "
|
||||
"Windows CreateProcess cannot run it. The gateway logic under test "
|
||||
"is platform-independent and covered by the POSIX CI jobs.")
|
||||
class ToolCallingE2E(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.tmp = tempfile.TemporaryDirectory()
|
||||
mock = Path(cls.tmp.name) / "mock_engine.py"
|
||||
mock.write_text(MOCK_ENGINE)
|
||||
mock.chmod(0o755)
|
||||
cls.mock_log = Path(cls.tmp.name) / "prompts.log"
|
||||
cls.mock_log.touch()
|
||||
with socket.socket() as probe: # free port, then hand it to the server
|
||||
probe.bind(("127.0.0.1", 0))
|
||||
cls.port = probe.getsockname()[1]
|
||||
env = dict(os.environ, MOCK_LOG=str(cls.mock_log))
|
||||
cls.server = subprocess.Popen(
|
||||
[sys.executable, str(SERVER), "--model", cls.tmp.name,
|
||||
"--engine", str(mock), "--port", str(cls.port)],
|
||||
env=env, stderr=subprocess.DEVNULL)
|
||||
cls.base = f"http://127.0.0.1:{cls.port}/v1"
|
||||
for _ in range(100):
|
||||
try:
|
||||
urllib.request.urlopen(cls.base + "/models", timeout=2)
|
||||
return
|
||||
except OSError:
|
||||
if cls.server.poll() is not None:
|
||||
raise RuntimeError("gateway exited during startup")
|
||||
import time
|
||||
time.sleep(0.1)
|
||||
raise RuntimeError("gateway did not come up")
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.terminate()
|
||||
cls.server.wait(timeout=5)
|
||||
cls.tmp.cleanup()
|
||||
|
||||
def post(self, body, stream=False):
|
||||
req = urllib.request.Request(
|
||||
self.base + "/chat/completions", json.dumps(body).encode(),
|
||||
{"Content-Type": "application/json"})
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
if not stream:
|
||||
return json.loads(resp.read())
|
||||
events = []
|
||||
for raw in resp:
|
||||
line = raw.decode().strip()
|
||||
if line.startswith("data: ") and line != "data: [DONE]":
|
||||
events.append(json.loads(line[6:]))
|
||||
return events
|
||||
|
||||
def test_tool_call_non_stream(self):
|
||||
r = self.post({"model": MODEL_ID, "tools": TOOLS,
|
||||
"messages": [{"role": "user",
|
||||
"content": "What is the weather in Rome?"}]})
|
||||
choice = r["choices"][0]
|
||||
self.assertEqual(choice["finish_reason"], "tool_calls")
|
||||
calls = choice["message"]["tool_calls"]
|
||||
self.assertEqual(len(calls), 1)
|
||||
self.assertEqual(calls[0]["function"]["name"], "get_weather")
|
||||
self.assertEqual(json.loads(calls[0]["function"]["arguments"]), {"city": "Rome"})
|
||||
self.assertNotIn("<tool_call>", choice["message"].get("content") or "")
|
||||
|
||||
def test_tool_call_streamed_markers_suppressed(self):
|
||||
events = self.post({"model": MODEL_ID, "tools": TOOLS, "stream": True,
|
||||
"messages": [{"role": "user",
|
||||
"content": "What is the weather in Milan?"}]},
|
||||
stream=True)
|
||||
deltas = [e["choices"][0]["delta"] for e in events if e["choices"]]
|
||||
text = "".join(d.get("content") or "" for d in deltas)
|
||||
self.assertNotIn("<tool_call>", text)
|
||||
self.assertNotIn("<arg_key>", text)
|
||||
calls = [d["tool_calls"] for d in deltas if d.get("tool_calls")]
|
||||
self.assertEqual(len(calls), 1)
|
||||
self.assertEqual(calls[0][0]["function"]["name"], "get_weather")
|
||||
self.assertEqual(json.loads(calls[0][0]["function"]["arguments"]),
|
||||
{"city": "Milan"})
|
||||
finish = [e["choices"][0]["finish_reason"] for e in events
|
||||
if e["choices"] and e["choices"][0].get("finish_reason")]
|
||||
self.assertEqual(finish, ["tool_calls"])
|
||||
|
||||
def test_tool_result_round_trip(self):
|
||||
r = self.post({"model": MODEL_ID, "tools": TOOLS, "messages": [
|
||||
{"role": "user", "content": "What is the weather in Rome?"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [
|
||||
{"id": "call_x", "type": "function",
|
||||
"function": {"name": "get_weather",
|
||||
"arguments": "{\"city\": \"Rome\"}"}}]},
|
||||
{"role": "tool", "tool_call_id": "call_x",
|
||||
"content": "25 degrees, sunny"}]})
|
||||
choice = r["choices"][0]
|
||||
self.assertEqual(choice["finish_reason"], "stop")
|
||||
self.assertFalse(choice["message"].get("tool_calls"))
|
||||
self.assertIn("25 degrees", choice["message"]["content"])
|
||||
rendered = self.mock_log.read_text().split("\x00")[-2]
|
||||
self.assertIn("<|observation|><tool_response>25 degrees, sunny</tool_response>",
|
||||
rendered)
|
||||
self.assertIn("# Tools", rendered)
|
||||
self.assertIn('"get_weather"', rendered)
|
||||
|
||||
def test_no_tools_plain_text(self):
|
||||
r = self.post({"model": MODEL_ID,
|
||||
"messages": [{"role": "user", "content": "Hi!"}]})
|
||||
choice = r["choices"][0]
|
||||
self.assertEqual(choice["finish_reason"], "stop")
|
||||
self.assertIn("mock engine", choice["message"]["content"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,39 @@
|
||||
#include "../backend_cuda.h"
|
||||
|
||||
#include <cmath>
|
||||
#include <cstdio>
|
||||
#include <vector>
|
||||
|
||||
int main(){
|
||||
int dev=0;if(!coli_cuda_init(&dev,1))return 77;
|
||||
constexpr int S=3,H=2,Q=2,R=1,V=2,K=3,D=H*V,O=3,T=3;
|
||||
std::vector<float> w(H*(Q+V)*K),p(O*D),q(S*H*(Q+R));
|
||||
for(size_t i=0;i<w.size();i++)w[i]=((int)(i%11)-5)*.07f;
|
||||
for(size_t i=0;i<p.size();i++)p[i]=((int)(i%7)-3)*.09f;
|
||||
for(size_t i=0;i<q.size();i++)q[i]=((int)(i%13)-6)*.05f;
|
||||
ColiCudaTensor *tw=nullptr,*tp=nullptr;
|
||||
if(!coli_cuda_tensor_upload(&tw,w.data(),nullptr,0,K,H*(Q+V),dev)||
|
||||
!coli_cuda_tensor_upload(&tp,p.data(),nullptr,0,D,O,dev))return 1;
|
||||
int n[S]={1,2,3};std::vector<std::vector<float>> l(S),r(S);
|
||||
const float *lp[S],*rp[S];
|
||||
const void *keys[S];
|
||||
for(int s=0;s<S;s++){
|
||||
l[s].resize(n[s]*K);r[s].resize(n[s]*R);
|
||||
for(size_t i=0;i<l[s].size();i++)l[s][i]=((int)((i+s*3)%9)-4)*.08f;
|
||||
for(size_t i=0;i<r[s].size();i++)r[s][i]=((int)((i+s)%5)-2)*.06f;
|
||||
lp[s]=l[s].data();rp[s]=r[s].data();keys[s]=&l[s];
|
||||
}
|
||||
float got[S*O],ref[S*O],warm[S*O];
|
||||
int first[S]={1,1,1};
|
||||
if(!coli_cuda_attention_project_ragged(tw,tp,warm,q.data(),keys,lp,rp,first,
|
||||
S,H,Q,R,V,K,1,.2f))return 2;
|
||||
if(!coli_cuda_attention_project_ragged(tw,tp,got,q.data(),keys,lp,rp,n,S,H,Q,R,V,K,T,.2f))return 2;
|
||||
for(int s=0;s<S;s++)if(!coli_cuda_attention_project_batch(tw,tp,ref+s*O,
|
||||
q.data()+s*H*(Q+R),lp[s],rp[s],1,H,Q,R,V,K,n[s],.2f))return 3;
|
||||
double e=0,z=0;for(int i=0;i<S*O;i++){
|
||||
double d=got[i]-ref[i];e+=d*d;z+=(double)ref[i]*ref[i];
|
||||
}
|
||||
double rms=std::sqrt(e/(z+1e-30));std::printf("ragged_relative_rms=%.9g\n",rms);
|
||||
coli_cuda_tensor_free(tw);coli_cuda_tensor_free(tp);coli_cuda_shutdown();
|
||||
return rms<1e-6?0:4;
|
||||
}
|
||||
@@ -5,14 +5,17 @@ import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from resource_plan import (
|
||||
GB,
|
||||
analyze_model,
|
||||
build_plan,
|
||||
cpu_socket_count,
|
||||
environment_for_plan,
|
||||
format_plan,
|
||||
memory_available,
|
||||
physical_cpu_count,
|
||||
)
|
||||
|
||||
|
||||
@@ -66,15 +69,19 @@ class ResourcePlanTest(unittest.TestCase):
|
||||
# 0 slots/layer. The value must be a sane positive number of bytes.
|
||||
self.assertGreater(memory_available(), 0)
|
||||
|
||||
def test_cpu_socket_count_is_positive(self):
|
||||
self.assertGreaterEqual(cpu_socket_count(), 1)
|
||||
|
||||
def test_builds_bounded_three_tier_plan(self):
|
||||
gpus = [{"index": 0, "name": "test-gpu", "total_bytes": 12 * GB,
|
||||
"free_bytes": 10 * GB}]
|
||||
plan = build_plan(self.model, ram_gb=16, context=32, vram_gb=20,
|
||||
available_memory=32 * GB, available_disk=100 * GB, gpus=gpus,
|
||||
physical_cpus=24)
|
||||
physical_cpus=24, cpu_sockets=2)
|
||||
self.assertEqual(plan["version"], 2)
|
||||
self.assertEqual(plan["policy"]["name"], "quality")
|
||||
self.assertEqual(plan["cpu"]["physical_cores"], 24)
|
||||
self.assertEqual(plan["cpu"]["sockets"], 2)
|
||||
self.assertTrue(plan["policy"]["preserve_quantization"])
|
||||
self.assertFalse(plan["tiers"]["vram"]["requires_host_backing"])
|
||||
self.assertEqual(plan["tiers"]["ram"]["budget_bytes"], 16 * GB)
|
||||
@@ -82,6 +89,79 @@ class ResourcePlanTest(unittest.TestCase):
|
||||
self.assertIn("clamped", plan["warnings"][0])
|
||||
self.assertIn("0:test-gpu", format_plan(plan))
|
||||
|
||||
def test_auto_tier_thread_count_uses_physical_cores(self):
|
||||
# End-to-end for #325: build_plan + environment_for_plan must export the
|
||||
# physical (not logical SMT) core count as OMP_NUM_THREADS. The original
|
||||
# suite passed physical_cpus=24 explicitly, so it never exercised the
|
||||
# real physical_cpu_count() probe whose single-core failure pinned decode.
|
||||
def lscpu(stdout):
|
||||
return subprocess.CompletedProcess(args=[], returncode=0,
|
||||
stdout=stdout, stderr="")
|
||||
# 1 socket, 12 cores, 2 SMT siblings -> 24 threads, 12 physical cores.
|
||||
|
||||
# The parser must return 12 physical cores under BOTH lscpu layouts:
|
||||
# - 2-col: `lscpu -p=core,socket` emits exactly [core,socket] (this is
|
||||
# what the probe actually requests; the previous fields[1]/[2]
|
||||
# indexing skipped every line here and fell through to the
|
||||
# logical count -> the regression JustVugg caught).
|
||||
# - 3-col: bare `lscpu -p` prepends a CPU column -> [cpu,core,socket].
|
||||
# Taking the last two fields is correct in both cases.
|
||||
layouts = {
|
||||
"2-col (-p=core,socket)": (
|
||||
"# core,socket\n" +
|
||||
"\n".join(f"{core},0" for core in range(12) for _ in range(2))),
|
||||
"3-col (bare -p, CPU prefix)": (
|
||||
"# CPU,Core,Socket\n" +
|
||||
"\n".join(f"{cpu},{core},0" for core in range(12) for cpu in range(2))),
|
||||
}
|
||||
for label, blob in layouts.items():
|
||||
with mock.patch("resource_plan.subprocess.run",
|
||||
return_value=lscpu(blob)), \
|
||||
mock.patch.object(sys, "platform", "linux"):
|
||||
plan = build_plan(self.model, available_memory=16 * GB,
|
||||
available_disk=1, gpus=[])
|
||||
env = environment_for_plan(plan)
|
||||
self.assertEqual(plan["cpu"]["physical_cores"], 12, label)
|
||||
self.assertEqual(env["OMP_NUM_THREADS"], "12", label)
|
||||
|
||||
def test_plan_does_not_set_omp_affinity_vars(self):
|
||||
# The real #325 regression: --auto-tier set OMP_PROC_BIND=spread +
|
||||
# OMP_PLACES=cores, which ran before the engine's overwrite=0 setenv and
|
||||
# so won, collapsing the OpenMP team to one CPU on the reporter's 64-core
|
||||
# Linux box even though OMP_NUM_THREADS was correct. The plan must leave
|
||||
# affinity to the engine's own hot-thread tuning (which prefers 'close').
|
||||
plan = build_plan(self.model, available_memory=16 * GB,
|
||||
available_disk=1, gpus=[], physical_cpus=64)
|
||||
env = environment_for_plan(plan)
|
||||
self.assertEqual(env["OMP_NUM_THREADS"], "64")
|
||||
self.assertNotIn("OMP_PROC_BIND", env)
|
||||
self.assertNotIn("OMP_PLACES", env)
|
||||
|
||||
def test_plan_conserves_budget_and_experts_above_256gb(self):
|
||||
# Regression for #325's reporter: a 512 GB machine loading the whole
|
||||
# model into RAM. Verify the budget math stays exact at large RAM sizes
|
||||
# (no integer truncation, no over-allocation, no experts lost between
|
||||
# tiers). Checked at 256/512/800 GB to bracket the reporter's box.
|
||||
for ram_gb in (256, 512, 800):
|
||||
plan = build_plan(self.model, ram_gb=ram_gb, available_disk=1,
|
||||
gpus=[], physical_cpus=64)
|
||||
ram = plan["tiers"]["ram"]
|
||||
# RAM budget never over-allocated: dense + runtime + cache <= budget.
|
||||
allocated = (ram["dense_bytes"] + ram["runtime_bytes"]
|
||||
+ ram["expert_cache_bytes"])
|
||||
self.assertLessEqual(allocated, ram["budget_bytes"],
|
||||
f"over-allocated RAM at {ram_gb} GB")
|
||||
# Every expert byte is accounted for exactly once across the tiers.
|
||||
tiers = plan["tiers"]
|
||||
tiered = (tiers["vram"]["hot_expert_bytes"]
|
||||
+ ram["warm_expert_bytes"]
|
||||
+ tiers["disk"]["cold_expert_bytes"])
|
||||
self.assertEqual(tiered, plan["model"]["expert_bytes"],
|
||||
f"expert bytes lost/duplicated at {ram_gb} GB")
|
||||
# A positive RAM budget yields a non-negative cache and a sensible cap.
|
||||
self.assertGreaterEqual(ram["expert_cache_bytes"], 0)
|
||||
self.assertGreaterEqual(ram["cache_slots_per_layer"], 0)
|
||||
|
||||
def test_filters_requested_devices(self):
|
||||
gpus = [{"index": 0, "name": "a", "total_bytes": 8 * GB, "free_bytes": 8 * GB}]
|
||||
plan = build_plan(self.model, available_memory=16 * GB, available_disk=1,
|
||||
@@ -105,26 +185,111 @@ class ResourcePlanTest(unittest.TestCase):
|
||||
{"index": 1, "name": "b", "total_bytes": 12 * GB, "free_bytes": 10 * GB},
|
||||
]
|
||||
plan = build_plan(self.model, ram_gb=16, available_memory=32 * GB,
|
||||
available_disk=1, gpus=gpus)
|
||||
available_disk=1, gpus=gpus, cpu_sockets=2)
|
||||
env = environment_for_plan(plan, {"RAM_GB": "12", "PIN": "stats.txt",
|
||||
"COLI_GPUS": "1"})
|
||||
self.assertEqual(env["RAM_GB"], "12")
|
||||
self.assertEqual(env["COLI_CUDA"], "1")
|
||||
self.assertEqual(env["COLI_GPUS"], "1")
|
||||
self.assertEqual(env["OMP_NUM_THREADS"], str(plan["cpu"]["physical_cores"]))
|
||||
if sys.platform == "win32":
|
||||
# MinGW libgomp: niente affinity su Windows, le chiavi non vanno emesse
|
||||
self.assertNotIn("OMP_PROC_BIND", env)
|
||||
self.assertNotIn("OMP_PLACES", env)
|
||||
else:
|
||||
self.assertEqual(env["OMP_PROC_BIND"], "spread")
|
||||
self.assertEqual(env["OMP_PLACES"], "cores")
|
||||
# The plan must NOT set OMP_PROC_BIND / OMP_PLACES on any platform:
|
||||
# the engine's own hot-thread tuning owns affinity (it prefers
|
||||
# OMP_PROC_BIND=close for the back-to-back per-expert matmuls). Setting
|
||||
# spread + cores here ran before the engine's overwrite=0 setenv and so
|
||||
# won, collapsing the team to one CPU on some libgomp topologies (#325).
|
||||
self.assertNotIn("OMP_PROC_BIND", env)
|
||||
self.assertNotIn("OMP_PLACES", env)
|
||||
self.assertEqual(env["PIN_GB"], env["CUDA_EXPERT_GB"])
|
||||
|
||||
explicit_threads = environment_for_plan(plan, {"OMP_NUM_THREADS": "7",
|
||||
"OMP_PROC_BIND": "close"})
|
||||
self.assertEqual(explicit_threads["OMP_NUM_THREADS"], "7")
|
||||
self.assertEqual(explicit_threads["OMP_PROC_BIND"], "close")
|
||||
|
||||
if sys.platform.startswith("linux"):
|
||||
self.assertEqual(env["COLI_NUMA"], "1")
|
||||
explicit_numa = environment_for_plan(plan, {"COLI_NUMA": "0"})
|
||||
self.assertEqual(explicit_numa["COLI_NUMA"], "0")
|
||||
|
||||
def test_single_socket_plan_does_not_enable_numa(self):
|
||||
plan = build_plan(self.model, available_memory=16 * GB, available_disk=1,
|
||||
gpus=[], physical_cpus=8, cpu_sockets=1)
|
||||
self.assertNotIn("COLI_NUMA", environment_for_plan(plan))
|
||||
|
||||
def test_auto_tune_mtp_off_when_compute_bound(self):
|
||||
# Tiny model with 64 GB RAM and no GPU: all experts fit in RAM with no
|
||||
# warm tier, so the plan classifies as compute-bound.
|
||||
plan = build_plan(self.model, ram_gb=64, available_memory=64 * GB,
|
||||
available_disk=100 * GB, gpus=[], physical_cpus=24,
|
||||
cpu_sockets=2)
|
||||
# With such a small model fully in RAM and no GPU, bottleneck is compute
|
||||
self.assertEqual(plan["bottleneck_class"], "compute")
|
||||
self.assertIn("DRAFT", plan["tune"])
|
||||
self.assertEqual(plan["tune"]["DRAFT"]["value"], "0")
|
||||
env = environment_for_plan(plan)
|
||||
self.assertEqual(env["DRAFT"], "0")
|
||||
explicit = environment_for_plan(plan, {"DRAFT": "3"})
|
||||
self.assertEqual(explicit["DRAFT"], "3")
|
||||
|
||||
def test_auto_tune_mtp_off_when_disk_low_hit(self):
|
||||
# Use a model large enough that 8 GB RAM can't hold all experts.
|
||||
big = tempfile.TemporaryDirectory()
|
||||
bigmodel = Path(big.name)
|
||||
(bigmodel / "config.json").write_text(json.dumps({
|
||||
"num_hidden_layers": 2, "n_routed_experts": 4,
|
||||
"kv_lora_rank": 4, "qk_rope_head_dim": 2,
|
||||
"qk_nope_head_dim": 3, "v_head_dim": 5, "num_attention_heads": 2,
|
||||
}))
|
||||
expert_size = 3 * GB # each expert 3 GB → 12 GB total, won't fit in 8 GB budget
|
||||
write_shard(bigmodel / "out-00000.safetensors", [
|
||||
("model.embed_tokens.weight", 100),
|
||||
("model.layers.0.self_attn.q_a_proj.weight", 200),
|
||||
])
|
||||
for i in range(4):
|
||||
write_shard(bigmodel / f"out-{i+1:05d}.safetensors", [
|
||||
(f"model.layers.1.mlp.experts.{i}.gate_proj.weight", expert_size),
|
||||
])
|
||||
plan = build_plan(bigmodel, ram_gb=0, available_memory=4 * GB,
|
||||
available_disk=100 * GB, gpus=[], physical_cpus=8,
|
||||
cpu_sockets=1)
|
||||
big.cleanup()
|
||||
self.assertEqual(plan["bottleneck_class"], "disk")
|
||||
self.assertLess(plan["projected_hit_rate"], 0.90)
|
||||
self.assertEqual(plan["tune"]["DRAFT"]["value"], "0")
|
||||
|
||||
def test_auto_tune_pipe_multi_gpu(self):
|
||||
gpus = [
|
||||
{"index": 0, "name": "a", "total_bytes": 32 * GB, "free_bytes": 30 * GB},
|
||||
{"index": 1, "name": "b", "total_bytes": 32 * GB, "free_bytes": 30 * GB},
|
||||
]
|
||||
plan = build_plan(self.model, ram_gb=16, available_memory=32 * GB,
|
||||
available_disk=1, gpus=gpus, cpu_sockets=2)
|
||||
self.assertEqual(plan["tune"]["COLI_CUDA_PIPE"]["value"], "2")
|
||||
env = environment_for_plan(plan)
|
||||
self.assertEqual(env["COLI_CUDA_PIPE"], "2")
|
||||
|
||||
def test_auto_tune_pipe_single_gpu(self):
|
||||
gpus = [{"index": 0, "name": "a", "total_bytes": 12 * GB, "free_bytes": 10 * GB}]
|
||||
plan = build_plan(self.model, ram_gb=16, available_memory=32 * GB,
|
||||
available_disk=1, gpus=gpus, cpu_sockets=1)
|
||||
self.assertEqual(plan["tune"]["COLI_CUDA_PIPE"]["value"], "1")
|
||||
|
||||
def test_auto_tune_numa_hint_for_cpu_only(self):
|
||||
plan = build_plan(self.model, ram_gb=64, available_memory=64 * GB,
|
||||
available_disk=1, gpus=[], physical_cpus=64, cpu_sockets=2)
|
||||
self.assertIn("_numa_hint", plan["tune"])
|
||||
self.assertIn("numactl", plan["tune"]["_numa_hint"])
|
||||
self.assertIn("auto-tune", format_plan(plan))
|
||||
|
||||
def test_format_plan_shows_tune_and_hit_rate(self):
|
||||
plan = build_plan(self.model, ram_gb=64, available_memory=64 * GB,
|
||||
available_disk=100 * GB, gpus=[], physical_cpus=24,
|
||||
cpu_sockets=1)
|
||||
text = format_plan(plan)
|
||||
self.assertIn("hit", text)
|
||||
self.assertIn("auto-tune", text)
|
||||
self.assertIn("DRAFT", text)
|
||||
|
||||
def test_cpu_binary_does_not_apply_gpu_tier(self):
|
||||
plan = build_plan(self.model, available_memory=16 * GB, available_disk=1,
|
||||
gpus=[{"index": 0, "name": "a", "total_bytes": 8 * GB,
|
||||
@@ -163,5 +328,73 @@ class ResourcePlanTest(unittest.TestCase):
|
||||
self.assertIn("expected_bottleneck", plan)
|
||||
|
||||
|
||||
class PhysicalCpuCountTest(unittest.TestCase):
|
||||
"""Regression for #325: --auto-tier pinned decode to one core because
|
||||
physical_cpu_count() silently returned 1.
|
||||
|
||||
Two root causes this locks down:
|
||||
1. lscpu -p prepends a CPU column, so `-p=core,socket` emits
|
||||
CPU,Core,Socket; counting rows counted logical SMT siblings.
|
||||
2. any probe failure fell through to ``os.cpu_count() or 1`` and the
|
||||
``or 1`` could pin a constrained/cgroup'd box to a single core.
|
||||
"""
|
||||
|
||||
def _lscpu(self, stdout):
|
||||
return subprocess.CompletedProcess(args=[], returncode=0,
|
||||
stdout=stdout, stderr="")
|
||||
|
||||
def _lscpu_topology(self, sockets, cores_per_socket, threads_per_core):
|
||||
# Real lscpu shape: socket-local core IDs repeat across sockets; the
|
||||
# CPU column (always prepended) is a unique logical-CPU index.
|
||||
rows, cpu = [], 0
|
||||
for sock in range(sockets):
|
||||
for core in range(cores_per_socket):
|
||||
for _ in range(threads_per_core):
|
||||
rows.append(f"{cpu},{core},{sock}")
|
||||
cpu += 1
|
||||
return "# CPU,Core,Socket\n" + "\n".join(rows)
|
||||
|
||||
def test_counts_physical_cores_not_smt_threads(self):
|
||||
blob = self._lscpu_topology(sockets=2, cores_per_socket=16, threads_per_core=2)
|
||||
with mock.patch("resource_plan.subprocess.run", return_value=self._lscpu(blob)), \
|
||||
mock.patch.object(sys, "platform", "linux"):
|
||||
self.assertEqual(physical_cpu_count(), 32)
|
||||
|
||||
def test_single_socket_no_smt(self):
|
||||
blob = self._lscpu_topology(sockets=1, cores_per_socket=8, threads_per_core=1)
|
||||
with mock.patch("resource_plan.subprocess.run", return_value=self._lscpu(blob)), \
|
||||
mock.patch.object(sys, "platform", "linux"):
|
||||
self.assertEqual(physical_cpu_count(), 8)
|
||||
|
||||
def test_skips_offline_core_socket_fields(self):
|
||||
# VMs / large NUMA boxes emit "-" for offline core or socket IDs; that
|
||||
# used to raise ValueError, discard the whole parse, and fall through
|
||||
# to the single-core fallback.
|
||||
blob = "# CPU,Core,Socket\n0,0,0\n1,-,0\n2,1,0\n3,1,0\n"
|
||||
with mock.patch("resource_plan.subprocess.run", return_value=self._lscpu(blob)), \
|
||||
mock.patch.object(sys, "platform", "linux"):
|
||||
self.assertEqual(physical_cpu_count(), 2)
|
||||
|
||||
def test_lscpu_missing_falls_back_to_logical_not_silent_one(self):
|
||||
# The bug: lscpu absent -> os.cpu_count() or 1. On a constrained box
|
||||
# os.cpu_count() can be 1. We still must never silently pick 1 without
|
||||
# a warning, and when logical cores exist they must be used.
|
||||
import os
|
||||
with mock.patch("resource_plan.subprocess.run", side_effect=FileNotFoundError), \
|
||||
mock.patch.object(sys, "platform", "linux"), \
|
||||
mock.patch("resource_plan.os.cpu_count", return_value=16), \
|
||||
mock.patch("sys.stderr"):
|
||||
self.assertEqual(physical_cpu_count(), 16)
|
||||
|
||||
def test_zero_logical_cores_warns_and_returns_one(self):
|
||||
# The genuine degenerate case: no probe works and os.cpu_count() is
|
||||
# None/1. Must return 1 (engine needs a positive team size) but warn.
|
||||
with mock.patch("resource_plan.subprocess.run", side_effect=FileNotFoundError), \
|
||||
mock.patch.object(sys, "platform", "linux"), \
|
||||
mock.patch("resource_plan.os.cpu_count", return_value=None), \
|
||||
mock.patch("sys.stderr"):
|
||||
self.assertEqual(physical_cpu_count(), 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
/* Regressione #369: un solo logit NaN o +Inf faceva emettere a dist_build/dist_sample
|
||||
* il token 0 PER SEMPRE, in silenzio (il fallback `g_pbuf[i]>0` e' falso su NaN ovunque).
|
||||
* Ora la distribuzione degenere ripiega sull'argmax dei logit FINITI e avvisa una volta.
|
||||
*
|
||||
* Il test verifica: (a) con logit sani il campionamento resta corretto; (b) con NaN/+Inf
|
||||
* iniettato il token scelto e' l'argmax dei FINITI (mai 0 per default), su ogni posizione
|
||||
* del NaN inclusa lo[0]; (c) nessun NaN sopravvive in g_pbuf. */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
#include <stdio.h>
|
||||
|
||||
static uint32_t rs=0x1234abcd; static uint32_t xr(){rs^=rs<<13;rs^=rs>>17;rs^=rs<<5;return rs;}
|
||||
|
||||
static int pbuf_has_nan(int V){ for(int i=0;i<V;i++) if(!isfinite(g_pbuf[i])) return 1; return 0; }
|
||||
|
||||
int main(void){
|
||||
int fail=0, V=2000;
|
||||
float *lo=malloc(V*sizeof(float));
|
||||
g_temp=0.7f; g_nuc=0.90f; /* il path serve di default */
|
||||
|
||||
/* (a) logit sani: il campionato deve avere prob > 0 e nessun NaN nel buffer */
|
||||
for(int i=0;i<V;i++) lo[i]=(float)((int)(xr()%2000)-1000)/100.f;
|
||||
int known=1337; lo[known]=50.f; /* picco netto */
|
||||
dist_build(lo,V);
|
||||
if(pbuf_has_nan(V)){ printf(" FAIL: NaN in g_pbuf con logit sani\n"); fail=1; }
|
||||
if(g_pbuf[known]<=0.f){ printf(" FAIL: il picco ha prob 0\n"); fail=1; }
|
||||
if(!fail) printf(" logit sani: distribuzione valida, picco vivo ok\n");
|
||||
|
||||
/* (b) NaN/+Inf iniettato in varie posizioni; il finito-argmax deve vincere */
|
||||
float bad[3]; bad[0]=NAN; bad[1]=INFINITY; bad[2]=-INFINITY;
|
||||
const char *bn[3]={"NaN","+Inf","-Inf"};
|
||||
for(int b=0;b<3;b++){
|
||||
for(int pos=0;pos<3;pos++){ /* lo[0], meta', ultimo */
|
||||
for(int i=0;i<V;i++) lo[i]=(float)((int)(xr()%400)-200)/100.f;
|
||||
int amax=777; lo[amax]=9.0f; /* massimo FINITO atteso */
|
||||
int at = pos==0?0 : pos==1?V/2 : V-1;
|
||||
if(at==amax) amax=amax+1; /* non sovrapporre */
|
||||
lo[amax]=9.0f;
|
||||
lo[at]=bad[b]; /* veleno */
|
||||
dist_build(lo,V);
|
||||
if(pbuf_has_nan(V)){ printf(" FAIL: NaN sopravvive (%s @ %d)\n",bn[b],at); fail=1; continue; }
|
||||
/* con -Inf il fallback puo' non scattare (max finito resta), ma il buffer
|
||||
* deve restare valido e sommare ~1: con +Inf/NaN scatta il delta su amax */
|
||||
int picked=-1; float pv=-1;
|
||||
for(int i=0;i<V;i++) if(g_pbuf[i]>pv){pv=g_pbuf[i];picked=i;}
|
||||
if(b<2 && picked!=amax){ /* NaN e +Inf: delta esatto su amax */
|
||||
printf(" FAIL: %s @ %d -> picked %d, atteso argmax finito %d\n",bn[b],at,picked,amax); fail=1;
|
||||
}
|
||||
}
|
||||
}
|
||||
if(!fail) printf(" NaN/+Inf iniettato: argmax dei finiti vince, mai 0/NaN ok\n");
|
||||
|
||||
/* (c) caso estremo: TUTTI non finiti -> non deve crashare, buffer valido */
|
||||
for(int i=0;i<V;i++) lo[i]=NAN;
|
||||
dist_build(lo,V);
|
||||
if(pbuf_has_nan(V)){ printf(" FAIL: tutti-NaN lascia NaN nel buffer\n"); fail=1; }
|
||||
else printf(" tutti non-finiti: nessun crash, buffer valido ok\n");
|
||||
|
||||
printf(fail?"test_sample_nan: FAIL\n":"test_sample_nan: ok\n");
|
||||
return fail;
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
/* st_pread_full: chunk loop + honest truncation errors.
|
||||
* Built with -DST_PREAD_CHUNK=7 so a ~100-byte tensor takes many pread calls —
|
||||
* exercising the loop that production only needs past 2^31 bytes (one pread
|
||||
* caps there on Linux; big bf16 tensors exceed it). Also forks a child against
|
||||
* a truncated shard and requires exit(1) with a "short read" message instead
|
||||
* of the old perror("... : Success"). */
|
||||
#define _GNU_SOURCE
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#ifndef _WIN32
|
||||
#include <sys/wait.h>
|
||||
#include <unistd.h>
|
||||
#endif
|
||||
|
||||
#include "../st.h"
|
||||
|
||||
#define CHECK(condition) do { \
|
||||
if (!(condition)) { \
|
||||
fprintf(stderr, "%s:%d: check failed: %s\n", __FILE__, __LINE__, #condition); \
|
||||
return 1; \
|
||||
} \
|
||||
} while (0)
|
||||
|
||||
static void write_snap(const char *dir, int truncate_bytes) {
|
||||
char path[512];
|
||||
snprintf(path, sizeof(path), "%s/model.safetensors", dir);
|
||||
unsigned char data[96];
|
||||
for (int i = 0; i < 96; i++) data[i] = (unsigned char)(i * 7 + 3);
|
||||
const char *hdr = "{\"t\":{\"dtype\":\"U8\",\"shape\":[96],\"data_offsets\":[0,96]}}";
|
||||
uint64_t hlen = strlen(hdr);
|
||||
FILE *f = fopen(path, "wb");
|
||||
fwrite(&hlen, 8, 1, f);
|
||||
fwrite(hdr, 1, hlen, f);
|
||||
fwrite(data, 1, (size_t)(96 - truncate_bytes), f);
|
||||
fclose(f);
|
||||
}
|
||||
|
||||
int main(void) {
|
||||
/* relative to the CWD, per test_stops: MinGW .exe files resolve Windows
|
||||
* paths and "/tmp" is not one */
|
||||
char dir[] = "test_st_pread_XXXXXX";
|
||||
if (!mkdtemp(dir)) { perror("mkdtemp"); return 1; }
|
||||
|
||||
/* 1) chunk loop: 96-byte tensor read 7 bytes at a time, content exact */
|
||||
write_snap(dir, 0);
|
||||
shards S; st_init(&S, dir);
|
||||
unsigned char out[96] = {0};
|
||||
st_read_raw(&S, "t", out, 0);
|
||||
for (int i = 0; i < 96; i++) CHECK(out[i] == (unsigned char)(i * 7 + 3));
|
||||
|
||||
#ifndef _WIN32
|
||||
/* 2) shard truncated AFTER st_init (init validates static bounds, so the
|
||||
* pread path only fires when the file shrinks underneath a live handle):
|
||||
* child must exit(1) with an honest message, not perror's "Success" */
|
||||
char shard[512]; snprintf(shard, sizeof(shard), "%s/model.safetensors", dir);
|
||||
struct stat sb; CHECK(stat(shard, &sb) == 0);
|
||||
CHECK(truncate(shard, sb.st_size - 40) == 0);
|
||||
int pipefd[2]; CHECK(pipe(pipefd) == 0);
|
||||
pid_t pid = fork(); CHECK(pid >= 0);
|
||||
if (pid == 0) {
|
||||
dup2(pipefd[1], 2); close(pipefd[0]); close(pipefd[1]);
|
||||
unsigned char buf[96];
|
||||
st_read_raw(&S, "t", buf, 0); /* inherited handles; must exit(1) inside */
|
||||
_exit(42); /* reaching here = bug */
|
||||
}
|
||||
close(pipefd[1]);
|
||||
char err[512] = {0};
|
||||
ssize_t n = read(pipefd[0], err, sizeof(err)-1); (void)n;
|
||||
close(pipefd[0]);
|
||||
int status = 0; waitpid(pid, &status, 0);
|
||||
CHECK(WIFEXITED(status) && WEXITSTATUS(status) == 1);
|
||||
CHECK(strstr(err, "short read") != NULL);
|
||||
CHECK(strstr(err, "Success") == NULL);
|
||||
#else
|
||||
/* fork/pipe/truncate are POSIX; Windows still runs the chunk-loop check */
|
||||
printf("test_st_pread: truncation subtest skipped on Windows\n");
|
||||
#endif
|
||||
|
||||
char cmd[600];
|
||||
#ifdef _WIN32
|
||||
snprintf(cmd, sizeof(cmd), "rmdir /s /q %s", dir);
|
||||
#else
|
||||
snprintf(cmd, sizeof(cmd), "rm -rf %s", dir);
|
||||
#endif
|
||||
if (system(cmd)) {}
|
||||
printf("test_st_pread: chunk loop + honest truncation error: ok\n");
|
||||
return 0;
|
||||
}
|
||||
@@ -20,7 +20,7 @@
|
||||
* Defense 2 is what makes this robust against checkpoints we don't control:
|
||||
* even with BOTH configs mutilated, a control token cannot leak into a reply. */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../glm.c"
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
static const char *TOKJSON =
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
/* Top-p (nucleus) truncation in dist_build: the partial-select rewrite (#335) must be
|
||||
* indistinguishable from the old full-vocab qsort for every shape dist_sample can see.
|
||||
*
|
||||
* Why this test exists (#335): dist_build() previously qsort-ed the entire 151936-entry
|
||||
* vocab on every sampled token to find the few-hundred-token head whose cumulative mass
|
||||
* reaches g_nuc. It now heapifies (O(V)) and pops only the head (k * O(log V)). The win
|
||||
* is structural; the risk is a silent sampling-distribution change, because the contract
|
||||
* is subtle:
|
||||
*
|
||||
* dist_sample() iterates g_pbuf[0..V-1] BY TOKEN ID and sums probabilities directly.
|
||||
* So dist_build MUST leave g_pbuf indexed by id (never reordered) AND must zero every
|
||||
* truncated tail entry -- merely excluding the tail from the head would leave mass on
|
||||
* it and the sampled distribution would drift with no crash and no error.
|
||||
*
|
||||
* Strategy: drive the REAL dist_build (via the test_stops.c include-glm.c pattern) on a
|
||||
* sweep of distributions and g_nuc values, and compare against an INDEPENDENT reference
|
||||
* that re-implements the OLD algorithm (full qsort + zero-tail + renorm) in double on a
|
||||
* private buffer. On shapes with no ties the renormalized head must be BIT-IDENTICAL to
|
||||
* the reference (the issue's stated invariant: s2 accumulates in the same descending
|
||||
* order). On tie shapes, where the unstable qsort already left ordering unspecified, we
|
||||
* check multiset equality instead. Every shape also checks: exact-zero tails, head sums
|
||||
* to 1.0, and a sane keep-count.
|
||||
*
|
||||
* No scratch files: the test runs entirely in memory (no mkdtemp), so it builds clean on
|
||||
* the Windows MinGW CI job without the unmerged compat shim (#352). */
|
||||
#define main coli_glm_main_unused
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
#include <math.h>
|
||||
|
||||
static int g_nfails = 0;
|
||||
|
||||
/* pointer set by ref_build so cmp_ref_desc can read the current reference buffer
|
||||
* (the qsort comparator gets no user-data argument in C). */
|
||||
static const double *g_ref_p = NULL;
|
||||
|
||||
#define FAIL(fmt, ...) do { \
|
||||
fprintf(stderr, " FAIL [%s V=%d nuc=%.3f shape=%s]: " fmt "\n", \
|
||||
label, V, nuc, shape_name, ##__VA_ARGS__); \
|
||||
g_nfails++; \
|
||||
return; \
|
||||
} while (0)
|
||||
|
||||
/* ---- independent reference: the OLD algorithm, in double, on a private buffer ------- */
|
||||
/* Stable qsort by descending probability (ties broken by ascending index, which makes
|
||||
* the reference deterministic regardless of the production comparator). */
|
||||
static int cmp_ref_desc(const void *a, const void *b){
|
||||
double pa = ((const double *)g_ref_p)[*(const int*)a];
|
||||
double pb = ((const double *)g_ref_p)[*(const int*)b];
|
||||
if (pa < pb) return 1;
|
||||
if (pa > pb) return -1;
|
||||
/* tie -> lower index first (stable, unlike the production comparator) */
|
||||
return *(const int*)a - *(const int*)b;
|
||||
}
|
||||
|
||||
/* Build the reference distribution into out[0..V-1] (indexed by token id), mirroring the
|
||||
* old dist_build: softmax(lo/temp) truncated to top-p nuc, tail zeroed, head renormalized.
|
||||
* Returns the keep-count through *keep_out. */
|
||||
static void ref_build(const float *lo, int V, double temp, double nuc,
|
||||
double *out, int *pidx, int *keep_out){
|
||||
double mx = lo[0]; for (int i = 1; i < V; i++) if (lo[i] > mx) mx = lo[i];
|
||||
double s = 0, invt = 1.0 / (temp > 1e-4 ? temp : 1e-4);
|
||||
for (int i = 0; i < V; i++){ out[i] = exp((lo[i]-mx)*invt); s += out[i]; }
|
||||
for (int i = 0; i < V; i++) out[i] /= s;
|
||||
|
||||
if (nuc > 0 && nuc < 1.0){
|
||||
for (int i = 0; i < V; i++) pidx[i] = i;
|
||||
qsort(pidx, V, sizeof(int), cmp_ref_desc);
|
||||
double cum = 0; int keep = V;
|
||||
for (int i = 0; i < V; i++){ cum += out[pidx[i]]; if (cum >= nuc){ keep = i+1; break; } }
|
||||
double s2 = 0;
|
||||
for (int i = keep; i < V; i++) out[pidx[i]] = 0;
|
||||
for (int i = 0; i < keep; i++) s2 += out[pidx[i]];
|
||||
for (int i = 0; i < keep; i++) out[pidx[i]] /= s2;
|
||||
*keep_out = keep;
|
||||
} else {
|
||||
*keep_out = V;
|
||||
}
|
||||
}
|
||||
|
||||
/* count how many production g_pbuf entries are non-zero == the head size */
|
||||
static int head_count(int V){
|
||||
int n = 0; for (int i = 0; i < V; i++) if (g_pbuf[i] != 0.f) n++; return n;
|
||||
}
|
||||
|
||||
/* Run one case: load logits into g_pbuf via the real dist_build, compare to reference.
|
||||
* shape_name is for diagnostics only. */
|
||||
static void check_case(const char *label, int V, double nuc, const char *shape_name,
|
||||
const float *lo){
|
||||
/* reference on a private buffer */
|
||||
double *ref = malloc((size_t)V * sizeof(double));
|
||||
int *ridx = malloc((size_t)V * sizeof(int));
|
||||
int ref_keep = 0;
|
||||
g_ref_p = ref; /* cmp_ref_desc reads this */
|
||||
ref_build(lo, V, g_temp, nuc, ref, ridx, &ref_keep);
|
||||
|
||||
/* production: drive the real dist_build (writes the global g_pbuf) */
|
||||
g_nuc = (float)nuc;
|
||||
dist_build(lo, V);
|
||||
|
||||
int got_keep = head_count(V);
|
||||
|
||||
/* 1. keep-count must match the reference exactly. The partial select and the old
|
||||
* qsort keep the same NUMBER of tokens by construction (same cumulative-mass rule);
|
||||
* a count divergence is a real bug, not a tie artifact. */
|
||||
if (got_keep != ref_keep)
|
||||
FAIL("keep-count mismatch: got %d, ref %d", got_keep, ref_keep);
|
||||
|
||||
/* 2. Detect ties across the WHOLE pre-truncation distribution, not just the kept set.
|
||||
* A tie at the head/tail boundary makes which-side-a-token-lands-on interchangeable:
|
||||
* both algorithms keep the right count but may keep different MEMBERS. So any input
|
||||
* with a duplicated softmax value needs the relaxed multiset comparison below. We
|
||||
* detect this on the reference softmax (pre-truncation) by sorting all V values. */
|
||||
int has_ties = 0;
|
||||
{
|
||||
double *all = malloc((size_t)V * sizeof(double));
|
||||
/* reconstruct the pre-truncation softmax the same way ref_build does */
|
||||
double mx = lo[0]; for (int i = 1; i < V; i++) if (lo[i] > mx) mx = lo[i];
|
||||
double s = 0, invt = 1.0 / (g_temp > 1e-4 ? g_temp : 1e-4);
|
||||
for (int i = 0; i < V; i++){ all[i] = exp((lo[i]-mx)*invt); s += all[i]; }
|
||||
for (int i = 0; i < V; i++) all[i] /= s;
|
||||
for (int a = 1; a < V; a++){ double k = all[a]; int b = a-1;
|
||||
while (b >= 0 && all[b] > k){ all[b+1] = all[b]; b--; } all[b+1] = k; }
|
||||
for (int a = 1; a < V; a++) if (all[a] == all[a-1]){ has_ties = 1; break; }
|
||||
free(all);
|
||||
}
|
||||
|
||||
if (has_ties){
|
||||
/* Multiset equality of the non-zero (head) values. Ties make membership
|
||||
* interchangeable, so we compare sorted value-multisets, not id-aligned values.
|
||||
* Tolerance is 1e-6 relative -- the engine uses float arithmetic, the reference
|
||||
* double, so sub-ULP noise is expected (matches test_i4_grouped.c's convention). */
|
||||
double *got = malloc((size_t)ref_keep * sizeof(double));
|
||||
int gm = 0;
|
||||
for (int i = 0; i < V; i++) if (g_pbuf[i] != 0.f) got[gm++] = (double)g_pbuf[i];
|
||||
if (gm != ref_keep)
|
||||
FAIL("tie-shape head size mismatch: got %d non-zero, ref %d", gm, ref_keep);
|
||||
for (int a = 1; a < gm; a++){ double k = got[a]; int b = a-1;
|
||||
while (b >= 0 && got[b] > k){ got[b+1] = got[b]; b--; } got[b+1] = k; }
|
||||
double *rsort = malloc((size_t)ref_keep * sizeof(double));
|
||||
int rm = 0;
|
||||
for (int i = 0; i < V; i++) if (ref[i] != 0.0) rsort[rm++] = ref[i];
|
||||
for (int a = 1; a < rm; a++){ double k = rsort[a]; int b = a-1;
|
||||
while (b >= 0 && rsort[b] > k){ rsort[b+1] = rsort[b]; b--; } rsort[b+1] = k; }
|
||||
int mm = 0; double worst = 0;
|
||||
for (int i = 0; i < gm; i++){
|
||||
double d = fabs(got[i] - rsort[i]);
|
||||
double rel = rsort[i] > 1e-30 ? d / rsort[i] : d;
|
||||
if (rel > worst) worst = rel;
|
||||
if (rel > 1e-6) mm++;
|
||||
}
|
||||
free(got); free(rsort);
|
||||
if (mm) FAIL("tie-shape multiset mismatch: %d/%d head values differ beyond 1e-6 rel (worst %.3g)",
|
||||
mm, ref_keep, worst);
|
||||
} else {
|
||||
/* No ties anywhere: membership is forced, so compare id-aligned head values. The
|
||||
* engine computes in float (g_pbuf /= (float)s2) while the reference uses double,
|
||||
* so the comparison is relative-tolerance (1e-6), not bit-exact -- the partial
|
||||
* select and qsort accumulate s2 in the same descending order, so any difference
|
||||
* is pure float-rounding noise, not an ordering bug. */
|
||||
int bad = 0; int first_id = -1; float gv = 0, rv = 0; double worst = 0;
|
||||
for (int i = 0; i < V; i++){
|
||||
if (ref[i] == 0.0) continue; /* tail */
|
||||
float want = (float)ref[i];
|
||||
double d = fabs((double)g_pbuf[i] - (double)want);
|
||||
double rel = fabs((double)want) > 1e-30 ? d / fabs((double)want) : d;
|
||||
if (rel > worst) worst = rel;
|
||||
if (rel > 1e-6){
|
||||
bad++; if (first_id < 0){ first_id = i; gv = g_pbuf[i]; rv = want; }
|
||||
if (bad > 3) break;
|
||||
}
|
||||
}
|
||||
if (bad)
|
||||
FAIL("head not within 1e-6 rel of reference: %d entries differ (first id %d: got %.9g want %.9g, worst %.3g)",
|
||||
bad, first_id, (double)gv, (double)rv, worst);
|
||||
}
|
||||
|
||||
/* 3. head must renormalize to 1.0 (within float epsilon) */
|
||||
double sum = 0; for (int i = 0; i < V; i++) sum += g_pbuf[i];
|
||||
if (fabs(sum - 1.0) > 1e-5)
|
||||
FAIL("head does not sum to 1.0: sum=%.12g (keep=%d)", sum, got_keep);
|
||||
|
||||
free(ref); free(ridx);
|
||||
printf(" ok [V=%d nuc=%.3f shape=%s keep=%d%s sum=%.10f]\n",
|
||||
V, nuc, shape_name, got_keep, has_ties ? " (ties)" : "", sum);
|
||||
}
|
||||
#undef FAIL
|
||||
|
||||
/* deterministic xorshift32 RNG (matches the test_i4_grouped.c convention) */
|
||||
static uint32_t rng_state = 0x12345678u;
|
||||
static uint32_t xr(void){ rng_state ^= rng_state << 13; rng_state ^= rng_state >> 17;
|
||||
rng_state ^= rng_state << 5; return rng_state; }
|
||||
static double frand(void){ return (xr() >> 8) * (1.0 / 16777216.0); } /* [0,1) */
|
||||
|
||||
/* fill logits for a given shape. Shapes chosen to stress the comparator and the head/tail
|
||||
* boundary differently. */
|
||||
static void fill_shape(float *lo, int V, int shape){
|
||||
switch (shape){
|
||||
case 0: /* uniform -> every token equal probability -> massive tie plateau */
|
||||
for (int i = 0; i < V; i++) lo[i] = 0.f; break;
|
||||
case 1: /* peaked: one dominant token, rest small and distinct (no ties).
|
||||
* The fixed hot-spots are clamped to V-1 so small V (incl. V=1) doesn't
|
||||
* write out of bounds and corrupt heap metadata on the later free(lo). */
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(-1.0 - frand()*4.0);
|
||||
lo[0] = 3.f; lo[V/3<V?V/3:V-1] = 1.f; lo[V/2<V?V/2:V-1] = 0.5f; break;
|
||||
case 2: /* all-equal distinct decay (no ties): geometric, strictly decreasing */
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(-0.001 * (double)i); break;
|
||||
case 3: /* plateau ties: blocks of equal value -> comparator tie handling */
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(-(double)(i / 7)); /* 7-wide plateaus */
|
||||
break;
|
||||
case 4: /* sharp-tail: a few hot, then a long flat floor (small tie at the floor).
|
||||
* Hot count is min(12,V) so V<12 (incl. V=1) stays in bounds. */
|
||||
for (int i = 0; i < V; i++) lo[i] = -8.f;
|
||||
{ int hot = V<12 ? V : 12; for (int i = 0; i < hot; i++) lo[i] = (float)(2.0 - frand()); } break;
|
||||
}
|
||||
}
|
||||
|
||||
int main(void){
|
||||
/* sizes: small for exhaustive tie detection up to near-production scale */
|
||||
int sizes[] = {1, 2, 8, 64, 257, 1519}; /* 1519 ~= V/100 of GLM-5.2 */
|
||||
double nucs[] = {0.001, 0.5, 0.9, 0.999}; /* tight -> almost-everything */
|
||||
int n_shapes = 5;
|
||||
|
||||
/* temperature used by dist_build: pick a normal serving value */
|
||||
g_temp = 0.7f;
|
||||
|
||||
int cases = 0;
|
||||
for (size_t si = 0; si < sizeof(sizes)/sizeof(sizes[0]); si++){
|
||||
int V = sizes[si];
|
||||
/* dist_build allocates g_pbuf/g_pidx ONCE and reuses them (single-V invariant in
|
||||
* real serving, where V is the constant model vocab). This sweep varies V, so free
|
||||
* and force a reallocation per size -- otherwise a later, larger V would overflow
|
||||
* the buffer sized for the first (smallest) V. */
|
||||
free(g_pbuf); g_pbuf = NULL; free(g_pidx); g_pidx = NULL;
|
||||
float *lo = malloc((size_t)V * sizeof(float));
|
||||
for (int shape = 0; shape < n_shapes; shape++){
|
||||
fill_shape(lo, V, shape);
|
||||
for (size_t ni = 0; ni < sizeof(nucs)/sizeof(nucs[0]); ni++){
|
||||
char label[32]; snprintf(label, sizeof(label), "size[%zu]/shape[%d]", si, shape);
|
||||
const char *sn = (const char*[]){"uniform","peaked","geometric","plateau","sharptail"}[shape];
|
||||
check_case(label, V, nucs[ni], sn, lo);
|
||||
cases++;
|
||||
}
|
||||
}
|
||||
free(lo);
|
||||
}
|
||||
|
||||
/* guard-off path: g_nuc >= 1 must skip truncation entirely (full softmax kept) */
|
||||
{
|
||||
int V = 256; float lo[256];
|
||||
for (int i = 0; i < V; i++) lo[i] = (float)(frand()*4 - 2);
|
||||
g_nuc = 1.0f; dist_build(lo, V);
|
||||
int nz = 0; for (int i = 0; i < V; i++) if (g_pbuf[i] != 0.f) nz++;
|
||||
if (nz != V){ fprintf(stderr, " FAIL [guard-off nuc=1.0]: %d/%d entries kept, expected all\n", nz, V); g_nfails++; }
|
||||
else printf(" ok [guard-off nuc=1.0 keep=%d]\n", nz);
|
||||
cases++;
|
||||
|
||||
g_nuc = 0.0f; dist_build(lo, V);
|
||||
nz = 0; for (int i = 0; i < V; i++) if (g_pbuf[i] != 0.f) nz++;
|
||||
if (nz != V){ fprintf(stderr, " FAIL [guard-off nuc=0.0]: %d/%d entries kept, expected all\n", nz, V); g_nfails++; }
|
||||
else printf(" ok [guard-off nuc=0.0 keep=%d]\n", nz);
|
||||
cases++;
|
||||
}
|
||||
|
||||
/* extreme tie edge case: V=1, single token -> keep=1 regardless of nuc */
|
||||
{
|
||||
float lo[1] = {5.f};
|
||||
g_nuc = 0.5f; dist_build(lo, 1);
|
||||
if (g_pbuf[0] == 0.f || !(fabs((double)g_pbuf[0] - 1.0) < 1e-6)){
|
||||
fprintf(stderr, " FAIL [V=1]: g_pbuf[0]=%.9g, expected 1.0\n", (double)g_pbuf[0]); g_nfails++;
|
||||
} else printf(" ok [V=1 keep=1]\n");
|
||||
cases++;
|
||||
}
|
||||
|
||||
printf("\ntest_topp: %d cases run, %d failure(s)\n", cases, g_nfails);
|
||||
if (g_nfails){ printf("test_topp: FAIL\n"); return 1; }
|
||||
printf("test_topp: ok\n");
|
||||
return 0;
|
||||
}
|
||||
@@ -6,7 +6,7 @@
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
#define main coli_glm_main_unused
|
||||
#include "../glm.c"
|
||||
#include "../colibri.c"
|
||||
#undef main
|
||||
|
||||
static int fail(const char *s){ fprintf(stderr,"FAIL: %s\n",s); return 1; }
|
||||
|
||||
Reference in New Issue
Block a user