Changes in libcfa/src/concurrency/io.cfa [70b4aeb9:d529ad0]

File:

• libcfa/src/concurrency/io.cfa (modified) (8 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 );