Changeset 18f7858 for libcfa/src/concurrency/io.cfa

Timestamp:

Mar 28, 2022, 4:00:32 PM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

37a3aa23

Parents:

2377ca2

Message:

Refactored io to allow holding the lock duirng idle sleep

File:

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

libcfa/src/concurrency/io.cfa

@@ -94 +94 @@
         extern void __kernel_unpark( thread$ * thrd, unpark_hint );
 
-        static bool __cfa_do_drain( $io_context * ctx, cluster * cltr ) {
+        static void ioring_syscsll( struct $io_context & ctx, unsigned int min_comp, unsigned int flags ) {
+                __STATS__( true, io.calls.flush++; )
+                int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, flags, (sigset_t *)0p, _NSIG / 8);
+                if( ret < 0 ) {
+                        switch((int)errno) {
+                        case EAGAIN:
+                        case EINTR:
+                        case EBUSY:
+                                // Update statistics
+                                __STATS__( false, io.calls.errors.busy ++; )
+                                return false;
+                        default:
+                                abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
+                        }
+                }
+
+                __cfadbg_print_safe(io, "Kernel I/O : %u submitted to io_uring %d\n", ret, ctx.fd);
+                __STATS__( true, io.calls.submitted += ret; )
+                /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
+                /* paranoid */ verify( ctx.sq.to_submit >= ret );
+
+                ctx.sq.to_submit -= ret;
+
+                /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
+
+                // Release the consumed SQEs
+                __release_sqes( ctx );
+
+                /* paranoid */ verify( ! __preemption_enabled() );
+
+                __atomic_store_n(&ctx.proc->io.pending, false, __ATOMIC_RELAXED);
+        }
+
+        static bool try_acquire( $io_context * ctx ) __attribute__((nonnull(1))) {
                 /* paranoid */ verify( ! __preemption_enabled() );
                 /* paranoid */ verify( ready_schedule_islocked() );
-                /* paranoid */ verify( ctx );
-
-                const __u32 mask = *ctx->cq.mask;
 
 
@@ -115 +145 @@
                 }
 
+                return true;
+        }
+
+        static bool __cfa_do_drain( $io_context * ctx, cluster * cltr ) __attribute__((nonnull(1, 2))) {
+                /* paranoid */ verify( ! __preemption_enabled() );
+                /* paranoid */ verify( ready_schedule_islocked() );
+                /* paranoid */ verify( ctx->cq.lock == true );
+
+                const __u32 mask = *ctx->cq.mask;
                 unsigned long long ts_prev = ctx->cq.ts;
 
@@ -155 +194 @@
                 bool local = false;
                 bool remote = false;
+
+                ready_schedule_lock();
 
                 cluster * const cltr = proc->cltr;
@@ -186 +227 @@
                                 const unsigned target = proc->io.target;
                                 /* paranoid */ verify( io.tscs[target].tv != MAX );
-                                if(target < ctxs_count) {
+                                HELP: if(target < ctxs_count) {
                                         const unsigned long long cutoff = calc_cutoff(ctsc, ctx->cq.id, ctxs_count, io.data, io.tscs, __shard_factor.io);
                                         const unsigned long long age = moving_average(ctsc, io.tscs[target].tv, io.tscs[target].ma);
                                         // __cfadbg_print_safe(ready_queue, "Kernel : Help attempt on %u from %u, age %'llu vs cutoff %'llu, %s\n", target, this, age, cutoff, age > cutoff ? "yes" : "no");
-                                        if(age > cutoff) {
-                                                remote = __cfa_do_drain( io.data[target], cltr );
-                                                if(remote) __STATS__( false, io.calls.helped++; )
-                                        }
+                                        if(age <= cutoff) break HELP;
+
+                                        if(!try_acquire(io.data[target])) break HELP;
+
+                                        if(!__cfa_do_drain( io.data[target], cltr )) break HELP;
+
+                                        remote = true;
+                                        __STATS__( false, io.calls.helped++; )
                                 }
                                 proc->io.target = MAX;
@@ -201 +246 @@
 
                 // Drain the local queue
-                local = __cfa_do_drain( proc->io.ctx, cltr );
+                if(try_acquire( proc->io.ctx )) {
+                        local = __cfa_do_drain( proc->io.ctx, cltr );
+                }
 
                 /* paranoid */ verify( ready_schedule_islocked() );
                 /* paranoid */ verify( ! __preemption_enabled() );
                 /* paranoid */ verify( active_processor() == proc );
+
+                ready_schedule_unlock();
                 return local || remote;
         }
 
-        bool __cfa_io_flush( processor * proc, int min_comp ) {
+        bool __cfa_io_flush( processor * proc ) {
                 /* paranoid */ verify( ! __preemption_enabled() );
                 /* paranoid */ verify( proc );
@@ -219 +268 @@
                 __ioarbiter_flush( ctx );
 
-                if(ctx.sq.to_submit != 0 || min_comp > 0) {
-
-                        __STATS__( true, io.calls.flush++; )
-                        int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, min_comp > 0 ? IORING_ENTER_GETEVENTS : 0, (sigset_t *)0p, _NSIG / 8);
-                        if( ret < 0 ) {
-                                switch((int)errno) {
-                                case EAGAIN:
-                                case EINTR:
-                                case EBUSY:
-                                        // Update statistics
-                                        __STATS__( false, io.calls.errors.busy ++; )
-                                        return false;
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
-                                }
-                        }
-
-                        __cfadbg_print_safe(io, "Kernel I/O : %u submitted to io_uring %d\n", ret, ctx.fd);
-                        __STATS__( true, io.calls.submitted += ret; )
-                        /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
-                        /* paranoid */ verify( ctx.sq.to_submit >= ret );
-
-                        ctx.sq.to_submit -= ret;
-
-                        /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
-
-                        // Release the consumed SQEs
-                        __release_sqes( ctx );
-
-                        /* paranoid */ verify( ! __preemption_enabled() );
-
-                        __atomic_store_n(&ctx.proc->io.pending, false, __ATOMIC_RELAXED);
-                }
-
-                ready_schedule_lock();
-                bool ret = __cfa_io_drain( proc );
-                ready_schedule_unlock();
-                return ret;
+                if(ctx.sq.to_submit != 0) {
+                        ioring_syscsll(ctx, 0, 0);
+
+                }
+
+                return __cfa_io_drain( proc );
         }
 
@@ -389 +406 @@
                 if(sq.to_submit > 30) {
                         __tls_stats()->io.flush.full++;
-                        __cfa_io_flush( ctx->proc, 0 );
+                        __cfa_io_flush( ctx->proc );
                 }
                 if(!lazy) {
                         __tls_stats()->io.flush.eager++;
-                        __cfa_io_flush( ctx->proc, 0 );
+                        __cfa_io_flush( ctx->proc );
                 }
         }
@@ -656 +673 @@
                         return true;
                 }
+
+                void __cfa_io_idle( processor * proc ) {
+                        iovec iov;
+                        __atomic_acquire( &proc->io.ctx->cq.lock );
+
+                        with( this->idle_wctx) {
+
+                        // Do we already have a pending read
+                        if(available(*ftr)) {
+                                // There is no pending read, we need to add one
+                                reset(*ftr);
+
+                                iov.iov_base = rdbuf;
+                                iov.iov_len  = sizeof(eventfd_t);
+                                __kernel_read(proc, *ftr, iov, evfd );
+                        }
+
+                        __ioarbiter_flush( *proc->io.ctx );
+                        ioring_syscsll(ctx, 1, IORING_ENTER_GETEVENTS);
+
+                        __cfa_do_drain( proc->io.ctx, proc->cltr );
+                }
         #endif
 #endif