Changeset 11054eb

Timestamp:

Mar 17, 2021, 4:52:22 PM (3 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

5c2b454

Parents:

c407434e

Message:

Fix io to no longer use monitors since some usages aren't in threads

Location:

libcfa/src/concurrency

Files:

• io.cfa (modified) (7 diffs)
• io/setup.cfa (modified) (2 diffs)
• io/types.hfa (modified) (3 diffs)

libcfa/src/concurrency/io.cfa

@@ -80 +80 @@
         };
 
-        static $io_context * __ioarbiter_allocate( $io_arbiter & mutex this, processor *, __u32 idxs[], __u32 want );
-        static void __ioarbiter_submit( $io_arbiter & mutex this, $io_context * , __u32 idxs[], __u32 have, bool lazy );
-        static void __ioarbiter_flush ( $io_arbiter & mutex this, $io_context * );
+        static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want );
+        static void __ioarbiter_submit( $io_context * , __u32 idxs[], __u32 have, bool lazy );
+        static void __ioarbiter_flush ( $io_context & );
         static inline void __ioarbiter_notify( $io_context & ctx );
 //=============================================================================================
@@ -134 +134 @@
                 $io_context & ctx = *proc->io.ctx;
 
-                if(!ctx.ext_sq.empty) {
-                        __ioarbiter_flush( *ctx.arbiter, &ctx );
-                }
+                __ioarbiter_flush( ctx );
 
                 __STATS__( true, io.calls.flush++; )
@@ -263 +261 @@
                 __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for allocation\n");
 
-                struct $io_context * ret = __ioarbiter_allocate(*ioarb, proc, idxs, want);
+                struct $io_context * ret = __ioarbiter_allocate(*ioarb, idxs, want);
 
                 __cfadbg_print_safe(io, "Kernel I/O : slow allocation completed from ring %d\n", ret->fd);
@@ -326 +324 @@
                 __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for submission\n");
 
-                __ioarbiter_submit(*inctx->arbiter, inctx, idxs, have, lazy);
+                __ioarbiter_submit(inctx, idxs, have, lazy);
         }
 
@@ -384 +382 @@
 // I/O Arbiter
 //=============================================================================================
-        static $io_context * __ioarbiter_allocate( $io_arbiter & mutex this, processor * proc, __u32 idxs[], __u32 want ) {
+        static inline void block(__outstanding_io_queue & queue, __outstanding_io & item) {
+                // Lock the list, it's not thread safe
+                lock( queue.lock __cfaabi_dbg_ctx2 );
+                {
+                        // Add our request to the list
+                        add( queue.queue, item );
+
+                        // Mark as pending
+                        __atomic_store_n( &queue.empty, false, __ATOMIC_SEQ_CST );
+                }
+                unlock( queue.lock );
+
+                wait( item.sem );
+        }
+
+        static inline bool empty(__outstanding_io_queue & queue ) {
+                return __atomic_load_n( &queue.empty, __ATOMIC_SEQ_CST);
+        }
+
+        static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want ) {
                 __cfadbg_print_safe(io, "Kernel I/O : arbiter allocating\n");
 
@@ -390 +407 @@
 
                 // No one has any resources left, wait for something to finish
-                // Mark as pending
-                __atomic_store_n( &this.pending.flag, true, __ATOMIC_SEQ_CST );
-
-                // Wait for our turn to submit
-                wait( this.pending.blocked, want );
-
-                __attribute((unused)) bool ret =
-                __alloc( this.pending.ctx, idxs, want);
-                /* paranoid */ verify( ret );
-
-                return this.pending.ctx;
-
-        }
-
-        static void __ioarbiter_notify( $io_arbiter & mutex this, $io_context * ctx ) {
-                /* paranoid */ verify( !is_empty(this.pending.blocked) );
-                this.pending.ctx = ctx;
-
-                while( !is_empty(this.pending.blocked) ) {
-                        __cfadbg_print_safe(io, "Kernel I/O : notifying\n");
-                        __u32 have = ctx->sq.free_ring.tail - ctx->sq.free_ring.head;
-                        __u32 want = front( this.pending.blocked );
-
-                        if( have > want ) return;
-
-                        signal_block( this.pending.blocked );
-                }
-
-                this.pending.flag = false;
+                // We need to add ourself to a list of pending allocs and wait for an answer
+                __pending_alloc pa;
+                pa.idxs = idxs;
+                pa.want = want;
+
+                block(this.pending, (__outstanding_io&)pa);
+
+                return pa.ctx;
+
+        }
+
+        static void __ioarbiter_notify( $io_arbiter & this, $io_context * ctx ) {
+                /* paranoid */ verify( !empty(this.pending.queue) );
+
+                lock( this.pending.lock __cfaabi_dbg_ctx2 );
+                {
+                        while( !empty(this.pending.queue) ) {
+                                __cfadbg_print_safe(io, "Kernel I/O : notifying\n");
+                                __u32 have = ctx->sq.free_ring.tail - ctx->sq.free_ring.head;
+                                __pending_alloc & pa = (__pending_alloc&)head( this.pending.queue );
+
+                                if( have > pa.want ) goto DONE;
+                                drop( this.pending.queue );
+
+                                /* paranoid */__attribute__((unused)) bool ret =
+
+                                __alloc(ctx, pa.idxs, pa.want);
+
+                                /* paranoid */ verify( ret );
+
+                                pa.ctx = ctx;
+
+                                post( pa.sem );
+                        }
+
+                        this.pending.empty = true;
+                        DONE:;
+                }
+                unlock( this.pending.lock );
         }
 
         static void __ioarbiter_notify( $io_context & ctx ) {
-                if(__atomic_load_n( &ctx.arbiter->pending.flag, __ATOMIC_SEQ_CST)) {
+                if(!empty( ctx.arbiter->pending )) {
                         __ioarbiter_notify( *ctx.arbiter, &ctx );
                 }
@@ -428 +455 @@
 
         // Simply append to the pending
-        static void __ioarbiter_submit( $io_arbiter & mutex this, $io_context * ctx, __u32 idxs[], __u32 have, bool lazy ) {
+        static void __ioarbiter_submit( $io_context * ctx, __u32 idxs[], __u32 have, bool lazy ) {
                 __cfadbg_print_safe(io, "Kernel I/O : submitting %u from the arbiter to context %u\n", have, ctx->fd);
 
-                /* paranoid */ verify( &this == ctx->arbiter );
-
-                // Mark as pending
-                __atomic_store_n( &ctx->ext_sq.empty, false, __ATOMIC_SEQ_CST );
-
                 __cfadbg_print_safe(io, "Kernel I/O : waiting to submit %u\n", have);
 
-                // Wait for our turn to submit
-                wait( ctx->ext_sq.blocked );
-
-                // Submit our indexes
-                __submit(ctx, idxs, have, lazy);
+                __external_io ei;
+                ei.idxs = idxs;
+                ei.have = have;
+                ei.lazy = lazy;
+
+                block(ctx->ext_sq, (__outstanding_io&)ei);
 
                 __cfadbg_print_safe(io, "Kernel I/O : %u submitted from arbiter\n", have);
         }
 
-        static void __ioarbiter_flush( $io_arbiter & mutex this, $io_context * ctx ) {
-                /* paranoid */ verify( &this == ctx->arbiter );
-
-                __STATS__( false, io.flush.external += 1; )
-
-                __cfadbg_print_safe(io, "Kernel I/O : arbiter flushing\n");
-
-                condition & blcked = ctx->ext_sq.blocked;
-                /* paranoid */ verify( ctx->ext_sq.empty == is_empty( blcked ) );
-                while(!is_empty( blcked )) {
-                        signal_block( blcked );
-                }
-
-                ctx->ext_sq.empty = true;
+        static void __ioarbiter_flush( $io_context & ctx ) {
+                if(!empty( ctx.ext_sq )) {
+                        __STATS__( false, io.flush.external += 1; )
+
+                        __cfadbg_print_safe(io, "Kernel I/O : arbiter flushing\n");
+
+                        lock( ctx.ext_sq.lock __cfaabi_dbg_ctx2 );
+                        {
+                                while( !empty(ctx.ext_sq.queue) ) {
+                                        __external_io & ei = (__external_io&)drop( ctx.ext_sq.queue );
+
+                                        __submit(&ctx, ei.idxs, ei.have, ei.lazy);
+
+                                        post( ei.sem );
+                                }
+
+                                ctx.ext_sq.empty = true;
+                        }
+                        unlock(ctx.ext_sq.lock );
+                }
         }
 #endif

libcfa/src/concurrency/io/setup.cfa

@@ -110 +110 @@
                 this.arbiter = cl.io.arbiter;
                 this.ext_sq.empty = true;
-                (this.ext_sq.blocked){};
+                (this.ext_sq.queue){};
                 __io_uring_setup( this, cl.io.params, proc->idle );
                 __cfadbg_print_safe(io_core, "Kernel I/O : Created ring for io_context %u (%p)\n", this.fd, &this);
@@ -329 +329 @@
 //=============================================================================================
         void ?{}( $io_arbiter & this ) {
-                this.pending.flag = false;
-        }
-
-        void ^?{}( $io_arbiter & mutex this ) {
-                // /* paranoid */ verify( empty(this.assigned) );
-                // /* paranoid */ verify( empty(this.available) );
-                /* paranoid */ verify( is_empty(this.pending.blocked) );
-        }
+                this.pending.empty = true;
+        }
+
+        void ^?{}( $io_arbiter & this ) {}
 
         $io_arbiter * create(void) {

libcfa/src/concurrency/io/types.hfa

@@ -22 +22 @@
 
 #include "bits/locks.hfa"
+#include "bits/queue.hfa"
 #include "kernel/fwd.hfa"
 
@@ -95 +96 @@
         };
 
+        struct __outstanding_io {
+                inline Colable;
+                single_sem sem;
+        };
+        static inline __outstanding_io *& Next( __outstanding_io * n ) { return (__outstanding_io *)Next( (Colable *)n ); }
+
+        struct __outstanding_io_queue {
+                __spinlock_t lock;
+                Queue(__outstanding_io) queue;
+                volatile bool empty;
+        };
+
+        struct __external_io {
+                inline __outstanding_io;
+                __u32 * idxs;
+                __u32 have;
+                bool lazy;
+        };
+
+
         struct __attribute__((aligned(128))) $io_context {
                 $io_arbiter * arbiter;
                 processor * proc;
 
-                struct {
-                        volatile bool empty;
-                        condition blocked;
-                } ext_sq;
+                __outstanding_io_queue ext_sq;
 
                 struct __sub_ring_t sq;
@@ -110 +128 @@
         };
 
-        monitor __attribute__((aligned(128))) $io_arbiter {
-                struct {
-                        condition blocked;
-                        $io_context * ctx;
-                        volatile bool flag;
-                } pending;
+        struct __pending_alloc {
+                inline __outstanding_io;
+                __u32 * idxs;
+                __u32 want;
+                $io_context * ctx;
+        };
+
+        struct __attribute__((aligned(128))) $io_arbiter {
+                __outstanding_io_queue pending;
         };