Changeset bf8b77e for libcfa/src/concurrency

Timestamp:

Mar 3, 2022, 1:37:31 PM (2 years ago)

Author:

m3zulfiq <m3zulfiq@…>

Branches:

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

Children:

40a606d2, ba897d21

Parents:

9c5aef9 (diff), b0d0285 (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) (7 diffs)
• io/setup.cfa (modified) (2 diffs)
• iofwd.hfa (modified) (2 diffs)
• kernel/fwd.hfa (modified) (1 diff)
• kernel_private.hfa (modified) (1 diff)
• preemption.cfa (modified) (2 diffs)

libcfa/src/concurrency/io.cfa

@@ -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
@@ -306 +306 @@
                 ctx->proc->io.pending = true;
                 ctx->proc->io.dirty   = true;
+        }
+
+        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);
+
+                ctx->proc->io.pending = true;
+
+                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/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>
@@ -133 +134 @@
 // 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/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: