Changes in libcfa/src/concurrency/kernel.cfa [1f45c7d:c33c2af]

File:

• libcfa/src/concurrency/kernel.cfa (modified) (32 diffs)

libcfa/src/concurrency/kernel.cfa

@@ -22 +22 @@
 #include <errno.h>
 #include <stdio.h>
+#include <string.h>
 #include <signal.h>
 #include <unistd.h>
@@ -31 +32 @@
 #include "kernel_private.hfa"
 #include "preemption.hfa"
+#include "strstream.hfa"
+#include "device/cpu.hfa"
 
 //Private includes
@@ -110 +113 @@
 #endif
 
-extern $thread * mainThread;
+extern thread$ * mainThread;
 extern processor * mainProcessor;
 
 //-----------------------------------------------------------------------------
 // Kernel Scheduling logic
-static $thread * __next_thread(cluster * this);
-static $thread * __next_thread_slow(cluster * this);
-static inline bool __must_unpark( $thread * thrd ) __attribute((nonnull(1)));
-static void __run_thread(processor * this, $thread * dst);
+static thread$ * __next_thread(cluster * this);
+static thread$ * __next_thread_slow(cluster * this);
+static inline bool __must_unpark( thread$ * thrd ) __attribute((nonnull(1)));
+static void __run_thread(processor * this, thread$ * dst);
 static void __wake_one(cluster * cltr);
 
@@ -181 +184 @@
                 __cfadbg_print_safe(runtime_core, "Kernel : core %p started\n", this);
 
-                $thread * readyThread = 0p;
+                thread$ * readyThread = 0p;
                 MAIN_LOOP:
                 for() {
@@ -193 +196 @@
 
                         if( !readyThread ) {
+                                ready_schedule_lock();
                                 __cfa_io_flush( this );
+                                ready_schedule_unlock();
+
                                 readyThread = __next_thread_slow( this->cltr );
                         }
@@ -231 +237 @@
                                 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle);
 
-                                __disable_interrupts_hard();
-                                eventfd_t val;
-                                eventfd_read( this->idle, &val );
-                                __enable_interrupts_hard();
+                                {
+                                        eventfd_t val;
+                                        ssize_t ret = read( this->idle, &val, sizeof(val) );
+                                        if(ret < 0) {
+                                                switch((int)errno) {
+                                                case EAGAIN:
+                                                #if EAGAIN != EWOULDBLOCK
+                                                        case EWOULDBLOCK:
+                                                #endif
+                                                case EINTR:
+                                                        // No need to do anything special here, just assume it's a legitimate wake-up
+                                                        break;
+                                                default:
+                                                        abort( "KERNEL : internal error, read failure on idle eventfd, error(%d) %s.", (int)errno, strerror( (int)errno ) );
+                                                }
+                                        }
+                                }
 
                                 #if !defined(__CFA_NO_STATISTICS__)
@@ -261 +280 @@
 
                         if(this->io.pending && !this->io.dirty) {
+                                ready_schedule_lock();
                                 __cfa_io_flush( this );
+                                ready_schedule_unlock();
                         }
 
@@ -301 +322 @@
 
                                 // Don't block if we are done
-                                if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
+                                if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) {
+                                        ready_schedule_unlock();
+                                        break MAIN_LOOP;
+                                }
 
                                 __STATS( __tls_stats()->ready.sleep.halts++; )
@@ -325 +349 @@
                                 }
 
-                                        __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); )
+                                __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->unique_id, rdtscl()); )
                                 __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle);
 
-                                // __disable_interrupts_hard();
-                                eventfd_t val;
-                                eventfd_read( this->idle, &val );
-                                // __enable_interrupts_hard();
+                                {
+                                        eventfd_t val;
+                                        ssize_t ret = read( this->idle, &val, sizeof(val) );
+                                        if(ret < 0) {
+                                                switch((int)errno) {
+                                                case EAGAIN:
+                                                #if EAGAIN != EWOULDBLOCK
+                                                        case EWOULDBLOCK:
+                                                #endif
+                                                case EINTR:
+                                                        // No need to do anything special here, just assume it's a legitimate wake-up
+                                                        break;
+                                                default:
+                                                        abort( "KERNEL : internal error, read failure on idle eventfd, error(%d) %s.", (int)errno, strerror( (int)errno ) );
+                                                }
+                                        }
+                                }
 
                                         __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->unique_id, rdtscl()); )
@@ -388 +425 @@
 // runThread runs a thread by context switching
 // from the processor coroutine to the target thread
-static void __run_thread(processor * this, $thread * thrd_dst) {
+static void __run_thread(processor * this, thread$ * thrd_dst) {
         /* paranoid */ verify( ! __preemption_enabled() );
         /* paranoid */ verifyf( thrd_dst->state == Ready || thrd_dst->preempted != __NO_PREEMPTION, "state : %d, preempted %d\n", thrd_dst->state, thrd_dst->preempted);
@@ -394 +431 @@
         __builtin_prefetch( thrd_dst->context.SP );
 
-        int curr = __kernel_getcpu();
-        if(thrd_dst->last_cpu != curr) {
-                int64_t l = thrd_dst->last_cpu;
-                int64_t c = curr;
-                int64_t v = (l << 32) | c;
-                __push_stat( __tls_stats(), v, false, "Processor", this );
-        }
-
-        thrd_dst->last_cpu = curr;
-
         __cfadbg_print_safe(runtime_core, "Kernel : core %p running thread %p (%s)\n", this, thrd_dst, thrd_dst->self_cor.name);
 
-        $coroutine * proc_cor = get_coroutine(this->runner);
+        coroutine$ * proc_cor = get_coroutine(this->runner);
 
         // set state of processor coroutine to inactive
@@ -425 +452 @@
                 /* paranoid */ verify( thrd_dst->context.SP );
                 /* paranoid */ verify( thrd_dst->state != Halted );
-                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too small.\n", thrd_dst ); // add escape condition if we are setting up the processor
-                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too large.\n", thrd_dst ); // add escape condition if we are setting up the processor
+                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too small.\n", thrd_dst ); // add escape condition if we are setting up the processor
+                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too large.\n", thrd_dst ); // add escape condition if we are setting up the processor
                 /* paranoid */ verify( 0x0D15EA5E0D15EA5Ep == thrd_dst->canary );
 
@@ -438 +465 @@
 
                 /* paranoid */ verify( 0x0D15EA5E0D15EA5Ep == thrd_dst->canary );
-                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->corctx_flag, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too large.\n", thrd_dst );
-                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->corctx_flag, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too small.\n", thrd_dst );
+                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too large.\n", thrd_dst );
+                /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too small.\n", thrd_dst );
                 /* paranoid */ verify( thrd_dst->context.SP );
                 /* paranoid */ verify( thrd_dst->curr_cluster == this->cltr );
@@ -457 +484 @@
                 if(unlikely(thrd_dst->preempted != __NO_PREEMPTION)) {
                         // The thread was preempted, reschedule it and reset the flag
-                        schedule_thread$( thrd_dst );
+                        schedule_thread$( thrd_dst, UNPARK_LOCAL );
                         break RUNNING;
                 }
@@ -505 +532 @@
 void returnToKernel() {
         /* paranoid */ verify( ! __preemption_enabled() );
-        $coroutine * proc_cor = get_coroutine(kernelTLS().this_processor->runner);
-        $thread * thrd_src = kernelTLS().this_thread;
+        coroutine$ * proc_cor = get_coroutine(kernelTLS().this_processor->runner);
+        thread$ * thrd_src = kernelTLS().this_thread;
 
         __STATS( thrd_src->last_proc = kernelTLS().this_processor; )
@@ -534 +561 @@
 
         /* paranoid */ verify( ! __preemption_enabled() );
-        /* paranoid */ verifyf( ((uintptr_t)thrd_src->context.SP) < ((uintptr_t)__get_stack(thrd_src->curr_cor)->base ) || thrd_src->corctx_flag, "ERROR : Returning $thread %p has been corrupted.\n StackPointer too small.\n", thrd_src );
-        /* paranoid */ verifyf( ((uintptr_t)thrd_src->context.SP) > ((uintptr_t)__get_stack(thrd_src->curr_cor)->limit) || thrd_src->corctx_flag, "ERROR : Returning $thread %p has been corrupted.\n StackPointer too large.\n", thrd_src );
+        /* paranoid */ verifyf( ((uintptr_t)thrd_src->context.SP) < ((uintptr_t)__get_stack(thrd_src->curr_cor)->base ) || thrd_src->corctx_flag, "ERROR : Returning thread$ %p has been corrupted.\n StackPointer too small.\n", thrd_src );
+        /* paranoid */ verifyf( ((uintptr_t)thrd_src->context.SP) > ((uintptr_t)__get_stack(thrd_src->curr_cor)->limit) || thrd_src->corctx_flag, "ERROR : Returning thread$ %p has been corrupted.\n StackPointer too large.\n", thrd_src );
 }
 
@@ -541 +568 @@
 // Scheduler routines
 // KERNEL ONLY
-static void __schedule_thread( $thread * thrd ) {
+static void __schedule_thread( thread$ * thrd, unpark_hint hint ) {
         /* paranoid */ verify( ! __preemption_enabled() );
         /* paranoid */ verify( ready_schedule_islocked());
@@ -561 +588 @@
         // Dereference the thread now because once we push it, there is not guaranteed it's still valid.
         struct cluster * cl = thrd->curr_cluster;
-        __STATS(bool outside = thrd->last_proc && thrd->last_proc != kernelTLS().this_processor; )
+        __STATS(bool outside = hint == UNPARK_LOCAL && thrd->last_proc && thrd->last_proc != kernelTLS().this_processor; )
 
         // push the thread to the cluster ready-queue
-        push( cl, thrd, local );
+        push( cl, thrd, hint );
 
         // variable thrd is no longer safe to use
@@ -589 +616 @@
 }
 
-void schedule_thread$( $thread * thrd ) {
+void schedule_thread$( thread$ * thrd, unpark_hint hint ) {
         ready_schedule_lock();
-                __schedule_thread( thrd );
+                __schedule_thread( thrd, hint );
         ready_schedule_unlock();
 }
 
 // KERNEL ONLY
-static inline $thread * __next_thread(cluster * this) with( *this ) {
+static inline thread$ * __next_thread(cluster * this) with( *this ) {
         /* paranoid */ verify( ! __preemption_enabled() );
 
         ready_schedule_lock();
-                $thread * thrd = pop_fast( this );
+                thread$ * thrd = pop_fast( this );
         ready_schedule_unlock();
 
@@ -608 +635 @@
 
 // KERNEL ONLY
-static inline $thread * __next_thread_slow(cluster * this) with( *this ) {
+static inline thread$ * __next_thread_slow(cluster * this) with( *this ) {
         /* paranoid */ verify( ! __preemption_enabled() );
 
         ready_schedule_lock();
-                $thread * thrd;
+                thread$ * thrd;
                 for(25) {
                         thrd = pop_slow( this );
@@ -626 +653 @@
 }
 
-static inline bool __must_unpark( $thread * thrd ) {
+static inline bool __must_unpark( thread$ * thrd ) {
         int old_ticket = __atomic_fetch_add(&thrd->ticket, 1, __ATOMIC_SEQ_CST);
         switch(old_ticket) {
@@ -642 +669 @@
 }
 
-void __kernel_unpark( $thread * thrd ) {
+void __kernel_unpark( thread$ * thrd, unpark_hint hint ) {
         /* paranoid */ verify( ! __preemption_enabled() );
         /* paranoid */ verify( ready_schedule_islocked());
@@ -650 +677 @@
         if(__must_unpark(thrd)) {
                 // Wake lost the race,
-                __schedule_thread( thrd );
+                __schedule_thread( thrd, hint );
         }
 
@@ -657 +684 @@
 }
 
-void unpark( $thread * thrd ) {
+void unpark( thread$ * thrd, unpark_hint hint ) {
         if( !thrd ) return;
 
@@ -663 +690 @@
                 disable_interrupts();
                         // Wake lost the race,
-                        schedule_thread$( thrd );
+                        schedule_thread$( thrd, hint );
                 enable_interrupts(false);
         }
@@ -681 +708 @@
         // Should never return
         void __cfactx_thrd_leave() {
-                $thread * thrd = active_thread();
-                $monitor * this = &thrd->self_mon;
+                thread$ * thrd = active_thread();
+                monitor$ * this = &thrd->self_mon;
 
                 // Lock the monitor now
@@ -694 +721 @@
                 /* paranoid */ verify( kernelTLS().this_thread == thrd );
                 /* paranoid */ verify( thrd->context.SP );
-                /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) > ((uintptr_t)__get_stack(thrd->curr_cor)->limit), "ERROR : $thread %p has been corrupted.\n StackPointer too large.\n", thrd );
-                /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) < ((uintptr_t)__get_stack(thrd->curr_cor)->base ), "ERROR : $thread %p has been corrupted.\n StackPointer too small.\n", thrd );
+                /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) > ((uintptr_t)__get_stack(thrd->curr_cor)->limit), "ERROR : thread$ %p has been corrupted.\n StackPointer too large.\n", thrd );
+                /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) < ((uintptr_t)__get_stack(thrd->curr_cor)->base ), "ERROR : thread$ %p has been corrupted.\n StackPointer too small.\n", thrd );
 
                 thrd->state = Halting;
@@ -713 +740 @@
 bool force_yield( __Preemption_Reason reason ) {
         __disable_interrupts_checked();
-                $thread * thrd = kernelTLS().this_thread;
+                thread$ * thrd = kernelTLS().this_thread;
                 /* paranoid */ verify(thrd->state == Active);
 
@@ -825 +852 @@
 //=============================================================================================
 void __kernel_abort_msg( char * abort_text, int abort_text_size ) {
-        $thread * thrd = __cfaabi_tls.this_thread;
+        thread$ * thrd = __cfaabi_tls.this_thread;
 
         if(thrd) {
@@ -920 +947 @@
                         /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
                         /* paranoid */ verify( it->local_data->this_stats );
+                        // __print_stats( it->local_data->this_stats, cltr->print_stats, "Processor", it->name, (void*)it );
                         __tally_stats( cltr->stats, it->local_data->this_stats );
                         it = &(*it)`next;
@@ -929 +957 @@
                 // this doesn't solve all problems but does solve many
                 // so it's probably good enough
+                disable_interrupts();
                 uint_fast32_t last_size = ready_mutate_lock();
 
@@ -936 +965 @@
                 // Unlock the RWlock
                 ready_mutate_unlock( last_size );
+                enable_interrupts();
         }