Changes in libcfa/src/concurrency/kernel.cfa [4ccc150:c9c1c1cb]

File:

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

libcfa/src/concurrency/kernel.cfa

@@ -19 +19 @@
 // #define __CFA_DEBUG_PRINT_RUNTIME_CORE__
 
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Waddress-of-packed-member"
-
 //C Includes
 #include <errno.h>
@@ -28 +25 @@
 #include <signal.h>
 #include <unistd.h>
-
 extern "C" {
         #include <sys/eventfd.h>
@@ -35 +31 @@
 
 //CFA Includes
-#include "kernel/private.hfa"
+#include "kernel_private.hfa"
 #include "preemption.hfa"
 #include "strstream.hfa"
@@ -44 +40 @@
 #define __CFA_INVOKE_PRIVATE__
 #include "invoke.h"
-#pragma GCC diagnostic pop
 
 #if !defined(__CFA_NO_STATISTICS__)
@@ -132 +127 @@
 static void __wake_one(cluster * cltr);
 
-static void idle_sleep(processor * proc);
+static void idle_sleep(processor * proc, io_future_t & future, iovec & iov);
 static bool mark_idle (__cluster_proc_list & idles, processor & proc);
 static void mark_awake(__cluster_proc_list & idles, processor & proc);
 
-extern bool __cfa_io_drain( processor * proc ) __attribute__((nonnull (1)));
-extern bool __cfa_io_flush( processor * ) __attribute__((nonnull (1)));
-extern void __cfa_io_idle( processor * ) __attribute__((nonnull (1)));
+extern void __cfa_io_start( processor * );
+extern bool __cfa_io_drain( processor * );
+extern bool __cfa_io_flush( processor *, int min_comp );
+extern void __cfa_io_stop ( processor * );
+static inline bool __maybe_io_drain( processor * );
 
 #if defined(CFA_WITH_IO_URING_IDLE)
@@ -162 +159 @@
         verify(this);
 
-        /* paranoid */ verify( this->idle_wctx.ftr   != 0p );
-        /* paranoid */ verify( this->idle_wctx.rdbuf != 0p );
-
-        // used for idle sleep when io_uring is present
-        // mark it as already fulfilled so we know if there is a pending request or not
-        this->idle_wctx.ftr->self.ptr = 1p;
+        io_future_t future; // used for idle sleep when io_uring is present
+        future.self.ptr = 1p;  // mark it as already fulfilled so we know if there is a pending request or not
+        eventfd_t idle_val;
+        iovec idle_iovec = { &idle_val, sizeof(idle_val) };
+
+        __cfa_io_start( this );
 
         __cfadbg_print_safe(runtime_core, "Kernel : core %p starting\n", this);
@@ -192 +189 @@
                 for() {
                         // Check if there is pending io
-                        __cfa_io_drain( this );
+                        __maybe_io_drain( this );
 
                         // Try to get the next thread
@@ -198 +195 @@
 
                         if( !readyThread ) {
-                                // there is no point in holding submissions if we are idle
                                 __IO_STATS__(true, io.flush.idle++; )
-                                __cfa_io_flush( this );
-
-                                // drain again in case something showed up
-                                __cfa_io_drain( this );
+                                __cfa_io_flush( this, 0 );
 
                                 readyThread = __next_thread( this->cltr );
@@ -209 +202 @@
 
                         if( !readyThread ) for(5) {
+                                __IO_STATS__(true, io.flush.idle++; )
+
                                 readyThread = __next_thread_slow( this->cltr );
 
                                 if( readyThread ) break;
 
-                                // It's unlikely we still I/O to submit, but the arbiter could
-                                __IO_STATS__(true, io.flush.idle++; )
-                                __cfa_io_flush( this );
-
-                                // drain again in case something showed up
-                                __cfa_io_drain( this );
+                                __cfa_io_flush( this, 0 );
                         }
 
@@ -241 +231 @@
                                 }
 
-                                idle_sleep( this );
+                                idle_sleep( this, future, idle_iovec );
 
                                 // We were woken up, remove self from idle
@@ -261 +251 @@
                         if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
 
-                        if(__atomic_load_n(&this->io.pending, __ATOMIC_RELAXED) && !__atomic_load_n(&this->io.dirty, __ATOMIC_RELAXED)) {
+                        if(this->io.pending && !this->io.dirty) {
                                 __IO_STATS__(true, io.flush.dirty++; )
-                                __cfa_io_flush( this );
+                                __cfa_io_flush( this, 0 );
                         }
                 }
@@ -269 +259 @@
                 __cfadbg_print_safe(runtime_core, "Kernel : core %p stopping\n", this);
         }
+
+        for(int i = 0; !available(future); i++) {
+                if(i > 1000) __cfaabi_dbg_write( "ERROR: kernel has bin spinning on a flush after exit loop.\n", 60);
+                __cfa_io_flush( this, 1 );
+        }
+
+        __cfa_io_stop( this );
 
         post( this->terminated );
@@ -637 +634 @@
 
         int fd = 1;
-        if( __atomic_load_n(&fdp->sem, __ATOMIC_SEQ_CST) != 1 ) {
-                fd = __atomic_exchange_n(&fdp->sem, 1, __ATOMIC_RELAXED);
+        if( __atomic_load_n(&fdp->fd, __ATOMIC_SEQ_CST) != 1 ) {
+                fd = __atomic_exchange_n(&fdp->fd, 1, __ATOMIC_RELAXED);
         }
 
         switch(fd) {
-                __attribute__((unused)) int ret;
         case 0:
                 // If the processor isn't ready to sleep then the exchange will already wake it up
@@ -660 +656 @@
                 // If the processor was ready to sleep, we need to wake it up with an actual write
                 val = 1;
-                ret = eventfd_write( fd, val );
-                /* paranoid */ verifyf( ret == 0, "Expected return to be 0, was %d\n", ret );
+                eventfd_write( fd, val );
 
                 #if !defined(__CFA_NO_STATISTICS__)
@@ -682 +677 @@
         __cfadbg_print_safe(runtime_core, "Kernel : waking Processor %p\n", this);
 
-        this->idle_wctx.sem = 1;
-
-        this->idle_wctx.wake__time = rdtscl();
+        this->idle_wctx.fd = 1;
 
         eventfd_t val;
         val = 1;
-        __attribute__((unused)) int ret = eventfd_write( this->idle_wctx.evfd, val );
-
-        /* paranoid */ verifyf( ret == 0, "Expected return to be 0, was %d\n", ret );
-        /* paranoid */ verify( ! __preemption_enabled() );
-}
-
-static void idle_sleep(processor * this) {
-        /* paranoid */ verify( this->idle_wctx.evfd != 1 );
-        /* paranoid */ verify( this->idle_wctx.evfd != 2 );
-
+        eventfd_write( this->idle_fd, val );
+
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
+static void idle_sleep(processor * this, io_future_t & future, iovec & iov) {
         // Tell everyone we are ready to go do sleep
         for() {
-                int expected = this->idle_wctx.sem;
+                int expected = this->idle_wctx.fd;
 
                 // Someone already told us to wake-up! No time for a nap.
@@ -706 +695 @@
 
                 // Try to mark that we are going to sleep
-                if(__atomic_compare_exchange_n(&this->idle_wctx.sem, &expected, this->idle_wctx.evfd, false,  __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) {
+                if(__atomic_compare_exchange_n(&this->idle_wctx.fd, &expected, this->idle_fd, false,  __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) {
                         // Every one agreed, taking a nap
                         break;
@@ -724 +713 @@
                 {
                         eventfd_t val;
-                        ssize_t ret = read( this->idle_wctx.evfd, &val, sizeof(val) );
+                        ssize_t ret = read( this->idle_fd, &val, sizeof(val) );
                         if(ret < 0) {
                                 switch((int)errno) {
@@ -746 +735 @@
                 #endif
         #else
-                __cfa_io_idle( this );
+                // Do we already have a pending read
+                if(available(future)) {
+                        // There is no pending read, we need to add one
+                        reset(future);
+
+                        __kernel_read(this, future, iov, this->idle_fd );
+                }
+
+                __cfa_io_flush( this, 1 );
         #endif
 }
@@ -753 +750 @@
         __STATS__(true, ready.sleep.halts++; )
 
-        proc.idle_wctx.sem = 0;
+        proc.idle_wctx.fd = 0;
 
         /* paranoid */ verify( ! __preemption_enabled() );
@@ -834 +831 @@
 #endif
 
-
+static inline bool __maybe_io_drain( processor * proc ) {
+        bool ret = false;
+        #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) return false;
+                ready_schedule_lock();
+                ret = __cfa_io_drain( proc );
+                ready_schedule_unlock();
+        #endif
+        return ret;
+}
 
 //-----------------------------------------------------------------------------
@@ -891 +903 @@
         void print_stats_now( cluster & this, int flags ) {
                 crawl_cluster_stats( this );
-                __print_stats( this.stats, flags, "Cluster", this.name, (void*)&this );
+                __print_stats( this.stats, this.print_stats, "Cluster", this.name, (void*)&this );
         }
 #endif