Changes in libcfa/src/concurrency/kernel.cfa [da3963a:dddb3dd0]

File:

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

libcfa/src/concurrency/kernel.cfa

@@ -22 +22 @@
 #include <signal.h>
 #include <unistd.h>
+extern "C" {
+        #include <sys/eventfd.h>
+}
 
 //CFA Includes
@@ -109 +112 @@
 static void __run_thread(processor * this, $thread * dst);
 static void __wake_one(cluster * cltr);
-static void wait(__bin_sem_t & this);
 
 static void push  (__cluster_idles & idles, processor & proc);
@@ -115 +117 @@
 static [unsigned idle, unsigned total, * processor] query( & __cluster_idles idles );
 
+extern void __cfa_io_start( processor * );
+extern void __cfa_io_drain( processor * );
+extern void __cfa_io_flush( processor * );
+extern void __cfa_io_stop ( processor * );
+static inline void __maybe_io_drain( processor * );
+
+extern void __disable_interrupts_hard();
+extern void __enable_interrupts_hard();
 
 //=============================================================================================
@@ -130 +140 @@
         verify(this);
 
+        __cfa_io_start( this );
+
         __cfadbg_print_safe(runtime_core, "Kernel : core %p starting\n", this);
         #if !defined(__CFA_NO_STATISTICS__)
@@ -151 +163 @@
                 MAIN_LOOP:
                 for() {
+                        // Check if there is pending io
+                        __maybe_io_drain( this );
+
                         // Try to get the next thread
                         readyThread = __next_thread( this->cltr );
 
                         if( !readyThread ) {
+                                __cfa_io_flush( this );
                                 readyThread = __next_thread_slow( this->cltr );
                         }
@@ -190 +206 @@
                                 #endif
 
-                                wait( this->idle );
+                                __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();
 
                                 #if !defined(__CFA_NO_STATISTICS__)
@@ -206 +227 @@
 
                         /* paranoid */ verify( readyThread );
+
+                        // Reset io dirty bit
+                        this->io.dirty = false;
 
                         // We found a thread run it
@@ -220 +244 @@
                                 }
                         #endif
+
+                        if(this->io.pending && !this->io.dirty) {
+                                __cfa_io_flush( this );
+                        }
                 }
 
@@ -225 +253 @@
         }
 
+        __cfa_io_stop( this );
+
         post( this->terminated );
+
 
         if(this == mainProcessor) {
@@ -248 +279 @@
         /* paranoid */ verifyf( thrd_dst->link.next == 0p, "Expected null got %p", thrd_dst->link.next );
         __builtin_prefetch( thrd_dst->context.SP );
+
+        __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);
@@ -330 +363 @@
         // Just before returning to the processor, set the processor coroutine to active
         proc_cor->state = Active;
+
+        __cfadbg_print_safe(runtime_core, "Kernel : core %p finished running thread %p\n", this, thrd_dst);
 
         /* paranoid */ verify( ! __preemption_enabled() );
@@ -549 +584 @@
 // Kernel Idle Sleep
 //=============================================================================================
-extern "C" {
-        char * strerror(int);
-}
-#define CHECKED(x) { int err = x; if( err != 0 ) abort("KERNEL ERROR: Operation \"" #x "\" return error %d - %s\n", err, strerror(err)); }
-
-static void wait(__bin_sem_t & this) with( this ) {
-        verify(__cfaabi_dbg_in_kernel());
-        CHECKED( pthread_mutex_lock(&lock) );
-                while(val < 1) {
-                        pthread_cond_wait(&cond, &lock);
-                }
-                val -= 1;
-        CHECKED( pthread_mutex_unlock(&lock) );
-}
-
-static bool post(__bin_sem_t & this) with( this ) {
-        bool needs_signal = false;
-
-        CHECKED( pthread_mutex_lock(&lock) );
-                if(val < 1) {
-                        val += 1;
-                        pthread_cond_signal(&cond);
-                        needs_signal = true;
-                }
-        CHECKED( pthread_mutex_unlock(&lock) );
-
-        return needs_signal;
-}
-
-#undef CHECKED
-
 // Wake a thread from the front if there are any
 static void __wake_one(cluster * this) {
@@ -595 +599 @@
 
         // We found a processor, wake it up
-        post( p->idle );
+        eventfd_t val;
+        val = 1;
+        eventfd_write( p->idle, val );
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -613 +619 @@
         disable_interrupts();
                 /* paranoid */ verify( ! __preemption_enabled() );
-                post( this->idle );
+                eventfd_t val;
+                val = 1;
+                eventfd_read( this->idle, &val );
         enable_interrupts( __cfaabi_dbg_ctx );
 }
@@ -696 +704 @@
 // Kernel Utilities
 //=============================================================================================
+#if defined(CFA_HAVE_LINUX_IO_URING_H)
+#include "io/types.hfa"
+#endif
+
+static inline void __maybe_io_drain( processor * proc ) {
+        #if defined(CFA_HAVE_LINUX_IO_URING_H)
+                __cfadbg_print_safe(runtime_core, "Kernel : core %p checking io for ring %d\n", proc, proc->io.ctx->fd);
+
+                // Check if we should drain the queue
+                $io_context * ctx = proc->io.ctx;
+                unsigned head = *ctx->cq.head;
+                unsigned tail = *ctx->cq.tail;
+                if(head != tail) __cfa_io_drain( proc );
+        #endif
+}
+
 //-----------------------------------------------------------------------------
 // Debug