diff --git a/c/Makefile b/c/Makefile index 27902cb2..db94a9a3 100644 --- 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 index 645be12b..6bd66f40 100644 --- a/c/glm.c +++ b/c/glm.c @@ -2146,6 +2146,22 @@ 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 */ @@ -2153,6 +2169,7 @@ typedef struct { _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 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 +#include +#include +#include +#include +#include +#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;ec.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;eS.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;jg.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;genws[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",T[i].pw_env?T[i].pw_env:"",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