Changeset d672350 for libcfa/src/concurrency/io.cfa

Timestamp:

Mar 21, 2022, 1:44:06 PM (4 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

a76202d

Parents:

ef3c383 (diff), dbe2533 (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

File:

• libcfa/src/concurrency/io.cfa (modified) (12 diffs)

libcfa/src/concurrency/io.cfa

@@ -41 +41 @@
         #include "kernel.hfa"
         #include "kernel/fwd.hfa"
-        #include "kernel_private.hfa"
+        #include "kernel/private.hfa"
         #include "io/types.hfa"
 
@@ -93 +93 @@
         extern void __kernel_unpark( thread$ * thrd, unpark_hint );
 
-        bool __cfa_io_drain( processor * proc ) {
+        bool __cfa_io_drain( $io_context * ctx ) {
                 /* paranoid */ verify( ! __preemption_enabled() );
                 /* paranoid */ verify( ready_schedule_islocked() );
-                /* paranoid */ verify( proc );
-                /* paranoid */ verify( proc->io.ctx );
+                /* paranoid */ verify( ctx );
 
                 // Drain the queue
-                $io_context * ctx = proc->io.ctx;
                 unsigned head = *ctx->cq.head;
                 unsigned tail = *ctx->cq.tail;
@@ -110 +108 @@
                 if(count == 0) return false;
 
+                if(!__atomic_try_acquire(&ctx->cq.lock)) {
+                        return false;
+                }
+
                 for(i; count) {
                         unsigned idx = (head + i) & mask;
@@ -130 +132 @@
                 /* paranoid */ verify( ready_schedule_islocked() );
                 /* paranoid */ verify( ! __preemption_enabled() );
+
+                __atomic_unlock(&ctx->cq.lock);
 
                 return true;
@@ -175 +179 @@
                         /* paranoid */ verify( ! __preemption_enabled() );
 
-                        ctx.proc->io.pending = false;
+                        __atomic_store_n(&ctx.proc->io.pending, false, __ATOMIC_RELAXED);
                 }
 
                 ready_schedule_lock();
-                bool ret = __cfa_io_drain( proc );
+                bool ret = __cfa_io_drain( &ctx );
                 ready_schedule_unlock();
                 return ret;
@@ -287 +291 @@
         //=============================================================================================
         // 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 +308 @@
                 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 +412 @@
 // 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 +428 @@
                 unlock( queue.lock );
 
-                wait( item.sem );
+                return was_empty;
         }
 
@@ -432 +446 @@
                 pa.want = want;
 
-                block(this.pending, (__outstanding_io&)pa);
+                enqueue(this.pending, (__outstanding_io&)pa);
+
+                wait( pa.sem );
 
                 return pa.ctx;
@@ -485 +501 @@
                 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 +526 @@
                                         __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 );