use async and fix workers
Signed-off-by: Robert Landers <landers.robert@gmail.com>
This commit is contained in:
@@ -5259,6 +5259,7 @@ int main(int argc, char **argv){
|
||||
g_pipe = getenv("PIPE")?atoi(getenv("PIPE")):0; /* default OFF: overlap expert load ‖ matmul (byte-identical; reorders I/O). PIPE=1 opts in */
|
||||
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_direct = getenv("DIRECT")?atoi(getenv("DIRECT")):0;
|
||||
g_uring = getenv("URING")?atoi(getenv("URING")):0;
|
||||
if(g_uring){
|
||||
#ifdef __linux__
|
||||
@@ -5267,13 +5268,17 @@ int main(int argc, char **argv){
|
||||
if(uring_batch_init(&g_ub_pipe) || (g_pilot_real&&uring_batch_init(&g_ub_pilot))){
|
||||
fprintf(stderr,"URING=1: io_uring_setup failed: %s\n",strerror(errno)); return 2;
|
||||
}
|
||||
fprintf(stderr,"[URING] queued expert I/O active (depth=%d%s)\n",URING_REQ_MAX,
|
||||
g_pilot_real?", batched PILOT_REAL":"");
|
||||
unsigned uw=(unsigned)(g_pipe_nw>64?64:g_pipe_nw);
|
||||
if(coli_uring_set_workers(&g_ub_pipe.ring,uw) ||
|
||||
(g_pilot_real&&coli_uring_set_workers(&g_ub_pilot.ring,uw)))
|
||||
fprintf(stderr,"[URING] warning: cannot set io-wq workers=%u: %s\n",uw,strerror(errno));
|
||||
fprintf(stderr,"[URING] queued expert I/O active (depth=%d, workers=%u, %s%s)\n",URING_REQ_MAX,uw,
|
||||
g_direct?"O_DIRECT":"buffered",g_pilot_real?", batched PILOT_REAL":"");
|
||||
if(!g_direct) fprintf(stderr,"[URING] cold NVMe: DIRECT=1 avoids page-cache copy/readahead bottlenecks\n");
|
||||
#else
|
||||
fprintf(stderr,"URING=1 is supported only on Linux\n"); return 2;
|
||||
#endif
|
||||
}
|
||||
g_direct = getenv("DIRECT")?atoi(getenv("DIRECT")):0;
|
||||
g_idot = getenv("IDOT")?atoi(getenv("IDOT")):1; /* 0 = kernel f32 esatti (A/B) */
|
||||
if(getenv("ROUTE_TRACE")&&*getenv("ROUTE_TRACE")){
|
||||
g_route_fp=fopen(getenv("ROUTE_TRACE"),"w");
|
||||
|
||||
@@ -84,6 +84,9 @@ int main(void){
|
||||
}
|
||||
close(fd); return fail("io_uring_setup");
|
||||
}
|
||||
if(coli_uring_set_workers(&ring,4)){
|
||||
coli_uring_close(&ring); close(fd); return fail("io-wq worker limit");
|
||||
}
|
||||
for(int i=0;i<N;i++) if(coli_uring_prep_read(&ring,fd,dst[i],SZ,(off_t)i*SZ,(uint64_t)i+1)){
|
||||
coli_uring_close(&ring); close(fd); return fail("prepare read");
|
||||
}
|
||||
|
||||
@@ -81,6 +81,11 @@ static inline int coli_uring_init(ColiUring *r,unsigned entries){
|
||||
return 0;
|
||||
}
|
||||
|
||||
static inline int coli_uring_set_workers(ColiUring *r,unsigned workers){
|
||||
unsigned limits[2]={workers,workers}; /* bounded, unbounded io-wq workers */
|
||||
return (int)syscall(SYS_io_uring_register,r->fd,IORING_REGISTER_IOWQ_MAX_WORKERS,limits,2);
|
||||
}
|
||||
|
||||
static inline int coli_uring_prep_read(ColiUring *r,int fd,void *buf,size_t len,
|
||||
int64_t off,uint64_t user_data){
|
||||
if(!len || len>UINT32_MAX){ errno=EINVAL; return -1; }
|
||||
@@ -91,6 +96,12 @@ static inline int coli_uring_prep_read(ColiUring *r,int fd,void *buf,size_t len,
|
||||
struct io_uring_sqe *sqe=&r->sqes[idx];
|
||||
memset(sqe,0,sizeof(*sqe));
|
||||
sqe->opcode=IORING_OP_READ;
|
||||
/* Cold regular-file reads are allowed to execute inline during
|
||||
* io_uring_enter() unless forced async. That serializes the submitter on
|
||||
* filesystems without native nonblocking buffered reads and destroys the
|
||||
* intended I/O/compute overlap. io-wq gives the ring a real bounded worker
|
||||
* pool while CQEs retain completion ordering/ownership here. */
|
||||
sqe->flags=IOSQE_ASYNC;
|
||||
sqe->fd=fd;
|
||||
sqe->off=(uint64_t)off;
|
||||
sqe->addr=(uint64_t)(uintptr_t)buf;
|
||||
|
||||
Reference in New Issue
Block a user