Changes in libcfa/src/concurrency/io/setup.cfa [dddb3dd0:2fab24e3]

File:

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

libcfa/src/concurrency/io/setup.cfa

@@ -26 +26 @@
 
 #if !defined(CFA_HAVE_LINUX_IO_URING_H)
+        void __kernel_io_startup() {
+                // Nothing to do without io_uring
+        }
+
+        void __kernel_io_shutdown() {
+                // Nothing to do without io_uring
+        }
+
         void ?{}(io_context_params & this) {}
 
-        void  ?{}($io_context & this, struct cluster & cl) {}
-        void ^?{}($io_context & this) {}
-
-        $io_arbiter * create(void) { return 0p; }
-        void destroy($io_arbiter *) {}
+        void ?{}(io_context & this, struct cluster & cl) {}
+        void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {}
+
+        void ^?{}(io_context & this) {}
+        void ^?{}(io_context & this, bool cluster_context) {}
+
+        void register_fixed_files( io_context &, int *, unsigned ) {}
+        void register_fixed_files( cluster    &, int *, unsigned ) {}
 
 #else
@@ -57 +68 @@
         void ?{}(io_context_params & this) {
                 this.num_entries = 256;
+                this.num_ready = 256;
+                this.submit_aff = -1;
+                this.eager_submits = false;
+                this.poller_submits = false;
+                this.poll_submit = false;
+                this.poll_complete = false;
         }
 
@@ -89 +106 @@
 
 //=============================================================================================
+// I/O Startup / Shutdown logic + Master Poller
+//=============================================================================================
+
+        // IO Master poller loop forward
+        static void * iopoll_loop( __attribute__((unused)) void * args );
+
+        static struct {
+                      pthread_t  thrd;    // pthread handle to io poller thread
+                      void *     stack;   // pthread stack for io poller thread
+                      int        epollfd; // file descriptor to the epoll instance
+                volatile     bool run;     // Whether or not to continue
+                volatile     bool stopped; // Whether the poller has finished running
+                volatile uint64_t epoch;   // Epoch used for memory reclamation
+        } iopoll;
+
+        void __kernel_io_startup(void) {
+                __cfadbg_print_safe(io_core, "Kernel : Creating EPOLL instance\n" );
+
+                iopoll.epollfd = epoll_create1(0);
+                if (iopoll.epollfd == -1) {
+                        abort( "internal error, epoll_create1\n");
+                }
+
+                __cfadbg_print_safe(io_core, "Kernel : Starting io poller thread\n" );
+
+                iopoll.stack   = __create_pthread( &iopoll.thrd, iopoll_loop, 0p );
+                iopoll.run     = true;
+                iopoll.stopped = false;
+                iopoll.epoch   = 0;
+        }
+
+        void __kernel_io_shutdown(void) {
+                // Notify the io poller thread of the shutdown
+                iopoll.run = false;
+                sigval val = { 1 };
+                pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
+
+                // Wait for the io poller thread to finish
+
+                __destroy_pthread( iopoll.thrd, iopoll.stack, 0p );
+
+                int ret = close(iopoll.epollfd);
+                if (ret == -1) {
+                        abort( "internal error, close epoll\n");
+                }
+
+                // Io polling is now fully stopped
+
+                __cfadbg_print_safe(io_core, "Kernel : IO poller stopped\n" );
+        }
+
+        static void * iopoll_loop( __attribute__((unused)) void * args ) {
+                __processor_id_t id;
+                id.full_proc = false;
+                id.id = doregister(&id);
+                __cfaabi_tls.this_proc_id = &id;
+                __cfadbg_print_safe(io_core, "Kernel : IO poller thread starting\n" );
+
+                // Block signals to control when they arrive
+                sigset_t mask;
+                sigfillset(&mask);
+                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
+                abort( "internal error, pthread_sigmask" );
+                }
+
+                sigdelset( &mask, SIGUSR1 );
+
+                // Create sufficient events
+                struct epoll_event events[10];
+                // Main loop
+                while( iopoll.run ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : waiting on io_uring contexts\n");
+
+                        // increment the epoch to notify any deleters we are starting a new cycle
+                        __atomic_fetch_add(&iopoll.epoch, 1, __ATOMIC_SEQ_CST);
+
+                        // Wait for events
+                        int nfds = epoll_pwait( iopoll.epollfd, events, 10, -1, &mask );
+
+                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : %d io contexts events, waking up\n", nfds);
+
+                        // Check if an error occured
+                        if (nfds == -1) {
+                                if( errno == EINTR ) continue;
+                                abort( "internal error, pthread_sigmask" );
+                        }
+
+                        for(i; nfds) {
+                                $io_ctx_thread * io_ctx = ($io_ctx_thread *)(uintptr_t)events[i].data.u64;
+                                /* paranoid */ verify( io_ctx );
+                                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Unparking io poller %d (%p)\n", io_ctx->ring->fd, io_ctx);
+                                #if !defined( __CFA_NO_STATISTICS__ )
+                                        __cfaabi_tls.this_stats = io_ctx->self.curr_cluster->stats;
+                                #endif
+
+                                eventfd_t v;
+                                eventfd_read(io_ctx->ring->efd, &v);
+
+                                post( io_ctx->sem );
+                        }
+                }
+
+                __atomic_store_n(&iopoll.stopped, true, __ATOMIC_SEQ_CST);
+
+                __cfadbg_print_safe(io_core, "Kernel : IO poller thread stopping\n" );
+                unregister(&id);
+                return 0p;
+        }
+
+//=============================================================================================
 // I/O Context Constrution/Destruction
 //=============================================================================================
 
-
-
-        static void __io_uring_setup ( $io_context & this, const io_context_params & params_in, int procfd );
-        static void __io_uring_teardown( $io_context & this );
-        static void __epoll_register($io_context & ctx);
-        static void __epoll_unregister($io_context & ctx);
-        void __ioarbiter_register( $io_arbiter & mutex, $io_context & ctx );
-        void __ioarbiter_unregister( $io_arbiter & mutex, $io_context & ctx );
-
-        void ?{}($io_context & this, processor * proc, struct cluster & cl) {
-                /* paranoid */ verify( cl.io.arbiter );
-                this.proc = proc;
-                this.arbiter = cl.io.arbiter;
-                this.ext_sq.empty = true;
-                (this.ext_sq.blocked){};
-                __io_uring_setup( this, cl.io.params, proc->idle );
-                __cfadbg_print_safe(io_core, "Kernel I/O : Created ring for io_context %u (%p)\n", this.fd, &this);
-        }
-
-        void ^?{}($io_context & this) {
-                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %u\n", this.fd);
-
-                __io_uring_teardown( this );
-                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %u\n", this.fd);
+        void ?{}($io_ctx_thread & this, struct cluster & cl) { (this.self){ "IO Poller", cl }; }
+        void main( $io_ctx_thread & this );
+        static inline $thread * get_thread( $io_ctx_thread & this ) { return &this.self; }
+        void ^?{}( $io_ctx_thread & mutex this ) {}
+
+        static void __io_create ( __io_data & this, const io_context_params & params_in );
+        static void __io_destroy( __io_data & this );
+
+        void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {
+                (this.thrd){ cl };
+                this.thrd.ring = malloc();
+                __cfadbg_print_safe(io_core, "Kernel I/O : Creating ring for io_context %p\n", &this);
+                __io_create( *this.thrd.ring, params );
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : Starting poller thread for io_context %p\n", &this);
+                this.thrd.done = false;
+                __thrd_start( this.thrd, main );
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : io_context %p ready\n", &this);
+        }
+
+        void ?{}(io_context & this, struct cluster & cl) {
+                io_context_params params;
+                (this){ cl, params };
+        }
+
+        void ^?{}(io_context & this, bool cluster_context) {
+                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %p\n", &this);
+
+                // Notify the thread of the shutdown
+                __atomic_store_n(&this.thrd.done, true, __ATOMIC_SEQ_CST);
+
+                // If this is an io_context within a cluster, things get trickier
+                $thread & thrd = this.thrd.self;
+                if( cluster_context ) {
+                        // We are about to do weird things with the threads
+                        // we don't need interrupts to complicate everything
+                        disable_interrupts();
+
+                        // Get cluster info
+                        cluster & cltr = *thrd.curr_cluster;
+                        /* paranoid */ verify( cltr.idles.total == 0 || &cltr == 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 ) {
+                                // This is the tricky case
+                                // The thread was preempted or ready to run and now it is on the ready queue
+                                // but the cluster is shutting down, so there aren't any processors to run the ready queue
+                                // the solution is to steal the thread from the ready-queue and pretend it was blocked all along
+
+                                ready_schedule_lock();
+                                        // The thread should on the list
+                                        /* paranoid */ verify( thrd.link.next != 0p );
+
+                                        // Remove the thread from the ready queue of this cluster
+                                        // The thread should be the last on the list
+                                        __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
+                                        /* paranoid */ verify( removed );
+                                        thrd.link.next = 0p;
+                                        thrd.link.prev = 0p;
+
+                                        // Fixup the thread state
+                                        thrd.state = Blocked;
+                                        thrd.ticket = TICKET_BLOCKED;
+                                        thrd.preempted = __NO_PREEMPTION;
+
+                                ready_schedule_unlock();
+
+                                // Pretend like the thread was blocked all along
+                        }
+                        // !!! This is not an else if !!!
+                        // Ok, now the thread is blocked (whether we cheated to get here or not)
+                        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 );
+                        }
+                        else {
+                                // The thread is in a weird state
+                                // I don't know what to do here
+                                abort("io_context poller thread is in unexpected state, cannot clean-up correctly\n");
+                        }
+
+                        // The weird thread kidnapping stuff is over, restore interrupts.
+                        enable_interrupts( __cfaabi_dbg_ctx );
+                } else {
+                        post( this.thrd.sem );
+                }
+
+                ^(this.thrd){};
+                __cfadbg_print_safe(io_core, "Kernel I/O : Stopped poller thread for io_context %p\n", &this);
+
+                __io_destroy( *this.thrd.ring );
+                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %p\n", &this);
+
+                free(this.thrd.ring);
+        }
+
+        void ^?{}(io_context & this) {
+                ^(this){ false };
         }
 
@@ -121 +329 @@
         extern void __enable_interrupts_hard();
 
-        static void __io_uring_setup( $io_context & this, const io_context_params & params_in, int procfd ) {
+        static void __io_create( __io_data & this, const io_context_params & params_in ) {
                 // Step 1 : call to setup
                 struct io_uring_params params;
                 memset(&params, 0, sizeof(params));
-                // if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
-                // if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
+                if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
+                if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
 
                 __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
@@ -132 +340 @@
                         abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
                 }
+                if( params_in.poller_submits && params_in.eager_submits ) {
+                        abort("ERROR: I/O setup 'poller_submits' and 'eager_submits' cannot be used together\n");
+                }
 
                 int fd = syscall(__NR_io_uring_setup, nentries, &params );
@@ -139 +350 @@
 
                 // Step 2 : mmap result
-                struct __sub_ring_t & sq = this.sq;
-                struct __cmp_ring_t & cq = this.cq;
+                memset( &this, 0, sizeof(struct __io_data) );
+                struct __submition_data  & sq = this.submit_q;
+                struct __completion_data & cq = this.completion_q;
 
                 // calculate the right ring size
@@ -189 +401 @@
                 // Get the pointers from the kernel to fill the structure
                 // submit queue
-                sq.kring.head  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
-                sq.kring.tail  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
-                sq.kring.array = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
-                sq.mask        = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
-                sq.num         = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
-                sq.flags       = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
-                sq.dropped     = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
-
-                sq.kring.released = 0;
-
-                sq.free_ring.head = 0;
-                sq.free_ring.tail = *sq.num;
-                sq.free_ring.array = alloc( *sq.num, 128`align );
-                for(i; (__u32)*sq.num) {
-                        sq.free_ring.array[i] = i;
-                }
-
-                sq.to_submit = 0;
+                sq.head    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
+                sq.tail    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
+                sq.mask    = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
+                sq.num     = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
+                sq.flags   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
+                sq.dropped = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
+                sq.array   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
+                sq.prev_head = *sq.head;
+
+                {
+                        const __u32 num = *sq.num;
+                        for( i; num ) {
+                                __sqe_clean( &sq.sqes[i] );
+                        }
+                }
+
+                (sq.submit_lock){};
+                (sq.release_lock){};
+
+                if( params_in.poller_submits || params_in.eager_submits ) {
+                        /* paranoid */ verify( is_pow2( params_in.num_ready ) || (params_in.num_ready < 8) );
+                        sq.ready_cnt = max( params_in.num_ready, 8 );
+                        sq.ready = alloc( sq.ready_cnt, 64`align );
+                        for(i; sq.ready_cnt) {
+                                sq.ready[i] = -1ul32;
+                        }
+                        sq.prev_ready = 0;
+                }
+                else {
+                        sq.ready_cnt = 0;
+                        sq.ready = 0p;
+                        sq.prev_ready = 0;
+                }
 
                 // completion queue
@@ -219 +446 @@
                 // io_uring_register is so f*cking slow on some machine that it
                 // will never succeed if preemption isn't hard blocked
-                __cfadbg_print_safe(io_core, "Kernel I/O : registering %d for completion with ring %d\n", procfd, fd);
-
                 __disable_interrupts_hard();
 
-                int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &procfd, 1);
+                int efd = eventfd(0, 0);
+                if (efd < 0) {
+                        abort("KERNEL ERROR: IO_URING EVENTFD - %s\n", strerror(errno));
+                }
+
+                int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &efd, 1);
                 if (ret < 0) {
                         abort("KERNEL ERROR: IO_URING EVENTFD REGISTER - %s\n", strerror(errno));
@@ -229 +459 @@
 
                 __enable_interrupts_hard();
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : registered %d for completion with ring %d\n", procfd, fd);
 
                 // some paranoid checks
@@ -240 +468 @@
                 /* 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.kring.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.kring.head );
-                /* paranoid */ verifyf( (*sq.kring.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.kring.tail );
+                /* 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.ring_flags = 0;
+                this.ring_flags = params.flags;
                 this.fd         = fd;
-        }
-
-        static void __io_uring_teardown( $io_context & this ) {
+                this.efd        = efd;
+                this.eager_submits  = params_in.eager_submits;
+                this.poller_submits = params_in.poller_submits;
+        }
+
+        static void __io_destroy( __io_data & this ) {
                 // Shutdown the io rings
-                struct __sub_ring_t & sq = this.sq;
-                struct __cmp_ring_t & cq = this.cq;
+                struct __submition_data  & sq = this.submit_q;
+                struct __completion_data & cq = this.completion_q;
 
                 // unmap the submit queue entries
@@ -266 +497 @@
                 // close the file descriptor
                 close(this.fd);
-
-                free( this.sq.free_ring.array ); // Maybe null, doesn't matter
-        }
-
-        void __cfa_io_start( processor * proc ) {
-                proc->io.ctx = alloc();
-                (*proc->io.ctx){proc, *proc->cltr};
-        }
-        void __cfa_io_stop ( processor * proc ) {
-                ^(*proc->io.ctx){};
-                free(proc->io.ctx);
+                close(this.efd);
+
+                free( this.submit_q.ready ); // Maybe null, doesn't matter
         }
 
@@ -282 +505 @@
 // I/O Context Sleep
 //=============================================================================================
-        // static inline void __epoll_ctl($io_context & ctx, int op, const char * error) {
-        //      struct epoll_event ev;
-        //      ev.events = EPOLLIN | EPOLLONESHOT;
-        //      ev.data.u64 = (__u64)&ctx;
-        //      int ret = epoll_ctl(iopoll.epollfd, op, ctx.efd, &ev);
-        //      if (ret < 0) {
-        //              abort( "KERNEL ERROR: EPOLL %s - (%d) %s\n", error, (int)errno, strerror(errno) );
-        //      }
-        // }
-
-        // static void __epoll_register($io_context & ctx) {
-        //      __epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
-        // }
-
-        // static void __epoll_unregister($io_context & ctx) {
-        //      // Read the current epoch so we know when to stop
-        //      size_t curr = __atomic_load_n(&iopoll.epoch, __ATOMIC_SEQ_CST);
-
-        //      // Remove the fd from the iopoller
-        //      __epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
-
-        //      // Notify the io poller thread of the shutdown
-        //      iopoll.run = false;
-        //      sigval val = { 1 };
-        //      pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
-
-        //      // Make sure all this is done
-        //      __atomic_thread_fence(__ATOMIC_SEQ_CST);
-
-        //      // Wait for the next epoch
-        //      while(curr == iopoll.epoch && !iopoll.stopped) Pause();
-        // }
-
-        // void __ioctx_prepare_block($io_context & ctx) {
-        //      __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.fd, &ctx);
-        //      __epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
-        // }
-
+        static inline void __ioctx_epoll_ctl($io_ctx_thread & ctx, int op, const char * error) {
+                struct epoll_event ev;
+                ev.events = EPOLLIN | EPOLLONESHOT;
+                ev.data.u64 = (__u64)&ctx;
+                int ret = epoll_ctl(iopoll.epollfd, op, ctx.ring->efd, &ev);
+                if (ret < 0) {
+                        abort( "KERNEL ERROR: EPOLL %s - (%d) %s\n", error, (int)errno, strerror(errno) );
+                }
+        }
+
+        void __ioctx_register($io_ctx_thread & ctx) {
+                __ioctx_epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
+        }
+
+        void __ioctx_prepare_block($io_ctx_thread & ctx) {
+                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.ring->fd, &ctx);
+                __ioctx_epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
+        }
+
+        void __ioctx_unregister($io_ctx_thread & ctx) {
+                // Read the current epoch so we know when to stop
+                size_t curr = __atomic_load_n(&iopoll.epoch, __ATOMIC_SEQ_CST);
+
+                // Remove the fd from the iopoller
+                __ioctx_epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
+
+                // Notify the io poller thread of the shutdown
+                iopoll.run = false;
+                sigval val = { 1 };
+                pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
+
+                // Make sure all this is done
+                __atomic_thread_fence(__ATOMIC_SEQ_CST);
+
+                // Wait for the next epoch
+                while(curr == iopoll.epoch && !iopoll.stopped) Pause();
+        }
 
 //=============================================================================================
 // I/O Context Misc Setup
 //=============================================================================================
-        void ?{}( $io_arbiter & this ) {
-                this.pending.flag = false;
-        }
-
-        void ^?{}( $io_arbiter & mutex this ) {
-                // /* paranoid */ verify( empty(this.assigned) );
-                // /* paranoid */ verify( empty(this.available) );
-                /* paranoid */ verify( is_empty(this.pending.blocked) );
-        }
-
-        $io_arbiter * create(void) {
-                return new();
-        }
-        void destroy($io_arbiter * arbiter) {
-                delete(arbiter);
-        }
-
-//=============================================================================================
-// I/O Context Misc Setup
-//=============================================================================================
-
+        void register_fixed_files( io_context & ctx, int * files, unsigned count ) {
+                int ret = syscall( __NR_io_uring_register, ctx.thrd.ring->fd, IORING_REGISTER_FILES, files, count );
+                if( ret < 0 ) {
+                        abort( "KERNEL ERROR: IO_URING REGISTER - (%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 );
+        }
+
+        void register_fixed_files( cluster & cltr, int * files, unsigned count ) {
+                for(i; cltr.io.cnt) {
+                        register_fixed_files( cltr.io.ctxs[i], files, count );
+                }
+        }
 #endif