Changes in libcfa/src/concurrency/io.cfa [3e2b9c9:5751a56]

File:

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

libcfa/src/concurrency/io.cfa

@@ -14 +14 @@
 //
 
-#define __cforall_thread__
-
 #if defined(__CFA_DEBUG__)
         // #define __CFA_DEBUG_PRINT_IO__
@@ -21 +19 @@
 #endif
 
-
-#if defined(CFA_HAVE_LINUX_IO_URING_H)
+#include "kernel.hfa"
+#include "bitmanip.hfa"
+
+#if !defined(CFA_HAVE_LINUX_IO_URING_H)
+        void __kernel_io_startup( cluster &, unsigned, bool ) {
+                // Nothing to do without io_uring
+        }
+
+        void __kernel_io_finish_start( cluster & ) {
+                // Nothing to do without io_uring
+        }
+
+        void __kernel_io_prepare_stop( cluster & ) {
+                // Nothing to do without io_uring
+        }
+
+        void __kernel_io_shutdown( cluster &, bool ) {
+                // Nothing to do without io_uring
+        }
+
+#else
         #define _GNU_SOURCE         /* See feature_test_macros(7) */
         #include <errno.h>
-        #include <signal.h>
         #include <stdint.h>
         #include <string.h>
         #include <unistd.h>
+        #include <sys/mman.h>
 
         extern "C" {
-                #include <sys/epoll.h>
                 #include <sys/syscall.h>
 
@@ -37 +53 @@
         }
 
-        #include "stats.hfa"
-        #include "kernel.hfa"
-        #include "kernel/fwd.hfa"
-        #include "io/types.hfa"
+        #include "bits/signal.hfa"
+        #include "kernel_private.hfa"
+        #include "thread.hfa"
+
+        uint32_t entries_per_cluster() {
+                return 256;
+        }
+
+        static void * __io_poller_slow( void * arg );
+
+        // Weirdly, some systems that do support io_uring don't actually define these
+        #ifdef __alpha__
+                /*
+                * alpha is the only exception, all other architectures
+                * have common numbers for new system calls.
+                */
+                #ifndef __NR_io_uring_setup
+                        #define __NR_io_uring_setup           535
+                #endif
+                #ifndef __NR_io_uring_enter
+                        #define __NR_io_uring_enter           536
+                #endif
+                #ifndef __NR_io_uring_register
+                        #define __NR_io_uring_register        537
+                #endif
+        #else /* !__alpha__ */
+                #ifndef __NR_io_uring_setup
+                        #define __NR_io_uring_setup           425
+                #endif
+                #ifndef __NR_io_uring_enter
+                        #define __NR_io_uring_enter           426
+                #endif
+                #ifndef __NR_io_uring_register
+                        #define __NR_io_uring_register        427
+                #endif
+        #endif
+
+        // Fast poller user-thread
+        // Not using the "thread" keyword because we want to control
+        // more carefully when to start/stop it
+        struct __io_poller_fast {
+                struct __io_data * ring;
+                $thread thrd;
+        };
+
+        void ?{}( __io_poller_fast & this, struct cluster & cltr ) {
+                this.ring = cltr.io;
+                (this.thrd){ "Fast I/O Poller", cltr };
+        }
+        void ^?{}( __io_poller_fast & mutex this );
+        void main( __io_poller_fast & this );
+        static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; }
+        void ^?{}( __io_poller_fast & mutex this ) {}
+
+        struct __submition_data {
+                // Head and tail of the ring (associated with array)
+                volatile uint32_t * head;
+                volatile uint32_t * tail;
+                volatile uint32_t prev_head;
+
+                // The actual kernel ring which uses head/tail
+                // indexes into the sqes arrays
+                uint32_t * array;
+
+                // number of entries and mask to go with it
+                const uint32_t * num;
+                const uint32_t * mask;
+
+                // Submission flags (Not sure what for)
+                uint32_t * flags;
+
+                // number of sqes not submitted (whatever that means)
+                uint32_t * dropped;
+
+                // Like head/tail but not seen by the kernel
+                volatile uint32_t * ready;
+                uint32_t ready_cnt;
+
+                __spinlock_t lock;
+                __spinlock_t release_lock;
+
+                // A buffer of sqes (not the actual ring)
+                struct io_uring_sqe * sqes;
+
+                // The location and size of the mmaped area
+                void * ring_ptr;
+                size_t ring_sz;
+        };
+
+        struct __completion_data {
+                // Head and tail of the ring
+                volatile uint32_t * head;
+                volatile uint32_t * tail;
+
+                // number of entries and mask to go with it
+                const uint32_t * mask;
+                const uint32_t * num;
+
+                // number of cqes not submitted (whatever that means)
+                uint32_t * overflow;
+
+                // the kernel ring
+                struct io_uring_cqe * cqes;
+
+                // The location and size of the mmaped area
+                void * ring_ptr;
+                size_t ring_sz;
+        };
+
+        struct __io_data {
+                struct __submition_data submit_q;
+                struct __completion_data completion_q;
+                uint32_t ring_flags;
+                int cltr_flags;
+                int fd;
+                semaphore submit;
+                volatile bool done;
+                struct {
+                        struct {
+                                __processor_id_t id;
+                                void * stack;
+                                pthread_t kthrd;
+                                volatile bool blocked;
+                        } slow;
+                        __io_poller_fast fast;
+                        __bin_sem_t sem;
+                } poller;
+        };
 
 //=============================================================================================
-// I/O Syscall
+// I/O Startup / Shutdown logic
 //=============================================================================================
-        static int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get ) {
+        void __kernel_io_startup( cluster & this, unsigned io_flags, bool main_cluster ) {
+                if( (io_flags & CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS) && (io_flags & CFA_CLUSTER_IO_EAGER_SUBMITS) ) {
+                        abort("CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS and CFA_CLUSTER_IO_EAGER_SUBMITS cannot be mixed\n");
+                }
+
+                this.io = malloc();
+
+                // Step 1 : call to setup
+                struct io_uring_params params;
+                memset(&params, 0, sizeof(params));
+                if( io_flags & CFA_CLUSTER_IO_KERNEL_POLL_SUBMITS   ) params.flags |= IORING_SETUP_SQPOLL;
+                if( io_flags & CFA_CLUSTER_IO_KERNEL_POLL_COMPLETES ) params.flags |= IORING_SETUP_IOPOLL;
+
+                uint32_t nentries = entries_per_cluster();
+
+                int fd = syscall(__NR_io_uring_setup, nentries, &params );
+                if(fd < 0) {
+                        abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
+                }
+
+                // Step 2 : mmap result
+                memset( this.io, 0, sizeof(struct __io_data) );
+                struct __submition_data  & sq = this.io->submit_q;
+                struct __completion_data & cq = this.io->completion_q;
+
+                // calculate the right ring size
+                sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned)           );
+                cq.ring_sz = params.cq_off.cqes  + (params.cq_entries * sizeof(struct io_uring_cqe));
+
+                // Requires features
+                #if defined(IORING_FEAT_SINGLE_MMAP)
+                        // adjust the size according to the parameters
+                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
+                                cq.ring_sz = sq.ring_sz = max(cq.ring_sz, sq.ring_sz);
+                        }
+                #endif
+
+                // mmap the Submit Queue into existence
+                sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING);
+                if (sq.ring_ptr == (void*)MAP_FAILED) {
+                        abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
+                }
+
+                // Requires features
+                #if defined(IORING_FEAT_SINGLE_MMAP)
+                        // mmap the Completion Queue into existence (may or may not be needed)
+                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
+                                cq.ring_ptr = sq.ring_ptr;
+                        }
+                        else
+                #endif
+                {
+                        // We need multiple call to MMAP
+                        cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING);
+                        if (cq.ring_ptr == (void*)MAP_FAILED) {
+                                munmap(sq.ring_ptr, sq.ring_sz);
+                                abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
+                        }
+                }
+
+                // mmap the submit queue entries
+                size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
+                sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES);
+                if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
+                        munmap(sq.ring_ptr, sq.ring_sz);
+                        if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
+                        abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
+                }
+
+                // Get the pointers from the kernel to fill the structure
+                // submit queue
+                sq.head    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
+                sq.tail    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
+                sq.mask    = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
+                sq.num     = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
+                sq.flags   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
+                sq.dropped = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
+                sq.array   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
+                sq.prev_head = *sq.head;
+
+                {
+                        const uint32_t num = *sq.num;
+                        for( i; num ) {
+                                sq.sqes[i].user_data = 0ul64;
+                        }
+                }
+
+                (sq.lock){};
+                (sq.release_lock){};
+
+                if( io_flags & ( CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS | CFA_CLUSTER_IO_EAGER_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 );
+                        for(i; sq.ready_cnt) {
+                                sq.ready[i] = -1ul32;
+                        }
+                }
+                else {
+                        sq.ready_cnt = 0;
+                        sq.ready = 0p;
+                }
+
+                // completion queue
+                cq.head     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
+                cq.tail     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
+                cq.mask     = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
+                cq.num      = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
+                cq.overflow = (         uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
+                cq.cqes   = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
+
+                // some paranoid checks
+                /* paranoid */ verifyf( (*cq.mask) == ((*cq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*cq.num) - 1ul32, *cq.num, *cq.mask  );
+                /* paranoid */ verifyf( (*cq.num)  >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
+                /* paranoid */ verifyf( (*cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
+                /* paranoid */ verifyf( (*cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
+
+                /* paranoid */ verifyf( (*sq.mask) == ((*sq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*sq.num) - 1ul32, *sq.num, *sq.mask );
+                /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
+                /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
+                /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
+
+                // Update the global ring info
+                this.io->ring_flags = params.flags;
+                this.io->cltr_flags = io_flags;
+                this.io->fd         = fd;
+                this.io->done       = false;
+                (this.io->submit){ min(*sq.num, *cq.num) };
+
+                if(!main_cluster) {
+                        __kernel_io_finish_start( this );
+                }
+        }
+
+        void __kernel_io_finish_start( cluster & this ) {
+                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
+                        (this.io->poller.fast){ this };
+                        __thrd_start( this.io->poller.fast, main );
+                }
+
+                // Create the poller thread
+                __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluster %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 );
+        }
+
+        void __kernel_io_prepare_stop( cluster & this ) {
+                __cfadbg_print_safe(io_core, "Kernel I/O : Stopping pollers for cluster\n", &this);
+                // Notify the poller thread of the shutdown
+                __atomic_store_n(&this.io->done, true, __ATOMIC_SEQ_CST);
+
+                // Stop the IO Poller
+                sigval val = { 1 };
+                pthread_sigqueue( this.io->poller.slow.kthrd, SIGUSR1, val );
+                post( this.io->poller.sem );
+
+                // Wait for the poller thread to finish
+                pthread_join( this.io->poller.slow.kthrd, 0p );
+                free( this.io->poller.slow.stack );
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this);
+
+                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
+                        with( this.io->poller.fast ) {
+                                /* paranoid */ verify( this.nprocessors == 0 || &this == mainCluster );
+                                /* paranoid */ verify( !ready_mutate_islocked() );
+
+                                // We need to adjust the clean-up based on where the thread is
+                                if( thrd.state == Ready || thrd.preempted != __NO_PREEMPTION ) {
+
+                                        ready_schedule_lock( (struct __processor_id_t *)active_processor() );
+
+                                                // This is the tricky case
+                                                // The thread was preempted and now it is on the ready queue
+                                                // The thread should be the last on the list
+                                                /* paranoid */ verify( thrd.link.next != 0p );
+
+                                                // Remove the thread from the ready queue of this cluster
+                                                __attribute__((unused)) bool removed = remove_head( &this, &thrd );
+                                                /* paranoid */ verify( removed );
+                                                thrd.link.next = 0p;
+                                                thrd.link.prev = 0p;
+                                                __cfaabi_dbg_debug_do( thrd.unpark_stale = true );
+
+                                                // Fixup the thread state
+                                                thrd.state = Blocked;
+                                                thrd.ticket = 0;
+                                                thrd.preempted = __NO_PREEMPTION;
+
+                                        ready_schedule_unlock( (struct __processor_id_t *)active_processor() );
+
+                                        // Pretend like the thread was blocked all along
+                                }
+                                // !!! This is not an else if !!!
+                                if( thrd.state == Blocked ) {
+
+                                        // This is the "easy case"
+                                        // The thread is parked and can easily be moved to active cluster
+                                        verify( thrd.curr_cluster != active_cluster() || thrd.curr_cluster == mainCluster );
+                                        thrd.curr_cluster = active_cluster();
+
+                                        // unpark the fast io_poller
+                                        unpark( &thrd __cfaabi_dbg_ctx2 );
+                                }
+                                else {
+
+                                        // The thread is in a weird state
+                                        // I don't know what to do here
+                                        abort("Fast poller thread is in unexpected state, cannot clean-up correctly\n");
+                                }
+
+                        }
+
+                        ^(this.io->poller.fast){};
+
+                        __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this);
+                }
+        }
+
+        void __kernel_io_shutdown( cluster & this, bool main_cluster ) {
+                if(!main_cluster) {
+                        __kernel_io_prepare_stop( this );
+                }
+
+                // Shutdown the io rings
+                struct __submition_data  & sq = this.io->submit_q;
+                struct __completion_data & cq = this.io->completion_q;
+
+                // unmap the submit queue entries
+                munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe));
+
+                // unmap the Submit Queue ring
+                munmap(sq.ring_ptr, sq.ring_sz);
+
+                // unmap the Completion Queue ring, if it is different
+                if (cq.ring_ptr != sq.ring_ptr) {
+                        munmap(cq.ring_ptr, cq.ring_sz);
+                }
+
+                // close the file descriptor
+                close(this.io->fd);
+
+                free( this.io->submit_q.ready ); // Maybe null, doesn't matter
+                free( this.io );
+        }
+
+        int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get, sigset_t * mask ) {
                 bool need_sys_to_submit = false;
                 bool need_sys_to_complete = false;
+                unsigned min_complete = 0;
                 unsigned flags = 0;
+
 
                 TO_SUBMIT:
@@ -62 +451 @@
                 }
 
+                TO_COMPLETE:
                 if( get && !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
                         flags |= IORING_ENTER_GETEVENTS;
+                        if( mask ) {
+                                need_sys_to_complete = true;
+                                min_complete = 1;
+                                break TO_COMPLETE;
+                        }
                         if( (ring.ring_flags & IORING_SETUP_IOPOLL) ) {
                                 need_sys_to_complete = true;
@@ -71 +466 @@
                 int ret = 0;
                 if( need_sys_to_submit || need_sys_to_complete ) {
-                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, 0p, _NSIG / 8);
+                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, min_complete, flags, mask, _NSIG / 8);
                         if( ret < 0 ) {
                                 switch((int)errno) {
@@ -95 +490 @@
         static uint32_t __release_consumed_submission( struct __io_data & ring );
 
-        static inline void process(struct io_uring_cqe & cqe ) {
+        static inline void process(struct io_uring_cqe & cqe, struct __processor_id_t * id ) {
                 struct __io_user_data_t * data = (struct __io_user_data_t *)(uintptr_t)cqe.user_data;
                 __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
 
                 data->result = cqe.res;
-                unpark( data->thrd __cfaabi_dbg_ctx2 );
+                if(!id) { unpark(     data->thrd __cfaabi_dbg_ctx2 ); }
+                else  { __unpark( id, data->thrd __cfaabi_dbg_ctx2 ); }
         }
 
         // Process a single completion message from the io_uring
         // This is NOT thread-safe
-        static [int, bool] __drain_io( & struct __io_data ring ) {
+        static [int, bool] __drain_io( & struct __io_data ring, * sigset_t mask ) {
                 /* paranoid */ verify( !kernelTLS.preemption_state.enabled );
 
                 unsigned to_submit = 0;
-                if( ring.poller_submits ) {
+                if( ring.cltr_flags & CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS ) {
                         // If the poller thread also submits, then we need to aggregate the submissions which are ready
                         to_submit = __collect_submitions( ring );
                 }
 
-                int ret = __io_uring_enter(ring, to_submit, true);
+                int ret = __io_uring_enter(ring, to_submit, true, mask);
                 if( ret < 0 ) {
                         return [0, true];
@@ -151 +547 @@
                         /* paranoid */ verify(&cqe);
 
-                        process( cqe );
-                }
+                        process( cqe, !mask ? (struct __processor_id_t *)0p : &ring.poller.slow.id );
+                }
+
+                // Allow new submissions to happen
+                // V(ring.submit, count);
 
                 // Mark to the kernel that the cqe has been seen
@@ -162 +561 @@
         }
 
-        void main( $io_ctx_thread & this ) {
-                epoll_event ev;
-                __ioctx_register( this, ev );
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %p for ring %p ready\n", &this, &this.ring);
+        static void * __io_poller_slow( void * arg ) {
+                #if !defined( __CFA_NO_STATISTICS__ )
+                        __stats_t local_stats;
+                        __init_stats( &local_stats );
+                        kernelTLS.this_stats = &local_stats;
+                #endif
+
+                cluster * cltr = (cluster *)arg;
+                struct __io_data & ring = *cltr->io;
+
+                ring.poller.slow.id.id = doregister( &ring.poller.slow.id );
+
+                sigset_t mask;
+                sigfillset(&mask);
+                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
+                        abort( "KERNEL ERROR: IO_URING - pthread_sigmask" );
+                }
+
+                sigdelset( &mask, SIGUSR1 );
+
+                verify( (*ring.submit_q.head) == (*ring.submit_q.tail) );
+                verify( (*ring.completion_q.head) == (*ring.completion_q.tail) );
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);
+
+                if( ring.cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
+                        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 );
+
+                                __atomic_store_n( &ring.poller.slow.blocked, false, __ATOMIC_SEQ_CST );
+
+                                // Update statistics
+                                __STATS__( true,
+                                        io.complete_q.completed_avg.val += count;
+                                        io.complete_q.completed_avg.slow_cnt += 1;
+                                )
+
+                                if(again) {
+                                        __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to fast poller\n", &ring);
+                                        __unpark( &ring.poller.slow.id, &ring.poller.fast.thrd __cfaabi_dbg_ctx2 );
+                                        wait( ring.poller.sem );
+                                }
+                        }
+                }
+                else {
+                        while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
+                                //In the naive approach, just poll the io completion queue directly
+                                int count;
+                                bool again;
+                                [count, again] = __drain_io( ring, &mask );
+
+                                // Update statistics
+                                __STATS__( true,
+                                        io.complete_q.completed_avg.val += count;
+                                        io.complete_q.completed_avg.slow_cnt += 1;
+                                )
+                        }
+                }
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p stopping\n", &ring);
+
+                unregister( &ring.poller.slow.id );
+
+                #if !defined(__CFA_NO_STATISTICS__)
+                        __tally_stats(cltr->stats, &local_stats);
+                #endif
+
+                return 0p;
+        }
+
+        void main( __io_poller_fast & this ) {
+                verify( this.ring->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD );
+
+                // Start parked
+                park( __cfaabi_dbg_ctx );
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p ready\n", &this.ring);
 
                 int reset = 0;
+
                 // Then loop until we need to start
-                while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
+                while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) {
+
                         // Drain the io
                         int count;
                         bool again;
                         disable_interrupts();
-                                [count, again] = __drain_io( *this.ring );
+                                [count, again] = __drain_io( *this.ring, 0p );
 
                                 if(!again) reset++;
@@ -192 +672 @@
                         // We didn't get anything baton pass to the slow poller
                         else {
-                                __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %p\n", &this.self);
+                                __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring);
                                 reset = 0;
 
-                                // block this thread
-                                __ioctx_prepare_block( this, ev );
-                                wait( this.sem );
+                                // wake up the slow poller
+                                post( this.ring->poller.sem );
+
+                                // park this thread
+                                park( __cfaabi_dbg_ctx );
                         }
                 }
 
                 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring);
+        }
+
+        static inline void __wake_poller( struct __io_data & ring ) __attribute__((artificial));
+        static inline void __wake_poller( struct __io_data & ring ) {
+                if(!__atomic_load_n( &ring.poller.slow.blocked, __ATOMIC_SEQ_CST)) return;
+
+                sigval val = { 1 };
+                pthread_sigqueue( ring.poller.slow.kthrd, SIGUSR1, val );
         }
 
@@ -316 +806 @@
         }
 
-        void __submit( struct io_context * ctx, uint32_t idx ) __attribute__((nonnull (1))) {
-                __io_data & ring = *ctx->thrd.ring;
+        void __submit( struct __io_data & ring, uint32_t idx ) {
                 // Get now the data we definetely need
                 uint32_t * const tail = ring.submit_q.tail;
-                const uint32_t mask  = *ring.submit_q.mask;
+                const uint32_t mask = *ring.submit_q.mask;
 
                 // There are 2 submission schemes, check which one we are using
-                if( ring.poller_submits ) {
+                if( ring.cltr_flags & CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS ) {
                         // If the poller thread submits, then we just need to add this to the ready array
                         __submit_to_ready_array( ring, idx, mask );
 
-                        post( ctx->thrd.sem );
+                        __wake_poller( ring );
 
                         __cfadbg_print_safe( io, "Kernel I/O : Added %u to ready for %p\n", idx, active_thread() );
                 }
-                else if( ring.eager_submits ) {
+                else if( ring.cltr_flags & CFA_CLUSTER_IO_EAGER_SUBMITS ) {
                         uint32_t picked = __submit_to_ready_array( ring, idx, mask );
 
@@ -360 +849 @@
                         // We got the lock
                         unsigned to_submit = __collect_submitions( ring );
-                        int ret = __io_uring_enter( ring, to_submit, false );
+                        int ret = __io_uring_enter( ring, to_submit, false, 0p );
                         if( ret < 0 ) {
                                 unlock(ring.submit_q.lock);
@@ -403 +892 @@
 
                         // Submit however, many entries need to be submitted
-                        int ret = __io_uring_enter( ring, 1, false );
+                        int ret = __io_uring_enter( ring, 1, false, 0p );
                         if( ret < 0 ) {
                                 switch((int)errno) {
@@ -469 +958 @@
                 return count;
         }
+
+//=============================================================================================
+// I/O Submissions
+//=============================================================================================
+
+        void register_fixed_files( cluster & cl, int * files, unsigned count ) {
+                int ret = syscall( __NR_io_uring_register, cl.io->fd, IORING_REGISTER_FILES, files, count );
+                if( ret < 0 ) {
+                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
+                }
+
+                __cfadbg_print_safe( io_core, "Kernel I/O : Performed io_register for %p, returned %d\n", active_thread(), ret );
+        }
 #endif