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

Timestamp:

Jan 9, 2021, 4:27:57 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:

561dd26

Parents:

35fd2c4

Message:

Added prints.
Naive implementation of cancel.
Server now shutdown cleanly.

File:

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

libcfa/src/concurrency/io.cfa

@@ -134 +134 @@
                 int ret = 0;
                 if( need_sys_to_submit || need_sys_to_complete ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING enter %d %u %u\n", ring.fd, to_submit, flags);
                         ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, (sigset_t *)0p, _NSIG / 8);
                         if( ret < 0 ) {
@@ -230 +231 @@
                 __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %p for ring %p ready\n", &this, &this.ring);
 
-                int reset = 0;
+                const int reset_cnt = 5;
+                int reset = reset_cnt;
                 // Then loop until we need to start
+                LOOP:
                 while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
                         // Drain the io
@@ -239 +242 @@
                                 [count, again] = __drain_io( *this.ring );
 
-                                if(!again) reset++;
+                                if(!again) reset--;
 
                                 // Update statistics
@@ -249 +252 @@
 
                         // If we got something, just yield and check again
-                        if(reset < 5) {
+                        if(reset > 1) {
                                 yield();
-                        }
-                        // We didn't get anything baton pass to the slow poller
-                        else {
+                                continue LOOP;
+                        }
+
+                        // We alread failed to find events a few time.
+                        if(reset == 1) {
+                                // Rearm the context so it can block
+                                // but don't block right away
+                                // we need to retry one last time in case
+                                // something completed *just now*
+                                __ioctx_prepare_block( this, ev );
+                                continue LOOP;
+                        }
+
                                 __STATS__( false,
                                         io.complete_q.blocks += 1;
                                 )
                                 __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %p\n", &this.self);
-                                reset = 0;
 
                                 // block this thread
-                                __ioctx_prepare_block( this, ev );
                                 wait( this.sem );
-                        }
+
+                        // restore counter
+                        reset = reset_cnt;
                 }
 
@@ -319 +332 @@
                                         )
 
+                                        __cfadbg_print_safe( io, "Kernel I/O : allocated [%p, %u] for %p (%p)\n", sqe, idx, active_thread(), (void*)data );
 
                                         // Success return the data
@@ -376 +390 @@
 
         void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
+                __cfadbg_print_safe( io, "Kernel I/O : submitting %u for %p\n", idx, active_thread() );
+
                 __io_data & ring = *ctx->thrd.ring;
                 // Get now the data we definetely need
@@ -443 +459 @@
                                 unlock(ring.submit_q.submit_lock);
                         #endif
-                        if( ret < 0 ) return;
+                        if( ret < 0 ) {
+                                return;
+                        }
 
                         // Release the consumed SQEs
@@ -454 +472 @@
                                 io.submit_q.submit_avg.cnt += 1;
                         )
+
+                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u (among %u) for %p\n", idx, ret, active_thread() );
                 }
                 else {