Changeset 342af53 for libcfa/src/concurrency/io/setup.cfa

Timestamp:

Jan 14, 2021, 12:23:14 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:

8e4aa05

Parents:

4468a70 (diff), ec19b21 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

File:

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

libcfa/src/concurrency/io/setup.cfa

@@ -52 +52 @@
                 #include <pthread.h>
                 #include <sys/epoll.h>
+                #include <sys/eventfd.h>
                 #include <sys/mman.h>
                 #include <sys/syscall.h>
@@ -169 +170 @@
                 // Main loop
                 while( iopoll.run ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : waiting on io_uring contexts\n");
+
                         // 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
@@ -181 +186 @@
                                 $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 : Unparking io poller %p\n", 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 );
                         }
@@ -233 +242 @@
                 $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 );
@@ -239 +253 @@
                         // 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();
-
-                                        // This is the tricky case
-                                        // The thread was preempted and now it is on the ready queue
+                                        // 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
-                                        /* paranoid */ verify( thrd.link.next != 0p );
-
-                                        // Remove the thread from the ready queue of this cluster
                                         __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
                                         /* paranoid */ verify( removed );
@@ -263 +279 @@
                         }
                         // !!! 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
@@ -274 +290 @@
                         }
                         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 );
@@ -365 +383 @@
                 }
 
+                // Step 3 : Initialize the data structure
                 // Get the pointers from the kernel to fill the structure
                 // submit queue
@@ -379 +398 @@
                         const __u32 num = *sq.num;
                         for( i; num ) {
-                                sq.sqes[i].user_data = 0ul64;
+                                __sqe_clean( &sq.sqes[i] );
                         }
                 }
@@ -409 +428 @@
                 cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
 
+                // Step 4 : eventfd
+                int efd;
+                for() {
+                        efd = eventfd(0, 0);
+                        if (efd < 0) {
+                                if (errno == EINTR) continue;
+                                abort("KERNEL ERROR: IO_URING EVENTFD - %s\n", strerror(errno));
+                        }
+                        break;
+                }
+
+                int ret;
+                for() {
+                        ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &efd, 1);
+                        if (ret < 0) {
+                                if (errno == EINTR) continue;
+                                abort("KERNEL ERROR: IO_URING EVENTFD REGISTER - %s\n", strerror(errno));
+                        }
+                        break;
+                }
+
                 // 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  );
@@ -423 +463 @@
                 this.ring_flags = params.flags;
                 this.fd         = fd;
+                this.efd        = efd;
                 this.eager_submits  = params_in.eager_submits;
                 this.poller_submits = params_in.poller_submits;
@@ -445 +486 @@
                 // close the file descriptor
                 close(this.fd);
+                close(this.efd);
 
                 free( this.submit_q.ready ); // Maybe null, doesn't matter
@@ -452 +494 @@
 // I/O Context Sleep
 //=============================================================================================
-
-        void __ioctx_register($io_ctx_thread & ctx, struct epoll_event & ev) {
-                ev.events = EPOLLIN | EPOLLONESHOT;
+        #define IOEVENTS EPOLLIN | EPOLLONESHOT
+
+        static inline void __ioctx_epoll_ctl($io_ctx_thread & ctx, int op, const char * error) {
+                struct epoll_event ev;
+                ev.events = IOEVENTS;
                 ev.data.u64 = (__u64)&ctx;
-                int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_ADD, ctx.ring->fd, &ev);
+                int ret = epoll_ctl(iopoll.epollfd, op, ctx.ring->efd, &ev);
                 if (ret < 0) {
-                        abort( "KERNEL ERROR: EPOLL ADD - (%d) %s\n", (int)errno, strerror(errno) );
-                }
-        }
-
-        void __ioctx_prepare_block($io_ctx_thread & ctx, struct epoll_event & ev) {
-                int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_MOD, ctx.ring->fd, &ev);
-                if (ret < 0) {
-                        abort( "KERNEL ERROR: EPOLL REARM - (%d) %s\n", (int)errno, strerror(errno) );
-                }
+                        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");
         }