Changes in libcfa/src/concurrency/io.cfa [a757ba1:a55472cc]

File:

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

libcfa/src/concurrency/io.cfa

@@ -610 +610 @@
                 if( we ) {
                         sigval_t value = { PREEMPT_IO };
-                        __cfaabi_pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
+                        pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
                 }
 
@@ -639 +639 @@
                 }
         }
+
+        #if defined(CFA_WITH_IO_URING_IDLE)
+                bool __kernel_read(struct processor * proc, io_future_t & future, iovec & iov, int fd) {
+                        io_context$ * ctx = proc->io.ctx;
+                        /* paranoid */ verify( ! __preemption_enabled() );
+                        /* paranoid */ verify( proc == __cfaabi_tls.this_processor );
+                        /* paranoid */ verify( ctx );
+
+                        __u32 idx;
+                        struct io_uring_sqe * sqe;
+
+                        // We can proceed to the fast path
+                        if( !__alloc(ctx, &idx, 1) ) {
+                                /* paranoid */ verify( false ); // for now check if this happens, next time just abort the sleep.
+                                return false;
+                        }
+
+                        // Allocation was successful
+                        __fill( &sqe, 1, &idx, ctx );
+
+                        sqe->user_data = (uintptr_t)&future;
+                        sqe->flags = 0;
+                        sqe->fd = fd;
+                        sqe->off = 0;
+                        sqe->ioprio = 0;
+                        sqe->fsync_flags = 0;
+                        sqe->__pad2[0] = 0;
+                        sqe->__pad2[1] = 0;
+                        sqe->__pad2[2] = 0;
+
+                        #if defined(CFA_HAVE_IORING_OP_READ)
+                                sqe->opcode = IORING_OP_READ;
+                                sqe->addr = (uint64_t)iov.iov_base;
+                                sqe->len = iov.iov_len;
+                        #elif defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV)
+                                sqe->opcode = IORING_OP_READV;
+                                sqe->addr = (uintptr_t)&iov;
+                                sqe->len = 1;
+                        #else
+                                #error CFA_WITH_IO_URING_IDLE but none of CFA_HAVE_READV, CFA_HAVE_IORING_OP_READV or CFA_HAVE_IORING_OP_READ defined
+                        #endif
+
+                        asm volatile("": : :"memory");
+
+                        /* paranoid */ verify( sqe->user_data == (uintptr_t)&future );
+                        __submit_only( ctx, &idx, 1 );
+
+                        /* paranoid */ verify( proc == __cfaabi_tls.this_processor );
+                        /* paranoid */ verify( ! __preemption_enabled() );
+
+                        return true;
+                }
+
+                void __cfa_io_idle( struct processor * proc ) {
+                        iovec iov;
+                        __atomic_acquire( &proc->io.ctx->cq.lock );
+
+                        __attribute__((used)) volatile bool was_reset = false;
+
+                        with( proc->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 );
+                                        ftr->result = 0xDEADDEAD;
+                                        *((eventfd_t *)rdbuf) = 0xDEADDEADDEADDEAD;
+                                        was_reset = true;
+                                }
+                        }
+
+                        if( !__atomic_load_n( &proc->do_terminate, __ATOMIC_SEQ_CST ) ) {
+                                __ioarbiter_flush( *proc->io.ctx );
+                                proc->idle_wctx.sleep_time = rdtscl();
+                                ioring_syscsll( *proc->io.ctx, 1, IORING_ENTER_GETEVENTS);
+                        }
+
+                        ready_schedule_lock();
+                        __cfa_do_drain( proc->io.ctx, proc->cltr );
+                        ready_schedule_unlock();
+
+                        asm volatile ("" :: "m" (was_reset));
+                }
+        #endif
 #endif