Changes in libcfa/src/concurrency/io.cfa [61dd73d:3f7d0b4]
- File:
-
- 1 edited
-
libcfa/src/concurrency/io.cfa (modified) (22 diffs)
Legend:
- Unmodified
- Added
- Removed
-
libcfa/src/concurrency/io.cfa
r61dd73d r3f7d0b4 20 20 21 21 #if !defined(HAVE_LINUX_IO_URING_H) 22 void __kernel_io_startup( cluster & ) {22 void __kernel_io_startup( cluster &, bool ) { 23 23 // Nothing to do without io_uring 24 24 } 25 25 26 void __kernel_io_ start_thrd( cluster & ) {26 void __kernel_io_finish_start( cluster & ) { 27 27 // Nothing to do without io_uring 28 28 } 29 29 30 void __kernel_io_ stop_thrd( cluster & ) {30 void __kernel_io_prepare_stop( cluster & ) { 31 31 // Nothing to do without io_uring 32 32 } 33 33 34 void __kernel_io_shutdown( cluster & ) {34 void __kernel_io_shutdown( cluster &, bool ) { 35 35 // Nothing to do without io_uring 36 36 } … … 86 86 #endif 87 87 88 // Fast poller user-thread 89 // Not using the "thread" keyword because we want to control 90 // more carefully when to start/stop it 91 struct __io_poller_fast { 92 struct __io_data * ring; 93 bool waiting; 94 $thread thrd; 95 }; 96 97 void ?{}( __io_poller_fast & this, struct cluster & cltr ) { 98 this.ring = cltr.io; 99 this.waiting = true; 100 (this.thrd){ "Fast I/O Poller", cltr }; 101 } 102 void ^?{}( __io_poller_fast & mutex this ); 103 void main( __io_poller_fast & this ); 104 static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; } 105 void ^?{}( __io_poller_fast & mutex this ) {} 106 107 struct __submition_data { 108 // Head and tail of the ring (associated with array) 109 volatile uint32_t * head; 110 volatile uint32_t * tail; 111 112 // The actual kernel ring which uses head/tail 113 // indexes into the sqes arrays 114 uint32_t * array; 115 116 // number of entries and mask to go with it 117 const uint32_t * num; 118 const uint32_t * mask; 119 120 // Submission flags (Not sure what for) 121 uint32_t * flags; 122 123 // number of sqes not submitted (whatever that means) 124 uint32_t * dropped; 125 126 // Like head/tail but not seen by the kernel 127 volatile uint32_t alloc; 128 volatile uint32_t ready; 129 130 __spinlock_t lock; 131 132 // A buffer of sqes (not the actual ring) 133 struct io_uring_sqe * sqes; 134 135 // The location and size of the mmaped area 136 void * ring_ptr; 137 size_t ring_sz; 138 139 // Statistics 140 #if !defined(__CFA_NO_STATISTICS__) 141 struct { 142 struct { 143 unsigned long long int val; 144 unsigned long long int cnt; 145 } submit_avg; 146 } stats; 147 #endif 148 }; 149 150 struct __completion_data { 151 // Head and tail of the ring 152 volatile uint32_t * head; 153 volatile uint32_t * tail; 154 155 // number of entries and mask to go with it 156 const uint32_t * mask; 157 const uint32_t * num; 158 159 // number of cqes not submitted (whatever that means) 160 uint32_t * overflow; 161 162 // the kernel ring 163 struct io_uring_cqe * cqes; 164 165 // The location and size of the mmaped area 166 void * ring_ptr; 167 size_t ring_sz; 168 169 // Statistics 170 #if !defined(__CFA_NO_STATISTICS__) 171 struct { 172 struct { 173 unsigned long long int val; 174 unsigned long long int slow_cnt; 175 unsigned long long int fast_cnt; 176 } completed_avg; 177 } stats; 178 #endif 179 }; 180 181 struct __io_data { 182 struct __submition_data submit_q; 183 struct __completion_data completion_q; 184 uint32_t flags; 185 int fd; 186 semaphore submit; 187 volatile bool done; 188 struct { 189 struct { 190 void * stack; 191 pthread_t kthrd; 192 } slow; 193 __io_poller_fast fast; 194 __bin_sem_t sem; 195 } poller; 196 }; 88 #if defined(__CFA_IO_POLLING_USER__) 89 void ?{}( __io_poller_fast & this, struct cluster & cltr ) { 90 this.ring = &cltr.io; 91 (this.thrd){ "Fast I/O Poller", cltr }; 92 } 93 void ^?{}( __io_poller_fast & mutex this ); 94 void main( __io_poller_fast & this ); 95 static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; } 96 void ^?{}( __io_poller_fast & mutex this ) {} 97 #endif 197 98 198 99 //============================================================================================= … … 200 101 //============================================================================================= 201 102 void __kernel_io_startup( cluster & this, bool main_cluster ) { 202 this.io = malloc();203 204 103 // Step 1 : call to setup 205 104 struct io_uring_params params; … … 214 113 215 114 // Step 2 : mmap result 216 memset( this.io, 0, sizeof(struct __io_data));217 struct __submition_data & sq = this.io->submit_q;218 struct __completion_data & cq = this.io->completion_q;115 memset(&this.io, 0, sizeof(struct io_ring)); 116 struct io_uring_sq & sq = this.io.submit_q; 117 struct io_uring_cq & cq = this.io.completion_q; 219 118 220 119 // calculate the right ring size … … 294 193 295 194 // Update the global ring info 296 this.io ->flags = params.flags;297 this.io ->fd = fd;298 this.io ->done = false;299 (this.io ->submit){ min(*sq.num, *cq.num) };195 this.io.flags = params.flags; 196 this.io.fd = fd; 197 this.io.done = false; 198 (this.io.submit){ min(*sq.num, *cq.num) }; 300 199 301 200 // Initialize statistics 302 201 #if !defined(__CFA_NO_STATISTICS__) 303 this.io->submit_q.stats.submit_avg.val = 0; 304 this.io->submit_q.stats.submit_avg.cnt = 0; 305 this.io->completion_q.stats.completed_avg.val = 0; 306 this.io->completion_q.stats.completed_avg.slow_cnt = 0; 307 this.io->completion_q.stats.completed_avg.fast_cnt = 0; 202 this.io.submit_q.stats.submit_avg.val = 0; 203 this.io.submit_q.stats.submit_avg.cnt = 0; 204 this.io.completion_q.stats.completed_avg.val = 0; 205 this.io.completion_q.stats.completed_avg.cnt = 0; 308 206 #endif 309 207 … … 314 212 315 213 void __kernel_io_finish_start( cluster & this ) { 316 __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this); 317 (this.io->poller.fast){ this }; 318 __thrd_start( this.io->poller.fast, main ); 214 #if defined(__CFA_IO_POLLING_USER__) 215 __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this); 216 (this.io.poller.fast){ this }; 217 __thrd_start( this.io.poller.fast, main ); 218 #endif 319 219 320 220 // Create the poller thread 321 221 __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this); 322 this.io ->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );222 this.io.poller.slow.stack = __create_pthread( &this.io.poller.slow.kthrd, __io_poller_slow, &this ); 323 223 } 324 224 … … 326 226 __cfadbg_print_safe(io_core, "Kernel I/O : Stopping pollers for cluster\n", &this); 327 227 // Notify the poller thread of the shutdown 328 __atomic_store_n(&this.io ->done, true, __ATOMIC_SEQ_CST);228 __atomic_store_n(&this.io.done, true, __ATOMIC_SEQ_CST); 329 229 330 230 // Stop the IO Poller 331 231 sigval val = { 1 }; 332 pthread_sigqueue( this.io->poller.slow.kthrd, SIGUSR1, val ); 333 post( this.io->poller.sem ); 232 pthread_sigqueue( this.io.poller.slow.kthrd, SIGUSR1, val ); 233 #if defined(__CFA_IO_POLLING_USER__) 234 post( this.io.poller.sem ); 235 #endif 334 236 335 237 // Wait for the poller thread to finish 336 pthread_join( this.io ->poller.slow.kthrd, 0p );337 free( this.io ->poller.slow.stack );238 pthread_join( this.io.poller.slow.kthrd, 0p ); 239 free( this.io.poller.slow.stack ); 338 240 339 241 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this); 340 242 341 243 #if defined(__CFA_IO_POLLING_USER__) 342 verify( this.io->poller.fast.waiting );343 verify( this.io->poller.fast.thrd.state == Blocked );344 345 this.io->poller.fast.thrd.curr_cluster = mainCluster;346 347 244 // unpark the fast io_poller 348 unpark( &this.io ->poller.fast.thrd __cfaabi_dbg_ctx2 );349 350 ^(this.io ->poller.fast){};245 unpark( &this.io.poller.fast.thrd __cfaabi_dbg_ctx2 ); 246 247 ^(this.io.poller.fast){}; 351 248 352 249 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this); … … 362 259 #if !defined(__CFA_NO_STATISTICS__) 363 260 if(this.print_stats) { 364 with(this.io->submit_q.stats, this.io->completion_q.stats) { 365 __cfaabi_bits_print_safe( STDERR_FILENO, 366 "----- I/O uRing Stats -----\n" 367 "- total submit calls : %llu\n" 368 "- avg submit : %lf\n" 369 "- total wait calls : %llu (%llu slow, %llu fast)\n" 370 "- avg completion/wait : %lf\n", 371 submit_avg.cnt, 372 ((double)submit_avg.val) / submit_avg.cnt, 373 completed_avg.slow_cnt + completed_avg.fast_cnt, 374 completed_avg.slow_cnt, completed_avg.fast_cnt, 375 ((double)completed_avg.val) / (completed_avg.slow_cnt + completed_avg.fast_cnt) 376 ); 377 } 261 __cfaabi_bits_print_safe( STDERR_FILENO, 262 "----- I/O uRing Stats -----\n" 263 "- total submit calls : %llu\n" 264 "- avg submit : %lf\n" 265 "- total wait calls : %llu\n" 266 "- avg completion/wait : %lf\n", 267 this.io.submit_q.stats.submit_avg.cnt, 268 ((double)this.io.submit_q.stats.submit_avg.val) / this.io.submit_q.stats.submit_avg.cnt, 269 this.io.completion_q.stats.completed_avg.cnt, 270 ((double)this.io.completion_q.stats.completed_avg.val) / this.io.completion_q.stats.completed_avg.cnt 271 ); 378 272 } 379 273 #endif 380 274 381 275 // Shutdown the io rings 382 struct __submition_data & sq = this.io->submit_q;383 struct __completion_data & cq = this.io->completion_q;276 struct io_uring_sq & sq = this.io.submit_q; 277 struct io_uring_cq & cq = this.io.completion_q; 384 278 385 279 // unmap the submit queue entries … … 395 289 396 290 // close the file descriptor 397 close(this.io->fd); 398 399 free( this.io ); 291 close(this.io.fd); 400 292 } 401 293 … … 410 302 // Process a single completion message from the io_uring 411 303 // This is NOT thread-safe 412 static int __drain_io( struct __io_data& ring, sigset_t * mask, int waitcnt, bool in_kernel ) {304 static int __drain_io( struct io_ring & ring, sigset_t * mask, int waitcnt, bool in_kernel ) { 413 305 int ret = syscall( __NR_io_uring_enter, ring.fd, 0, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8); 414 306 if( ret < 0 ) { … … 428 320 // Nothing was new return 0 429 321 if (head == tail) { 322 #if !defined(__CFA_NO_STATISTICS__) 323 ring.completion_q.stats.completed_avg.cnt += 1; 324 #endif 430 325 return 0; 431 326 } … … 453 348 __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED ); 454 349 350 // Update statistics 351 #if !defined(__CFA_NO_STATISTICS__) 352 ring.completion_q.stats.completed_avg.val += count; 353 ring.completion_q.stats.completed_avg.cnt += 1; 354 #endif 355 455 356 return count; 456 357 } … … 458 359 static void * __io_poller_slow( void * arg ) { 459 360 cluster * cltr = (cluster *)arg; 460 struct __io_data & ring = *cltr->io;361 struct io_ring & ring = cltr->io; 461 362 462 363 sigset_t mask; … … 471 372 verify( (*ring.completion_q.head) == (*ring.completion_q.tail) ); 472 373 473 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);474 475 374 while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) { 476 375 #if defined(__CFA_IO_POLLING_USER__) … … 479 378 // batton pass to the user-thread 480 379 int count = __drain_io( ring, &mask, 1, true ); 481 482 // Update statistics483 #if !defined(__CFA_NO_STATISTICS__)484 ring.completion_q.stats.completed_avg.val += count;485 ring.completion_q.stats.completed_avg.slow_cnt += 1;486 #endif487 488 380 if(count > 0) { 489 381 __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to fast poller\n", &ring); … … 495 387 496 388 //In the naive approach, just poll the io completion queue directly 497 int count = __drain_io( ring, &mask, 1, true ); 498 499 // Update statistics 500 #if !defined(__CFA_NO_STATISTICS__) 501 ring.completion_q.stats.completed_avg.val += count; 502 ring.completion_q.stats.completed_avg.slow_cnt += 1; 503 #endif 389 __drain_io( ring, &mask, 1, true ); 504 390 505 391 #endif 506 392 } 507 393 508 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p stopping\n", &ring);509 510 394 return 0p; 511 395 } 512 396 513 void main( __io_poller_fast & this ) { 514 // Start parked 515 park( __cfaabi_dbg_ctx ); 516 517 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p ready\n", &this.ring); 518 519 // Then loop until we need to start 520 while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) { 521 // Drain the io 522 this.waiting = false; 523 int count = __drain_io( *this.ring, 0p, 0, false ); 524 525 // Update statistics 526 #if !defined(__CFA_NO_STATISTICS__) 527 this.ring->completion_q.stats.completed_avg.val += count; 528 this.ring->completion_q.stats.completed_avg.fast_cnt += 1; 529 #endif 530 531 this.waiting = true; 532 if(0 > count) { 533 // If we got something, just yield and check again 534 yield(); 397 #if defined(__CFA_IO_POLLING_USER__) 398 void main( __io_poller_fast & this ) { 399 // Start parked 400 park( __cfaabi_dbg_ctx ); 401 402 // Then loop until we need to start 403 while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) { 404 // Drain the io 405 if(0 > __drain_io( *this.ring, 0p, 0, false )) { 406 // If we got something, just yield and check again 407 yield(); 408 } 409 else { 410 // We didn't get anything baton pass to the slow poller 411 __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring); 412 post( this.ring->poller.sem ); 413 park( __cfaabi_dbg_ctx ); 414 } 535 415 } 536 else { 537 // We didn't get anything baton pass to the slow poller 538 __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring); 539 post( this.ring->poller.sem ); 540 park( __cfaabi_dbg_ctx ); 541 } 542 } 543 544 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring); 545 } 416 } 417 #endif 546 418 547 419 //============================================================================================= … … 573 445 // 574 446 575 static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data& ring ) {447 static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct io_ring & ring ) { 576 448 // Wait for a spot to be available 577 449 P(ring.submit); … … 591 463 } 592 464 593 static inline void __submit( struct __io_data& ring, uint32_t idx ) {465 static inline void __submit( struct io_ring & ring, uint32_t idx ) { 594 466 // get mutual exclusion 595 467 lock(ring.submit_q.lock __cfaabi_dbg_ctx2); … … 652 524 653 525 #define __submit_prelude \ 654 struct __io_data & ring = *active_cluster()->io; \526 struct io_ring & ring = active_cluster()->io; \ 655 527 struct io_uring_sqe * sqe; \ 656 528 uint32_t idx; \ … … 695 567 extern int close(int fd); 696 568 697 struct statx;698 extern int statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf);699 700 569 extern ssize_t read (int fd, void *buf, size_t count); 701 570 } … … 907 776 908 777 (*sqe){ IORING_OP_CLOSE, fd }; 909 910 __submit_wait911 #endif912 }913 914 int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf) {915 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_STATX)916 //return statx( dirfd, pathname, flags, mask, statxbuf );917 return syscall( __NR_statx, dirfd, pathname, flags, mask, statxbuf );918 #else919 __submit_prelude920 921 (*sqe){ IORING_OP_STATX, dirfd };922 sqe->addr = (uint64_t)pathname;923 sqe->statx_flags = flags;924 sqe->len = mask;925 sqe->off = (uint64_t)statxbuf;926 778 927 779 __submit_wait … … 1040 892 return IS_DEFINED(IORING_OP_CLOSE); 1041 893 1042 if( /*func == (fptr_t)statx || */1043 func == (fptr_t)cfa_statx )1044 #define _CFA_IO_FEATURE_IORING_OP_STATX ,1045 return IS_DEFINED(IORING_OP_STATX);1046 1047 894 if( /*func == (fptr_t)read || */ 1048 895 func == (fptr_t)cfa_read )
Note:
See TracChangeset
for help on using the changeset viewer.
