Changeset d611995

Timestamp:

Jan 18, 2021, 12:42:41 PM (4 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:

78a8440

Parents:

6f94958

Message:

Fix crash on cluster exit.
(Added epoch based blocking to iopoll.)

Location:

libcfa/src/concurrency

Files:

• io.cfa (modified) (1 diff)
• io/setup.cfa (modified) (5 diffs)
• io/types.hfa (modified) (1 diff)

libcfa/src/concurrency/io.cfa

@@ -318 +318 @@
 
                 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller %d (%p) stopping\n", this.ring->fd, &this);
+
+                __ioctx_unregister( this );
         }
 

libcfa/src/concurrency/io/setup.cfa

@@ -113 +113 @@
 
         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
+                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 size_t epoch;   // Epoch used for memory reclamation
         } iopoll;
 
@@ -131 +132 @@
                 iopoll.run = true;
                 iopoll.stack = __create_pthread( &iopoll.thrd, iopoll_loop, 0p );
+                iopoll.epoch = 0;
         }
 
@@ -174 +176 @@
                 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
@@ -496 +501 @@
 // I/O Context Sleep
 //=============================================================================================
-        #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.events = EPOLLIN | EPOLLONESHOT;
                 ev.data.u64 = (__u64)&ctx;
                 int ret = epoll_ctl(iopoll.epollfd, op, ctx.ring->efd, &ev);
@@ -517 +520 @@
         }
 
+        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 == __atomic_load_n(&iopoll.epoch, __ATOMIC_RELAXED)) yield();
+        }
+
 //=============================================================================================
 // I/O Context Misc Setup

libcfa/src/concurrency/io/types.hfa

@@ -133 +133 @@
         struct $io_ctx_thread;
         void __ioctx_register($io_ctx_thread & ctx);
+        void __ioctx_unregister($io_ctx_thread & ctx);
         void __ioctx_prepare_block($io_ctx_thread & ctx);
         void __sqe_clean( volatile struct io_uring_sqe * sqe );