JustVugg · ebootheee · Jul 17, 2026
diff --git a/c/Makefile b/c/Makefile
@@ -166,7 +166,7 @@ else
 PYTHON    ?= python3
 endif
 CUDA_OBJ  =
-TEST_BINS = tests/test_json$(EXE) tests/test_st$(EXE) tests/test_tier$(EXE) tests/test_grammar$(EXE) tests/test_schema_gbnf$(EXE) tests/test_decode_batch$(EXE) tests/test_idot$(EXE) tests/test_i4_grouped$(EXE) tests/test_stops$(EXE) tests/test_kv_alloc$(EXE) tests/test_i4_acc512$(EXE) tests/test_compat_direct$(EXE)
+TEST_BINS = tests/test_json$(EXE) tests/test_st$(EXE) tests/test_tier$(EXE) tests/test_grammar$(EXE) tests/test_schema_gbnf$(EXE) tests/test_decode_batch$(EXE) tests/test_idot$(EXE) tests/test_i4_grouped$(EXE) tests/test_stops$(EXE) tests/test_kv_alloc$(EXE) tests/test_i4_acc512$(EXE) tests/test_compat_direct$(EXE) tests/test_pipe_block$(EXE)
 ifneq (,$(LINUX))
 TEST_BINS += tests/test_uring$(EXE)
 endif
@@ -329,6 +329,9 @@ tests/test_compat_direct$(EXE): tests/test_compat_direct.c compat.h
 tests/test_uring$(EXE): tests/test_uring.c glm.c st.h uring.h json.h tok.h tok_unicode.h compat.h grammar.h tier.h
 	$(CC) $(CFLAGS) $< -o $@ $(LDFLAGS)
 
+tests/test_pipe_block$(EXE): tests/test_pipe_block.c glm.c st.h uring.h json.h tok.h tok_unicode.h compat.h grammar.h tier.h
+	$(CC) $(CFLAGS) $< -o $@ $(LDFLAGS)
+
 test-c: $(TEST_BINS)
 	$(PYTHON) tools/run_tests.py $(TEST_BINS)
 

diff --git a/c/glm.c b/c/glm.c
@@ -2146,13 +2146,30 @@ static int g_pipe=0;      /* PIPE=1: async expert-load pipeline. Default ON for
                            * the matmul. PIPE=0 opts back into the blocking serial path. */
 static int g_pipe_nw=8;   /* PIPE_WORKERS=n: I/O worker threads (disk-parallel reads) */
 static int g_uring=0;     /* URING=1: Linux io_uring load/completion backend; implies PIPE */
+static int g_pipe_block=0;/* COLI_PIPE_BLOCK=1: pipe_wait blocca su una condvar invece dello
+                           * spin sched_yield (default OFF = spin byte-identico). EN: a yield
+                           * storm on the main thread fights the OpenMP team for cycles during
+                           * multi-ms loads; the condvar wake costs ~5us against reads that
+                           * cost 0.5-3ms (#159). Pthread pool only: the URING backend has no
+                           * waiter spin to replace. */
+/* PIPE_WORKERS>0 esplicito nell'env implica PIPE=1: dimensionare il pool
+ * dichiara l'intento di usarlo (una campagna intera l'ha impostato con la
+ * pipe spenta senza accorgersene). EN: fires ONLY when PIPE is unset in the
+ * env AND the platform default left the pipe off (on _WIN32 it already
+ * defaults to 1) AND PIPE_WORKERS parses positive — the internal default of
+ * 8 does not count, PIPE_WORKERS=0/empty does not, and an explicit PIPE=0
+ * always wins. */
+static int pipe_workers_imply_pipe(const char *pipe_env, const char *pw_env, int pipe_now){
+    return !pipe_now && !pipe_env && pw_env && atoi(pw_env)>0;
+}
 typedef struct {
     _Atomic uint64_t cur;                         /* (gen<<8)|index; gen main-only, index 0..njobs (≤64) */
     _Atomic int njobs;                            /* current batch job count */
     _Atomic int eids[64];                         /* current batch expert ids */
     _Atomic int layer;                            /* current batch layer */
     _Atomic int ready[64];                        /* per-slot load-done flag */
     pthread_mutex_t mx; pthread_cond_t cv;        /* ONLY for parking/waking idle workers */
+    pthread_cond_t cv_done;                       /* COLI_PIPE_BLOCK: signals ready[] transitions */
     Model *m;
     pthread_t th[16]; int nw; int started;
 } PipePool;
@@ -2177,6 +2194,11 @@ static void *pipe_worker(void *arg){
                 int eid=atomic_load_explicit(&p->eids[i],memory_order_relaxed); /* AFTER winning CAS */
                 expert_load(p->m,L,eid,&p->m->ws[i],1);  /* needed-now load: fatal on I/O error (matches serial path) */
                 atomic_store_explicit(&p->ready[i],1,memory_order_release);
+                if(g_pipe_block){                     /* wake a main thread parked in pipe_wait */
+                    pthread_mutex_lock(&p->mx);
+                    pthread_cond_broadcast(&p->cv_done);
+                    pthread_mutex_unlock(&p->mx);
+                }
             }
             /* CAS failed → another worker advanced index (or gen advanced): re-loop */
         }
@@ -2194,6 +2216,7 @@ static void pipe_init(Model *m){
     g_pp.m=m; g_pp.nw=g_pipe_nw; if(g_pp.nw>16) g_pp.nw=16; if(g_pp.nw<1) g_pp.nw=1;
     atomic_store(&g_pp.cur,0); atomic_store(&g_pp.njobs,0);
     pthread_mutex_init(&g_pp.mx,NULL); pthread_cond_init(&g_pp.cv,NULL);
+    pthread_cond_init(&g_pp.cv_done,NULL);
     for(int i=0;i<g_pp.nw;i++) pthread_create(&g_pp.th[i],NULL,pipe_worker,NULL);
     g_pp.started=1;
 }
@@ -2228,6 +2251,17 @@ static inline void pipe_wait(int q){
         return;
     }
 #endif
+    if(g_pipe_block){
+        /* Fast path senza lock; poi ri-verifica SOTTO il lock prima di ogni
+         * wait. EN: the worker stores ready (release) BEFORE it takes mx to
+         * broadcast, so a set flag can never be missed (no lost wakeup). */
+        if(atomic_load_explicit(&g_pp.ready[q],memory_order_acquire)) return;
+        pthread_mutex_lock(&g_pp.mx);
+        while(!atomic_load_explicit(&g_pp.ready[q],memory_order_acquire))
+            pthread_cond_wait(&g_pp.cv_done,&g_pp.mx);
+        pthread_mutex_unlock(&g_pp.mx);
+        return;
+    }
     while(!atomic_load_explicit(&g_pp.ready[q],memory_order_acquire)) sched_yield();
 }
 
@@ -6108,8 +6142,13 @@ int main(int argc, char **argv){
         0
 #endif
         ;
+    if(pipe_workers_imply_pipe(getenv("PIPE"),getenv("PIPE_WORKERS"),g_pipe)){
+        g_pipe=1;
+        fprintf(stderr,"[PIPE] PIPE_WORKERS is set — enabling the async pipe (PIPE=1 implied; set PIPE=0 to override)\n");
+    }
     g_pipe_nw = getenv("PIPE_WORKERS")?atoi(getenv("PIPE_WORKERS")):8; /* I/O worker threads */
     if(g_pipe_nw<1) g_pipe_nw=1;
+    g_pipe_block = getenv("COLI_PIPE_BLOCK")?atoi(getenv("COLI_PIPE_BLOCK")):0; /* blocking pipe_wait (default: spin) */
     g_direct = getenv("DIRECT")?atoi(getenv("DIRECT")):0;
     g_uring = getenv("URING")?atoi(getenv("URING")):0;
     if(g_uring){

diff --git a/c/tests/test_pipe_block.c b/c/tests/test_pipe_block.c
@@ -0,0 +1,169 @@
+/* COLI_PIPE_BLOCK: the pipe pool's condvar waiter must be observably
+ * equivalent to the sched_yield spin it replaces — same bytes land in the
+ * same ws[] slots, and no interleaving loses a wakeup (the worker RELEASE-
+ * stores ready[] BEFORE taking mx to broadcast; the waiter re-checks under
+ * the lock, so a flag set between its fast-path check and the wait cannot
+ * be missed). Both waiters are exercised against the same on-disk fixture,
+ * alternating parked waits (wait issued before the load finishes) with
+ * fast-path waits (load already done), across enough generations to cycle
+ * the pool's gen-tagged cursor.
+ *
+ * Also pins the PIPE_WORKERS => PIPE implication table: fires ONLY when
+ * PIPE is unset in the env AND the platform default left the pipe off AND
+ * PIPE_WORKERS parses positive (PIPE_WORKERS=0/empty/negative must NOT
+ * silently enable a clamped 1-worker pipe). */
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#define main coli_glm_main_unused
+#include "../glm.c"
+#undef main
+
+static int fail(const char *s){ fprintf(stderr,"FAIL: %s\n",s); return 1; }
+
+enum { NE=8, LAYER=1 };            /* experts 0..NE-1 on one MoE layer */
+/* per-expert file image: [gate 12][up 12][down 12][gate.qs 12][up.qs 12][down.qs 16] */
+enum { WB=12, QS_G=12, QS_U=12, QS_D=16, ESZ=3*WB+QS_G+QS_U+QS_D };
+
+static unsigned char wbyte(int e,int j){ return (unsigned char)(e*31+j+1); }
+static float scale(int e,int i){ return (float)(e*8+i)+0.5f; }
+
+#define TMPF "test_pipe_block.tmp"
+
+static int write_fixture(void){
+    FILE *w=fopen(TMPF,"wb"); if(!w) return fail("create temp");
+    for(int e=0;e<NE;e++){
+        unsigned char img[ESZ];
+        for(int j=0;j<3*WB;j++) img[j]=wbyte(e,j);
+        float sc[(QS_G+QS_U+QS_D)/4];
+        for(int i=0;i<(int)(sizeof(sc)/sizeof(sc[0]));i++) sc[i]=scale(e,i);
+        memcpy(img+3*WB,sc,sizeof(sc));
+        if(fwrite(img,1,ESZ,w)!=ESZ){ fclose(w); return fail("expert fixture write"); }
+    }
+    fclose(w);
+    return 0;
+}
+
+static int build_fixture(Model *m,int fd){
+    m->c.hidden=4; m->c.moe_inter=3; m->ebits=8;
+    m->S.n=NE*6; m->S.cap=NE*6; m->S.t=calloc(NE*6,sizeof(st_tensor));
+    if(!m->S.t) return fail("tensor metadata allocation");
+    const char *proj[3]={"gate_proj","up_proj","down_proj"};
+    int sbytes[3]={QS_G,QS_U,QS_D};
+    for(int e=0;e<NE;e++){
+        int64_t wo=(int64_t)e*ESZ, so=wo+3*WB;
+        for(int k=0;k<3;k++){
+            char name[300];
+            snprintf(name,sizeof(name),"model.layers.%d.mlp.experts.%d.%s.weight",LAYER,e,proj[k]);
+            m->S.t[e*6+k]=(st_tensor){strdup(name),fd,wo,WB,3,WB}; wo+=WB;
+            size_t n=strlen(name); memcpy(name+n,".qs",4);
+            m->S.t[e*6+3+k]=(st_tensor){strdup(name),fd,so,sbytes[k],2,sbytes[k]/4}; so+=sbytes[k];
+        }
+    }
+    return 0;
+}
+
+static int check_slot(ESlot *s,int e){
+    if(s->eid!=e || s->g.fmt!=1 || s->u.fmt!=1 || s->d.fmt!=1){
+        fprintf(stderr,"  slot: eid=%d (want %d) fmt g/u/d=%d/%d/%d (want 1/1/1)\n",
+                s->eid,e,s->g.fmt,s->u.fmt,s->d.fmt);
+        return 1;
+    }
+    const unsigned char *g=(const unsigned char*)s->g.q8,
+                        *u=(const unsigned char*)s->u.q8,
+                        *d=(const unsigned char*)s->d.q8;   /* q8 is int8_t; compare raw bytes */
+    for(int j=0;j<WB;j++)
+        if(g[j]!=wbyte(e,j) || u[j]!=wbyte(e,WB+j) || d[j]!=wbyte(e,2*WB+j)){
+            fprintf(stderr,"  slot e=%d weight byte %d: g=%d/%d u=%d/%d d=%d/%d (got/want)\n",e,j,
+                    g[j],wbyte(e,j),u[j],wbyte(e,WB+j),d[j],wbyte(e,2*WB+j));
+            return 1;
+        }
+    for(int i=0;i<3;i++)
+        if(s->g.s[i]!=scale(e,i) || s->u.s[i]!=scale(e,3+i)){
+            fprintf(stderr,"  slot e=%d scale %d: g=%g/%g u=%g/%g (got/want)\n",e,i,
+                    (double)s->g.s[i],(double)scale(e,i),(double)s->u.s[i],(double)scale(e,3+i));
+            return 1;
+        }
+    for(int i=0;i<4;i++)
+        if(s->d.s[i]!=scale(e,6+i)){
+            fprintf(stderr,"  slot e=%d scale %d: d=%g/%g (got/want)\n",e,i,
+                    (double)s->d.s[i],(double)scale(e,6+i));
+            return 1;
+        }
+    return 0;
+}
+
+static int run_generations(Model *m,int block,int gens){
+    g_pipe_block=block;
+    for(int gen=0;gen<gens;gen++){
+        int eids[NE];
+        for(int q=0;q<NE;q++) eids[q]=(gen*3+q)%NE;   /* deterministic shuffle across gens */
+        pipe_dispatch(m,LAYER,eids,NE);
+        if(gen%4==0) usleep(300);                     /* let loads finish → fast-path wait */
+        for(int i=0;i<NE;i++){
+            /* odd gens wait on the LAST-dispatched slot first: with jobs this
+             * small, in-order waits mostly find ready already set — reverse
+             * order is what actually parks the waiter on the condvar. */
+            int q=(gen&1)?NE-1-i:i;
+            pipe_wait(q);
+            if(!atomic_load_explicit(&g_pp.ready[q],memory_order_acquire))
+                return fail(block?"blocking wait returned before ready":"spin wait returned before ready");
+            if(check_slot(&m->ws[q],eids[q])) return fail(block?"slot contents (block)":"slot contents (spin)");
+        }
+    }
+    return 0;
+}
+
+static int test_implication_table(void){
+    struct { const char *pipe_env,*pw_env; int pipe_now,want; } T[]={
+        {NULL,"4",0,1},   /* pool sized, pipe off, PIPE unset → imply */
+        {NULL,"16",0,1},
+        {NULL,"0",0,0},   /* PIPE_WORKERS=0 must NOT enable a clamped pipe */
+        {NULL,"",0,0},
+        {NULL,"-3",0,0},
+        {"0","4",0,0},    /* explicit PIPE=0 always wins */
+        {"1","4",1,0},    /* explicit PIPE=1: nothing to imply */
+        {NULL,"4",1,0},   /* platform default already ON (win32) */
+        {NULL,NULL,0,0},
+    };
+    for(size_t i=0;i<sizeof(T)/sizeof(T[0]);i++)
+        if(pipe_workers_imply_pipe(T[i].pipe_env,T[i].pw_env,T[i].pipe_now)!=T[i].want){
+            fprintf(stderr,"FAIL: implication row %zu (PIPE=%s PIPE_WORKERS=%s pipe_now=%d)\n",
+                    i,T[i].pipe_env?T[i].pipe_env:"<unset>",T[i].pw_env?T[i].pw_env:"<unset>",T[i].pipe_now);
+            return 1;
+        }
+    return 0;
+}
+
+int main(void){
+    if(test_implication_table()) return 1;
+
+    /* Relative to the CWD, like test_compat_direct's TMPF — NOT "/tmp/...":
+     * the windows job builds native .exe files and "/tmp" is not a Windows
+     * path. fwrite then reopen read-only: Windows compat has pread, not pwrite. */
+    if(write_fixture()) return 1;
+    int fd=open(TMPF,COMPAT_O_RDONLY);
+    if(fd<0) return fail("open temp");
+
+    static Model m;                                   /* zeroed: buffered pread path, no mmap/cuda */
+    if(build_fixture(&m,fd)){ close(fd); remove(TMPF); return 1; }
+
+    g_pipe=1; g_pipe_nw=4;
+    pipe_init(&m);
+
+    /* spin waiter first (control), then the condvar waiter under the same
+     * dispatch pattern; 200 generations each cycles the gen-tagged cursor
+     * and alternates parked/fast-path waits. */
+    if(run_generations(&m,0,200) || run_generations(&m,1,200)){ close(fd); remove(TMPF); return 1; }
+
+    for(int q=0;q<NE;q++){ compat_aligned_free(m.ws[q].slab); free(m.ws[q].fslab); }
+    for(int i=0;i<m.S.n;i++) free(m.S.t[i].name);
+    free(m.S.t);
+    close(fd);
+    remove(TMPF);
+    puts("test_pipe_block: ok");
+    return 0;
+}