Changeset 5c581cc for libcfa

Timestamp:

May 11, 2020, 3:33:21 PM (4 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

70ac8d0

Parents:

6c12fd28

Message:

Fixed spinning of slow poller and added random offset to submit to avoid clashing

File:

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

libcfa/src/concurrency/io.cfa

@@ -18 +18 @@
 
 #include "kernel.hfa"
+#include "bitmanip.hfa"
 
 #if !defined(HAVE_LINUX_IO_URING_H)
@@ -196 +197 @@
                                 void * stack;
                                 pthread_t kthrd;
+                                volatile bool blocked;
                         } slow;
                         __io_poller_fast fast;
@@ -280 +282 @@
 
                 if( io_flags & CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS ) {
+                        /* paranoid */ verify( is_pow2( io_flags >> CFA_CLUSTER_IO_BUFFLEN_OFFSET ) || ((io_flags >> CFA_CLUSTER_IO_BUFFLEN_OFFSET) < 8)  );
                         sq.ready_cnt = max(io_flags >> CFA_CLUSTER_IO_BUFFLEN_OFFSET, 8);
                         sq.ready = alloc_align( 64, sq.ready_cnt );
@@ -344 +347 @@
                 // Create the poller thread
                 __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
+                this.io->poller.slow.blocked = false;
                 this.io->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );
         }
@@ -580 +584 @@
                         while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
 
+                                __atomic_store_n( &ring.poller.slow.blocked, true, __ATOMIC_SEQ_CST );
+
                                 // In the user-thread approach drain and if anything was drained,
                                 // batton pass to the user-thread
                                 int count;
                                 bool again;
-                                [count, again] = __drain_io( ring, &mask, 0, true );
+                                [count, again] = __drain_io( ring, &mask, 1, true );
+
+                                __atomic_store_n( &ring.poller.slow.blocked, false, __ATOMIC_SEQ_CST );
 
                                 // Update statistics
@@ -667 +675 @@
         static inline void __wake_poller( struct __io_data & ring ) __attribute__((artificial));
         static inline void __wake_poller( struct __io_data & ring ) {
-                // sigval val = { 1 };
-                // pthread_sigqueue( ring.poller.slow.kthrd, SIGUSR1, val );
+                if(!__atomic_load_n( &ring.poller.slow.blocked, __ATOMIC_SEQ_CST)) return;
+
+                sigval val = { 1 };
+                pthread_sigqueue( ring.poller.slow.kthrd, SIGUSR1, val );
         }
 
@@ -732 +742 @@
                         __attribute((unused)) int block = 0;
                         uint32_t expected = -1ul32;
-                        LOOKING: for(;;) {
+                        uint32_t ready_mask = ring.submit_q.ready_cnt - 1;
+                        uint32_t off = __tls_rand();
+                        LOOKING: for() {
                                 for(i; ring.submit_q.ready_cnt) {
-                                        if( __atomic_compare_exchange_n( &ring.submit_q.ready[i], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
+                                        uint32_t ii = (i + off) & ready_mask;
+                                        if( __atomic_compare_exchange_n( &ring.submit_q.ready[ii], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
                                                 break LOOKING;
                                         }