Changeset 510e6f9 for libcfa

Timestamp:

Mar 11, 2022, 1:56:07 PM (4 years ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, enum, master, pthread-emulation, qualifiedEnum

Children:

623d1c8

Parents:

eb3bc52 (diff), 630c4bb (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.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

Location:

libcfa/src/concurrency

Files:

• io.cfa (modified) (8 diffs)
• io/setup.cfa (modified) (2 diffs)
• io/types.hfa (modified) (2 diffs)
• iofwd.hfa (modified) (5 diffs)
• kernel.cfa (modified) (1 diff)
• kernel.hfa (modified) (1 diff)
• kernel/fwd.hfa (modified) (1 diff)
• kernel_private.hfa (modified) (1 diff)
• preemption.cfa (modified) (2 diffs)

libcfa/src/concurrency/io.cfa

@@ -175 +175 @@
                         /* paranoid */ verify( ! __preemption_enabled() );
 
-                        ctx.proc->io.pending = false;
+                        __atomic_store_n(&ctx.proc->io.pending, false, __ATOMIC_RELAXED);
                 }
 
@@ -287 +287 @@
         //=============================================================================================
         // submission
-        static inline void __submit( struct $io_context * ctx, __u32 idxs[], __u32 have, bool lazy) {
+        static inline void __submit_only( struct $io_context * ctx, __u32 idxs[], __u32 have) {
                 // We can proceed to the fast path
                 // Get the right objects
@@ -304 +304 @@
                 sq.to_submit += have;
 
-                ctx->proc->io.pending = true;
-                ctx->proc->io.dirty   = true;
+                __atomic_store_n(&ctx->proc->io.pending, true, __ATOMIC_RELAXED);
+                __atomic_store_n(&ctx->proc->io.dirty  , true, __ATOMIC_RELAXED);
+        }
+
+        static inline void __submit( struct $io_context * ctx, __u32 idxs[], __u32 have, bool lazy) {
+                __sub_ring_t & sq = ctx->sq;
+                __submit_only(ctx, idxs, have);
+
                 if(sq.to_submit > 30) {
                         __tls_stats()->io.flush.full++;
@@ -402 +408 @@
 // I/O Arbiter
 //=============================================================================================
-        static inline void block(__outstanding_io_queue & queue, __outstanding_io & item) {
+        static inline bool enqueue(__outstanding_io_queue & queue, __outstanding_io & item) {
+                bool was_empty;
+
                 // Lock the list, it's not thread safe
                 lock( queue.lock __cfaabi_dbg_ctx2 );
                 {
+                        was_empty = empty(queue.queue);
+
                         // Add our request to the list
                         add( queue.queue, item );
@@ -414 +424 @@
                 unlock( queue.lock );
 
-                wait( item.sem );
+                return was_empty;
         }
 
@@ -432 +442 @@
                 pa.want = want;
 
-                block(this.pending, (__outstanding_io&)pa);
+                enqueue(this.pending, (__outstanding_io&)pa);
+
+                wait( pa.sem );
 
                 return pa.ctx;
@@ -485 +497 @@
                 ei.lazy = lazy;
 
-                block(ctx->ext_sq, (__outstanding_io&)ei);
+                bool we = enqueue(ctx->ext_sq, (__outstanding_io&)ei);
+
+                __atomic_store_n(&ctx->proc->io.pending, true, __ATOMIC_SEQ_CST);
+
+                if( we ) {
+                        sigval_t value = { PREEMPT_IO };
+                        pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
+                }
+
+                wait( ei.sem );
 
                 __cfadbg_print_safe(io, "Kernel I/O : %u submitted from arbiter\n", have);
@@ -501 +522 @@
                                         __external_io & ei = (__external_io&)drop( ctx.ext_sq.queue );
 
-                                        __submit(&ctx, ei.idxs, ei.have, ei.lazy);
+                                        __submit_only(&ctx, ei.idxs, ei.have);
 
                                         post( ei.sem );

libcfa/src/concurrency/io/setup.cfa

@@ -56 +56 @@
 
         #include "bitmanip.hfa"
+        #include "fstream.hfa"
         #include "kernel_private.hfa"
         #include "thread.hfa"
@@ -258 +259 @@
                 struct __sub_ring_t & sq = this.sq;
                 struct __cmp_ring_t & cq = this.cq;
+                {
+                        __u32 fhead = sq.free_ring.head;
+                        __u32 ftail = sq.free_ring.tail;
+
+                        __u32 total = *sq.num;
+                        __u32 avail = ftail - fhead;
+
+                        if(avail != total) abort | "Processor (" | (void*)this.proc | ") tearing down ring with" | (total - avail) | "entries allocated but not submitted, out of" | total;
+                }
 
                 // unmap the submit queue entries

libcfa/src/concurrency/io/types.hfa

@@ -23 +23 @@
 #include "bits/locks.hfa"
 #include "bits/queue.hfa"
+#include "iofwd.hfa"
 #include "kernel/fwd.hfa"
 
@@ -170 +171 @@
         // void __ioctx_prepare_block($io_context & ctx);
 #endif
-
-//-----------------------------------------------------------------------
-// IO user data
-struct io_future_t {
-        future_t self;
-        __s32 result;
-};
-
-static inline {
-        thread$ * fulfil( io_future_t & this, __s32 result, bool do_unpark = true ) {
-                this.result = result;
-                return fulfil(this.self, do_unpark);
-        }
-
-        // Wait for the future to be fulfilled
-        bool wait     ( io_future_t & this ) { return wait     (this.self); }
-        void reset    ( io_future_t & this ) { return reset    (this.self); }
-        bool available( io_future_t & this ) { return available(this.self); }
-}

libcfa/src/concurrency/iofwd.hfa

@@ -19 +19 @@
 extern "C" {
         #include <asm/types.h>
+        #include <sys/stat.h> // needed for mode_t
         #if CFA_HAVE_LINUX_IO_URING_H
                 #include <linux/io_uring.h>
@@ -24 +25 @@
 }
 #include "bits/defs.hfa"
+#include "kernel/fwd.hfa"
 #include "time.hfa"
 
@@ -47 +49 @@
 
 struct cluster;
-struct io_future_t;
 struct $io_context;
 
@@ -57 +58 @@
 
 struct io_uring_sqe;
+
+//-----------------------------------------------------------------------
+// IO user data
+struct io_future_t {
+        future_t self;
+        __s32 result;
+};
+
+static inline {
+        thread$ * fulfil( io_future_t & this, __s32 result, bool do_unpark = true ) {
+                this.result = result;
+                return fulfil(this.self, do_unpark);
+        }
+
+        // Wait for the future to be fulfilled
+        bool wait     ( io_future_t & this ) { return wait     (this.self); }
+        void reset    ( io_future_t & this ) { return reset    (this.self); }
+        bool available( io_future_t & this ) { return available(this.self); }
+}
 
 //----------
@@ -133 +153 @@
 // Check if a function is blocks a only the user thread
 bool has_user_level_blocking( fptr_t func );
+
+#if CFA_HAVE_LINUX_IO_URING_H
+        static inline void zero_sqe(struct io_uring_sqe * sqe) {
+                sqe->flags = 0;
+                sqe->ioprio = 0;
+                sqe->fd = 0;
+                sqe->off = 0;
+                sqe->addr = 0;
+                sqe->len = 0;
+                sqe->fsync_flags = 0;
+                sqe->__pad2[0] = 0;
+                sqe->__pad2[1] = 0;
+                sqe->__pad2[2] = 0;
+                sqe->fd = 0;
+                sqe->off = 0;
+                sqe->addr = 0;
+                sqe->len = 0;
+        }
+#endif

libcfa/src/concurrency/kernel.cfa

@@ -251 +251 @@
                         if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
 
-                        if(this->io.pending && !this->io.dirty) {
+                        if(__atomic_load_n(&this->io.pending, __ATOMIC_RELAXED) && !__atomic_load_n(&this->io.dirty, __ATOMIC_RELAXED)) {
                                 __IO_STATS__(true, io.flush.dirty++; )
                                 __cfa_io_flush( this, 0 );

libcfa/src/concurrency/kernel.hfa

@@ -92 +92 @@
         struct {
                 $io_context * ctx;
-                bool pending;
-                bool dirty;
+                volatile bool pending;
+                volatile bool dirty;
         } io;
 

libcfa/src/concurrency/kernel/fwd.hfa

@@ -347 +347 @@
                                         struct oneshot * want = expected == 0p ? 1p : 2p;
                                         if(__atomic_compare_exchange_n(&this.ptr, &expected, want, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
-                                                if( expected == 0p ) { /* paranoid */ verify( this.ptr == 1p); return 0p; }
+                                                if( expected == 0p ) { return 0p; }
                                                 thread$ * ret = post( *expected, do_unpark );
                                                 __atomic_store_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);

libcfa/src/concurrency/kernel_private.hfa

@@ -60 +60 @@
 extern bool __preemption_enabled();
 
+enum {
+        PREEMPT_NORMAL    = 0,
+        PREEMPT_TERMINATE = 1,
+        PREEMPT_IO = 2,
+};
+
 static inline void __disable_interrupts_checked() {
         /* paranoid */ verify( __preemption_enabled() );

libcfa/src/concurrency/preemption.cfa

@@ -96 +96 @@
         lock{};
 }
-
-enum {
-        PREEMPT_NORMAL    = 0,
-        PREEMPT_TERMINATE = 1,
-};
 
 //=============================================================================================
@@ -664 +659 @@
         choose(sfp->si_value.sival_int) {
                 case PREEMPT_NORMAL   : ;// Normal case, nothing to do here
+                case PREEMPT_IO       : ;// I/O asked to stop spinning, nothing to do here
                 case PREEMPT_TERMINATE: verify( __atomic_load_n( &__cfaabi_tls.this_processor->do_terminate, __ATOMIC_SEQ_CST ) );
                 default: