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

Merge pull request #270 from ebootheee/pr-pipe-batch-sync

Browse Source 
pipe: blocking pipe_wait (COLI_PIPE_BLOCK=1) + PIPE_WORKERS implies PIPE=1
This commit is contained in:
Vincenzo Fornaro
2026-07-20 17:52:42 +02:00
committed by GitHub
parent d3362303f5 d34461c8d8
commit e809171e62
3 changed files with 212 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
c/Makefile
+4 -1
View File
@@ -171,7 +171,7 @@ else
PYTHON ?= python3
endif
CUDA_OBJ =
TEST_BINS = tests/test_json$(EXE) tests/test_st$(EXE) tests/test_st_pread$(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_topp$(EXE) tests/test_sample_nan$(EXE) tests/test_tok_o200k$(EXE) tests/test_kv_alloc$(EXE) tests/test_i4_acc512$(EXE) tests/test_compat_direct$(EXE) tests/test_dsa_select$(EXE) tests/test_logit_nan$(EXE)
TEST_BINS = tests/test_json$(EXE) tests/test_st$(EXE) tests/test_st_pread$(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_topp$(EXE) tests/test_sample_nan$(EXE) tests/test_tok_o200k$(EXE) tests/test_kv_alloc$(EXE) tests/test_i4_acc512$(EXE) tests/test_compat_direct$(EXE) tests/test_dsa_select$(EXE) tests/test_logit_nan$(EXE) tests/test_pipe_block$(EXE)
ifneq (,$(LINUX))
TEST_BINS += tests/test_uring$(EXE)
endif
@@ -363,6 +363,9 @@ tests/bench_dsa_select$(EXE): tests/bench_dsa_select.c colibri.c st.h uring.h js
tests/test_uring$(EXE): tests/test_uring.c colibri.c st.h uring.h json.h tok.h tok_unicode.h compat.h grammar.h tier.h quant.h sample.h kv_persist.h telemetry.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)
c/colibri.c
+39
View File
@@ -1589,6 +1589,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 */
@@ -1596,6 +1612,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;
@@ -1620,6 +1637,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,1); /* needed-now load: fatal on I/O error (matches serial path); demand=1: this IS moe()'s own miss 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 */
}
@@ -1637,6 +1659,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;
}
@@ -1671,6 +1694,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();
}
@@ -5419,8 +5453,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;
{ const char *dh=getenv("COLI_DISKCLASS_WINDOW"); /* DISK-CLASS recency window, see its declaration */
if(dh){ g_direct_heat_ticks=(uint32_t)strtoul(dh,NULL,10); g_direct_heat_explicit=1; } }
c/tests/test_pipe_block.c
+169
View File
@@ -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;
}