Changeset 6c53a93 for libcfa/src/concurrency
- Timestamp:
- Jan 5, 2022, 10:39:39 AM (4 years ago)
- Branches:
- ADT, ast-experimental, enum, forall-pointer-decay, master, pthread-emulation, qualifiedEnum
- Children:
- 0ac728b
- Parents:
- e2853eb (diff), 6111f1f (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the(diff)links above to see all the changes relative to each parent. - Location:
- libcfa/src/concurrency
- Files:
-
- 7 edited
-
io.cfa (modified) (8 diffs)
-
io/setup.cfa (modified) (3 diffs)
-
io/types.hfa (modified) (1 diff)
-
kernel.cfa (modified) (16 diffs)
-
kernel.hfa (modified) (2 diffs)
-
kernel/startup.cfa (modified) (3 diffs)
-
kernel_private.hfa (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
libcfa/src/concurrency/io.cfa
re2853eb r6c53a93 33 33 #include <sys/syscall.h> 34 34 #include <sys/eventfd.h> 35 #include <sys/uio.h> 35 36 36 37 #include <linux/io_uring.h> … … 133 134 } 134 135 135 void __cfa_io_flush( processor * proc) {136 bool __cfa_io_flush( processor * proc, int min_comp ) { 136 137 /* paranoid */ verify( ! __preemption_enabled() ); 137 138 /* paranoid */ verify( proc ); … … 141 142 $io_context & ctx = *proc->io.ctx; 142 143 143 // for(i; 2) {144 // unsigned idx = proc->rdq.id + i;145 // cltr->ready_queue.lanes.tscs[idx].tv = -1ull;146 // }147 148 144 __ioarbiter_flush( ctx ); 149 145 150 146 __STATS__( true, io.calls.flush++; ) 151 int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, 0,0, (sigset_t *)0p, _NSIG / 8);147 int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, min_comp > 0 ? IORING_ENTER_GETEVENTS : 0, (sigset_t *)0p, _NSIG / 8); 152 148 if( ret < 0 ) { 153 149 switch((int)errno) { … … 157 153 // Update statistics 158 154 __STATS__( false, io.calls.errors.busy ++; ) 159 // for(i; 2) { 160 // unsigned idx = proc->rdq.id + i; 161 // cltr->ready_queue.lanes.tscs[idx].tv = rdtscl(); 162 // } 163 return; 155 return false; 164 156 default: 165 157 abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) ); … … 182 174 183 175 ctx.proc->io.pending = false; 184 185 __cfa_io_drain( proc ); 186 // for(i; 2) { 187 // unsigned idx = proc->rdq.id + i; 188 // cltr->ready_queue.lanes.tscs[idx].tv = rdtscl(); 189 // } 176 ready_schedule_lock(); 177 bool ret = __cfa_io_drain( proc ); 178 ready_schedule_unlock(); 179 return ret; 190 180 } 191 181 … … 291 281 } 292 282 293 294 283 //============================================================================================= 295 284 // submission … … 314 303 ctx->proc->io.dirty = true; 315 304 if(sq.to_submit > 30 || !lazy) { 316 ready_schedule_lock(); 317 __cfa_io_flush( ctx->proc ); 318 ready_schedule_unlock(); 305 __cfa_io_flush( ctx->proc, 0 ); 319 306 } 320 307 } … … 515 502 } 516 503 } 504 505 #if defined(CFA_WITH_IO_URING_IDLE) 506 bool __kernel_read(processor * proc, io_future_t & future, iovec & iov, int fd) { 507 $io_context * ctx = proc->io.ctx; 508 /* paranoid */ verify( ! __preemption_enabled() ); 509 /* paranoid */ verify( proc == __cfaabi_tls.this_processor ); 510 /* paranoid */ verify( ctx ); 511 512 __u32 idx; 513 struct io_uring_sqe * sqe; 514 515 // We can proceed to the fast path 516 if( !__alloc(ctx, &idx, 1) ) return false; 517 518 // Allocation was successful 519 __fill( &sqe, 1, &idx, ctx ); 520 521 sqe->user_data = (uintptr_t)&future; 522 sqe->flags = 0; 523 sqe->fd = fd; 524 sqe->off = 0; 525 sqe->ioprio = 0; 526 sqe->fsync_flags = 0; 527 sqe->__pad2[0] = 0; 528 sqe->__pad2[1] = 0; 529 sqe->__pad2[2] = 0; 530 531 #if defined(CFA_HAVE_IORING_OP_READ) 532 sqe->opcode = IORING_OP_READ; 533 sqe->addr = (uint64_t)iov.iov_base; 534 sqe->len = iov.iov_len; 535 #elif defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV) 536 sqe->opcode = IORING_OP_READV; 537 sqe->addr = (uintptr_t)&iov; 538 sqe->len = 1; 539 #else 540 #error CFA_WITH_IO_URING_IDLE but none of CFA_HAVE_READV, CFA_HAVE_IORING_OP_READV or CFA_HAVE_IORING_OP_READ defined 541 #endif 542 543 asm volatile("": : :"memory"); 544 545 /* paranoid */ verify( sqe->user_data == (uintptr_t)&future ); 546 __submit( ctx, &idx, 1, true ); 547 548 /* paranoid */ verify( proc == __cfaabi_tls.this_processor ); 549 /* paranoid */ verify( ! __preemption_enabled() ); 550 } 551 #endif 517 552 #endif -
libcfa/src/concurrency/io/setup.cfa
re2853eb r6c53a93 32 32 33 33 void __cfa_io_start( processor * proc ) {} 34 void __cfa_io_flush( processor * proc) {}34 bool __cfa_io_flush( processor * proc, int ) {} 35 35 void __cfa_io_stop ( processor * proc ) {} 36 36 … … 111 111 this.ext_sq.empty = true; 112 112 (this.ext_sq.queue){}; 113 __io_uring_setup( this, cl.io.params, proc->idle );113 __io_uring_setup( this, cl.io.params, proc->idle_fd ); 114 114 __cfadbg_print_safe(io_core, "Kernel I/O : Created ring for io_context %u (%p)\n", this.fd, &this); 115 115 } … … 220 220 cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes); 221 221 222 // Step 4 : eventfd 223 // io_uring_register is so f*cking slow on some machine that it 224 // will never succeed if preemption isn't hard blocked 225 __cfadbg_print_safe(io_core, "Kernel I/O : registering %d for completion with ring %d\n", procfd, fd); 226 227 __disable_interrupts_hard(); 228 229 int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &procfd, 1); 230 if (ret < 0) { 231 abort("KERNEL ERROR: IO_URING EVENTFD REGISTER - %s\n", strerror(errno)); 232 } 233 234 __enable_interrupts_hard(); 235 236 __cfadbg_print_safe(io_core, "Kernel I/O : registered %d for completion with ring %d\n", procfd, fd); 222 #if !defined(CFA_WITH_IO_URING_IDLE) 223 // Step 4 : eventfd 224 // io_uring_register is so f*cking slow on some machine that it 225 // will never succeed if preemption isn't hard blocked 226 __cfadbg_print_safe(io_core, "Kernel I/O : registering %d for completion with ring %d\n", procfd, fd); 227 228 __disable_interrupts_hard(); 229 230 int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &procfd, 1); 231 if (ret < 0) { 232 abort("KERNEL ERROR: IO_URING EVENTFD REGISTER - %s\n", strerror(errno)); 233 } 234 235 __enable_interrupts_hard(); 236 237 __cfadbg_print_safe(io_core, "Kernel I/O : registered %d for completion with ring %d\n", procfd, fd); 238 #endif 237 239 238 240 // some paranoid checks -
libcfa/src/concurrency/io/types.hfa
re2853eb r6c53a93 185 185 186 186 // Wait for the future to be fulfilled 187 bool wait( io_future_t & this ) { 188 return wait(this.self); 189 } 190 191 void reset( io_future_t & this ) { 192 return reset(this.self); 193 } 187 bool wait ( io_future_t & this ) { return wait (this.self); } 188 void reset ( io_future_t & this ) { return reset (this.self); } 189 bool available( io_future_t & this ) { return available(this.self); } 194 190 } -
libcfa/src/concurrency/kernel.cfa
re2853eb r6c53a93 27 27 extern "C" { 28 28 #include <sys/eventfd.h> 29 #include <sys/uio.h> 29 30 } 30 31 … … 34 35 #include "strstream.hfa" 35 36 #include "device/cpu.hfa" 37 #include "io/types.hfa" 36 38 37 39 //Private includes … … 124 126 static void __wake_one(cluster * cltr); 125 127 126 static void mark_idle (__cluster_proc_list & idles, processor & proc); 128 static void idle_sleep(processor * proc, io_future_t & future, iovec & iov); 129 static bool mark_idle (__cluster_proc_list & idles, processor & proc); 127 130 static void mark_awake(__cluster_proc_list & idles, processor & proc); 128 static [unsigned idle, unsigned total, * processor] query_idles( & __cluster_proc_list idles );129 131 130 132 extern void __cfa_io_start( processor * ); 131 133 extern bool __cfa_io_drain( processor * ); 132 extern void __cfa_io_flush( processor *);134 extern bool __cfa_io_flush( processor *, int min_comp ); 133 135 extern void __cfa_io_stop ( processor * ); 134 136 static inline bool __maybe_io_drain( processor * ); 137 138 #if defined(CFA_WITH_IO_URING_IDLE) 139 extern bool __kernel_read(processor * proc, io_future_t & future, iovec &, int fd); 140 #endif 135 141 136 142 extern void __disable_interrupts_hard(); … … 148 154 /* paranoid */ verify( __preemption_enabled() ); 149 155 } 156 150 157 151 158 //============================================================================================= … … 163 170 verify(this); 164 171 172 io_future_t future; // used for idle sleep when io_uring is present 173 future.self.ptr = 1p; // mark it as already fulfilled so we know if there is a pending request or not 174 eventfd_t idle_val; 175 iovec idle_iovec = { &idle_val, sizeof(idle_val) }; 176 165 177 __cfa_io_start( this ); 166 178 … … 196 208 197 209 if( !readyThread ) { 198 ready_schedule_lock(); 199 __cfa_io_flush( this ); 200 ready_schedule_unlock(); 210 __cfa_io_flush( this, 0 ); 201 211 202 212 readyThread = __next_thread_slow( this->cltr ); … … 213 223 214 224 // Push self to idle stack 215 mark_idle(this->cltr->procs, * this);225 if(!mark_idle(this->cltr->procs, * this)) continue MAIN_LOOP; 216 226 217 227 // Confirm the ready-queue is empty … … 229 239 } 230 240 231 #if !defined(__CFA_NO_STATISTICS__) 232 if(this->print_halts) { 233 __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); 241 idle_sleep( this, future, idle_iovec ); 242 243 // We were woken up, remove self from idle 244 mark_awake(this->cltr->procs, * this); 245 246 // DON'T just proceed, start looking again 247 continue MAIN_LOOP; 248 } 249 250 /* paranoid */ verify( readyThread ); 251 252 // Reset io dirty bit 253 this->io.dirty = false; 254 255 // We found a thread run it 256 __run_thread(this, readyThread); 257 258 // Are we done? 259 if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP; 260 261 if(this->io.pending && !this->io.dirty) { 262 __cfa_io_flush( this, 0 ); 263 } 264 265 #else 266 #warning new kernel loop 267 SEARCH: { 268 /* paranoid */ verify( ! __preemption_enabled() ); 269 270 // First, lock the scheduler since we are searching for a thread 271 ready_schedule_lock(); 272 273 // Try to get the next thread 274 readyThread = pop_fast( this->cltr ); 275 if(readyThread) { ready_schedule_unlock(); break SEARCH; } 276 277 // If we can't find a thread, might as well flush any outstanding I/O 278 if(this->io.pending) { __cfa_io_flush( this, 0 ); } 279 280 // Spin a little on I/O, just in case 281 for(5) { 282 __maybe_io_drain( this ); 283 readyThread = pop_fast( this->cltr ); 284 if(readyThread) { ready_schedule_unlock(); break SEARCH; } 285 } 286 287 // no luck, try stealing a few times 288 for(5) { 289 if( __maybe_io_drain( this ) ) { 290 readyThread = pop_fast( this->cltr ); 291 } else { 292 readyThread = pop_slow( this->cltr ); 234 293 } 235 #endif 236 237 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle); 294 if(readyThread) { ready_schedule_unlock(); break SEARCH; } 295 } 296 297 // still no luck, search for a thread 298 readyThread = pop_search( this->cltr ); 299 if(readyThread) { ready_schedule_unlock(); break SEARCH; } 300 301 // Don't block if we are done 302 if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) { 303 ready_schedule_unlock(); 304 break MAIN_LOOP; 305 } 306 307 __STATS( __tls_stats()->ready.sleep.halts++; ) 308 309 // Push self to idle stack 310 ready_schedule_unlock(); 311 if(!mark_idle(this->cltr->procs, * this)) goto SEARCH; 312 ready_schedule_lock(); 313 314 // Confirm the ready-queue is empty 315 __maybe_io_drain( this ); 316 readyThread = pop_search( this->cltr ); 317 ready_schedule_unlock(); 318 319 if( readyThread ) { 320 // A thread was found, cancel the halt 321 mark_awake(this->cltr->procs, * this); 322 323 __STATS( __tls_stats()->ready.sleep.cancels++; ) 324 325 // continue the main loop 326 break SEARCH; 327 } 328 329 __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); ) 330 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle_fd); 238 331 239 332 { 240 333 eventfd_t val; 241 ssize_t ret = read( this->idle , &val, sizeof(val) );334 ssize_t ret = read( this->idle_fd, &val, sizeof(val) ); 242 335 if(ret < 0) { 243 336 switch((int)errno) { … … 255 348 } 256 349 257 #if !defined(__CFA_NO_STATISTICS__) 258 if(this->print_halts) { 259 __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->unique_id, rdtscl()); 260 } 261 #endif 350 __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->unique_id, rdtscl()); ) 262 351 263 352 // We were woken up, remove self from idle … … 268 357 } 269 358 270 /* paranoid */ verify( readyThread );271 272 // Reset io dirty bit273 this->io.dirty = false;274 275 // We found a thread run it276 __run_thread(this, readyThread);277 278 // Are we done?279 if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;280 281 if(this->io.pending && !this->io.dirty) {282 ready_schedule_lock();283 __cfa_io_flush( this );284 ready_schedule_unlock();285 }286 287 #else288 #warning new kernel loop289 SEARCH: {290 /* paranoid */ verify( ! __preemption_enabled() );291 292 // First, lock the scheduler since we are searching for a thread293 ready_schedule_lock();294 295 // Try to get the next thread296 readyThread = pop_fast( this->cltr );297 if(readyThread) { ready_schedule_unlock(); break SEARCH; }298 299 // If we can't find a thread, might as well flush any outstanding I/O300 if(this->io.pending) { __cfa_io_flush( this ); }301 302 // Spin a little on I/O, just in case303 for(5) {304 __maybe_io_drain( this );305 readyThread = pop_fast( this->cltr );306 if(readyThread) { ready_schedule_unlock(); break SEARCH; }307 }308 309 // no luck, try stealing a few times310 for(5) {311 if( __maybe_io_drain( this ) ) {312 readyThread = pop_fast( this->cltr );313 } else {314 readyThread = pop_slow( this->cltr );315 }316 if(readyThread) { ready_schedule_unlock(); break SEARCH; }317 }318 319 // still no luck, search for a thread320 readyThread = pop_search( this->cltr );321 if(readyThread) { ready_schedule_unlock(); break SEARCH; }322 323 // Don't block if we are done324 if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) {325 ready_schedule_unlock();326 break MAIN_LOOP;327 }328 329 __STATS( __tls_stats()->ready.sleep.halts++; )330 331 // Push self to idle stack332 ready_schedule_unlock();333 mark_idle(this->cltr->procs, * this);334 ready_schedule_lock();335 336 // Confirm the ready-queue is empty337 __maybe_io_drain( this );338 readyThread = pop_search( this->cltr );339 ready_schedule_unlock();340 341 if( readyThread ) {342 // A thread was found, cancel the halt343 mark_awake(this->cltr->procs, * this);344 345 __STATS( __tls_stats()->ready.sleep.cancels++; )346 347 // continue the main loop348 break SEARCH;349 }350 351 __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); )352 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle);353 354 {355 eventfd_t val;356 ssize_t ret = read( this->idle, &val, sizeof(val) );357 if(ret < 0) {358 switch((int)errno) {359 case EAGAIN:360 #if EAGAIN != EWOULDBLOCK361 case EWOULDBLOCK:362 #endif363 case EINTR:364 // No need to do anything special here, just assume it's a legitimate wake-up365 break;366 default:367 abort( "KERNEL : internal error, read failure on idle eventfd, error(%d) %s.", (int)errno, strerror( (int)errno ) );368 }369 }370 }371 372 __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->unique_id, rdtscl()); )373 374 // We were woken up, remove self from idle375 mark_awake(this->cltr->procs, * this);376 377 // DON'T just proceed, start looking again378 continue MAIN_LOOP;379 }380 381 359 RUN_THREAD: 382 360 /* paranoid */ verify( ! __preemption_enabled() ); … … 393 371 394 372 if(this->io.pending && !this->io.dirty) { 395 __cfa_io_flush( this );373 __cfa_io_flush( this, 0 ); 396 374 } 397 375 … … 403 381 404 382 __cfadbg_print_safe(runtime_core, "Kernel : core %p stopping\n", this); 383 } 384 385 for(int i = 0; !available(future); i++) { 386 if(i > 1000) __cfaabi_dbg_write( "ERROR: kernel has bin spinning on a flush after exit loop.\n", 60); 387 __cfa_io_flush( this, 1 ); 405 388 } 406 389 … … 766 749 767 750 // Check if there is a sleeping processor 768 processor * p; 769 unsigned idle; 770 unsigned total; 771 [idle, total, p] = query_idles(this->procs); 751 int fd = __atomic_load_n(&this->procs.fd, __ATOMIC_SEQ_CST); 772 752 773 753 // If no one is sleeping, we are done 774 if( idle== 0 ) return;754 if( fd == 0 ) return; 775 755 776 756 // We found a processor, wake it up 777 757 eventfd_t val; 778 758 val = 1; 779 eventfd_write( p->idle, val );759 eventfd_write( fd, val ); 780 760 781 761 #if !defined(__CFA_NO_STATISTICS__) … … 802 782 eventfd_t val; 803 783 val = 1; 804 eventfd_write( this->idle , val );784 eventfd_write( this->idle_fd, val ); 805 785 __enable_interrupts_checked(); 806 786 } 807 787 808 static void mark_idle(__cluster_proc_list & this, processor & proc) { 809 /* paranoid */ verify( ! __preemption_enabled() ); 810 lock( this ); 788 static void idle_sleep(processor * this, io_future_t & future, iovec & iov) { 789 #if !defined(CFA_WITH_IO_URING_IDLE) 790 #if !defined(__CFA_NO_STATISTICS__) 791 if(this->print_halts) { 792 __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); 793 } 794 #endif 795 796 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle_fd); 797 798 { 799 eventfd_t val; 800 ssize_t ret = read( this->idle_fd, &val, sizeof(val) ); 801 if(ret < 0) { 802 switch((int)errno) { 803 case EAGAIN: 804 #if EAGAIN != EWOULDBLOCK 805 case EWOULDBLOCK: 806 #endif 807 case EINTR: 808 // No need to do anything special here, just assume it's a legitimate wake-up 809 break; 810 default: 811 abort( "KERNEL : internal error, read failure on idle eventfd, error(%d) %s.", (int)errno, strerror( (int)errno ) ); 812 } 813 } 814 } 815 816 #if !defined(__CFA_NO_STATISTICS__) 817 if(this->print_halts) { 818 __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->unique_id, rdtscl()); 819 } 820 #endif 821 #else 822 // Do we already have a pending read 823 if(available(future)) { 824 // There is no pending read, we need to add one 825 reset(future); 826 827 __kernel_read(this, future, iov, this->idle_fd ); 828 } 829 830 __cfa_io_flush( this, 1 ); 831 #endif 832 } 833 834 static bool mark_idle(__cluster_proc_list & this, processor & proc) { 835 /* paranoid */ verify( ! __preemption_enabled() ); 836 if(!try_lock( this )) return false; 811 837 this.idle++; 812 838 /* paranoid */ verify( this.idle <= this.total ); 813 839 remove(proc); 814 840 insert_first(this.idles, proc); 841 842 __atomic_store_n(&this.fd, proc.idle_fd, __ATOMIC_SEQ_CST); 815 843 unlock( this ); 816 844 /* paranoid */ verify( ! __preemption_enabled() ); 845 846 return true; 817 847 } 818 848 … … 824 854 remove(proc); 825 855 insert_last(this.actives, proc); 856 857 { 858 int fd = 0; 859 if(!this.idles`isEmpty) fd = this.idles`first.idle_fd; 860 __atomic_store_n(&this.fd, fd, __ATOMIC_SEQ_CST); 861 } 862 826 863 unlock( this ); 827 /* paranoid */ verify( ! __preemption_enabled() );828 }829 830 static [unsigned idle, unsigned total, * processor] query_idles( & __cluster_proc_list this ) {831 /* paranoid */ verify( ! __preemption_enabled() );832 /* paranoid */ verify( ready_schedule_islocked() );833 834 for() {835 uint64_t l = __atomic_load_n(&this.lock, __ATOMIC_SEQ_CST);836 if( 1 == (l % 2) ) { Pause(); continue; }837 unsigned idle = this.idle;838 unsigned total = this.total;839 processor * proc = &this.idles`first;840 // Compiler fence is unnecessary, but gcc-8 and older incorrectly reorder code without it841 asm volatile("": : :"memory");842 if(l != __atomic_load_n(&this.lock, __ATOMIC_SEQ_CST)) { Pause(); continue; }843 return [idle, total, proc];844 }845 846 /* paranoid */ verify( ready_schedule_islocked() );847 864 /* paranoid */ verify( ! __preemption_enabled() ); 848 865 } … … 906 923 if(head == tail) return false; 907 924 #if OLD_MAIN 908 ready_schedule_lock();909 ret = __cfa_io_drain( proc );910 ready_schedule_unlock();925 ready_schedule_lock(); 926 ret = __cfa_io_drain( proc ); 927 ready_schedule_unlock(); 911 928 #else 912 929 ret = __cfa_io_drain( proc ); 913 #endif930 #endif 914 931 #endif 915 932 return ret; -
libcfa/src/concurrency/kernel.hfa
re2853eb r6c53a93 100 100 101 101 // Idle lock (kernel semaphore) 102 int idle ;102 int idle_fd; 103 103 104 104 // Termination synchronisation (user semaphore) … … 195 195 struct __cluster_proc_list { 196 196 // Spin lock protecting the queue 197 volatile uint64_t lock; 197 __spinlock_t lock; 198 199 // FD to use to wake a processor 200 volatile int fd; 198 201 199 202 // Total number of processors -
libcfa/src/concurrency/kernel/startup.cfa
re2853eb r6c53a93 527 527 this.local_data = 0p; 528 528 529 this.idle = eventfd(0, 0);530 if (idle < 0) {529 this.idle_fd = eventfd(0, 0); 530 if (idle_fd < 0) { 531 531 abort("KERNEL ERROR: PROCESSOR EVENTFD - %s\n", strerror(errno)); 532 532 } … … 542 542 // Not a ctor, it just preps the destruction but should not destroy members 543 543 static void deinit(processor & this) { 544 close(this.idle );544 close(this.idle_fd); 545 545 } 546 546 … … 584 584 // Cluster 585 585 static void ?{}(__cluster_proc_list & this) { 586 this. lock= 0;586 this.fd = 0; 587 587 this.idle = 0; 588 588 this.total = 0; -
libcfa/src/concurrency/kernel_private.hfa
re2853eb r6c53a93 39 39 } 40 40 41 // Defines whether or not we *want* to use io_uring_enter as the idle_sleep blocking call 42 #define CFA_WANT_IO_URING_IDLE 43 44 // Defines whether or not we *can* use io_uring_enter as the idle_sleep blocking call 45 #if defined(CFA_WANT_IO_URING_IDLE) && defined(CFA_HAVE_LINUX_IO_URING_H) 46 #if defined(CFA_HAVE_IORING_OP_READ) || (defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV)) 47 #define CFA_WITH_IO_URING_IDLE 48 #endif 49 #endif 50 41 51 //----------------------------------------------------------------------------- 42 52 // Scheduler … … 149 159 __atomic_store_n(ll, (bool)false, __ATOMIC_RELEASE); 150 160 } 151 152 153 154 155 161 156 162 //----------------------------------------------------------------------- … … 268 274 ready_schedule_lock(); 269 275 270 // Simple counting lock, acquired, acquired by incrementing the counter 271 // to an odd number 272 for() { 273 uint64_t l = this.lock; 274 if( 275 (0 == (l % 2)) 276 && __atomic_compare_exchange_n(&this.lock, &l, l + 1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) 277 ) return; 278 Pause(); 279 } 280 281 /* paranoid */ verify( ! __preemption_enabled() ); 276 lock( this.lock __cfaabi_dbg_ctx2 ); 277 278 /* paranoid */ verify( ! __preemption_enabled() ); 279 } 280 281 static inline bool try_lock(__cluster_proc_list & this) { 282 /* paranoid */ verify( ! __preemption_enabled() ); 283 284 // Start by locking the global RWlock so that we know no-one is 285 // adding/removing processors while we mess with the idle lock 286 ready_schedule_lock(); 287 288 if(try_lock( this.lock __cfaabi_dbg_ctx2 )) { 289 // success 290 /* paranoid */ verify( ! __preemption_enabled() ); 291 return true; 292 } 293 294 // failed to lock 295 ready_schedule_unlock(); 296 297 /* paranoid */ verify( ! __preemption_enabled() ); 298 return false; 282 299 } 283 300 … … 285 302 /* paranoid */ verify( ! __preemption_enabled() ); 286 303 287 /* paranoid */ verify( 1 == (this.lock % 2) ); 288 // Simple couting lock, release by incrementing to an even number 289 __atomic_fetch_add( &this.lock, 1, __ATOMIC_SEQ_CST ); 304 unlock(this.lock); 290 305 291 306 // Release the global lock, which we acquired when locking
Note:
See TracChangeset
for help on using the changeset viewer.
