Changeset 365e423

Timestamp:

Jul 10, 2020, 2:22:17 PM (4 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

7922158, e1801fc

Parents:

0a92c78 (diff), 59f74a2 (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

Files:

• benchmark/io/readv.cfa (modified) (9 diffs)
• libcfa/src/concurrency/io.cfa (modified) (13 diffs)
• libcfa/src/concurrency/iocall.cfa (modified) (1 diff)
• src/Makefile.in (modified) (1 diff)
• src/Parser/module.mk (modified) (1 diff)

benchmark/io/readv.cfa

@@ +1 @@
+#define _GNU_SOURCE
+
 #include <stdlib.h>
 #include <stdio.h>
@@ -22 +24 @@
 extern bool traceHeapOn();
 extern ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
+extern ssize_t cfa_preadv2_fixed(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
+extern void register_fixed_files( cluster &, int *, unsigned count );
 
 int fd;
@@ -28 +32 @@
 
 unsigned long int buflen = 50;
+bool fixed_file = false;
 
 thread __attribute__((aligned(128))) Reader {};
 void ?{}( Reader & this ) {
         ((thread&)this){ "Reader Thread", *the_benchmark_cluster };
+}
+
+int do_read(int fd, struct iovec * iov) {
+        if(fixed_file) {
+                return cfa_preadv2_fixed(fd, iov, 1, 0, 0);
+        }
+        else {
+                return cfa_preadv2(fd, iov, 1, 0, 0);
+        }
 }
 
@@ -42 +56 @@
 
         while(__atomic_load_n(&run, __ATOMIC_RELAXED)) {
-                int r = cfa_preadv2(fd, &iov, 1, 0, 0);
+                int r = do_read(fd, &iov);
                 if(r < 0) abort("%s\n", strerror(-r));
 
@@ -52 +66 @@
         BENCH_DECL
         unsigned flags = 0;
+        int file_flags = 0;
         unsigned sublen = 16;
 
@@ -96 +111 @@
                         case 'k':
                                 flags |= CFA_CLUSTER_IO_KERNEL_POLL_SUBMITS;
+                                fixed_file = true;
                                 break;
                         case 'i':
                                 flags |= CFA_CLUSTER_IO_KERNEL_POLL_COMPLETES;
+                                file_flags |= O_DIRECT;
                                 break;
                         case 'l':
@@ -123 +140 @@
         }
 
-        fd = open(__FILE__, 0);
-        if(fd < 0) {
+        int lfd = open(__FILE__, file_flags);
+        if(lfd < 0) {
                 fprintf(stderr, "Could not open source file\n");
                 exit(EXIT_FAILURE);
@@ -134 +151 @@
                 Time start, end;
                 BenchCluster cl = { flags, CFA_STATS_READY_Q | CFA_STATS_IO };
+
+                if(fixed_file) {
+                        fd = 0;
+                        register_fixed_files( cl.self, &lfd, 1 );
+                }
+                else {
+                        fd = lfd;
+                }
+
                 {
                         BenchProc procs[nprocs];
@@ -161 +187 @@
         }
 
-        close(fd);
+        close(lfd);
 }

libcfa/src/concurrency/io.cfa

@@ -14 +14 @@
 //
 
-// #define __CFA_DEBUG_PRINT_IO__
-// #define __CFA_DEBUG_PRINT_IO_CORE__
+#if defined(__CFA_DEBUG__)
+        // #define __CFA_DEBUG_PRINT_IO__
+        #define __CFA_DEBUG_PRINT_IO_CORE__
+#endif
 
 #include "kernel.hfa"
@@ -325 +327 @@
 
                 // Create the poller thread
-                __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
+                __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 );
@@ -430 +432 @@
         }
 
+        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:
+                if( to_submit > 0 ) {
+                        if( !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
+                                need_sys_to_submit = true;
+                                break TO_SUBMIT;
+                        }
+                        if( (*ring.submit_q.flags) & IORING_SQ_NEED_WAKEUP ) {
+                                need_sys_to_submit = true;
+                                flags |= IORING_ENTER_SQ_WAKEUP;
+                        }
+                }
+
+                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;
+                        }
+                }
+
+                int ret = 0;
+                if( need_sys_to_submit || need_sys_to_complete ) {
+                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, min_complete, flags, mask, _NSIG / 8);
+                        if( ret < 0 ) {
+                                switch((int)errno) {
+                                case EAGAIN:
+                                case EINTR:
+                                        ret = -1;
+                                        break;
+                                default:
+                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
+                                }
+                        }
+                }
+
+                // Memory barrier
+                __atomic_thread_fence( __ATOMIC_SEQ_CST );
+                return ret;
+        }
+
 //=============================================================================================
 // I/O Polling
@@ -438 +492 @@
         // Process a single completion message from the io_uring
         // This is NOT thread-safe
-        static [int, bool] __drain_io( & struct __io_data ring, * sigset_t mask, int waitcnt, bool in_kernel ) {
+        static [int, bool] __drain_io( & struct __io_data ring, * sigset_t mask ) {
                 /* paranoid */ verify( !kernelTLS.preemption_state.enabled );
 
@@ -447 +501 @@
                 }
 
-                if (to_submit > 0 || waitcnt > 0) {
-                        int ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8);
-                        if( ret < 0 ) {
-                                switch((int)errno) {
-                                case EAGAIN:
-                                case EINTR:
-                                        return [0, true];
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
-                                }
-                        }
-
-                        // Release the consumed SQEs
-                        __release_consumed_submission( ring );
-
-                        // update statistics
+                int ret = __io_uring_enter(ring, to_submit, true, mask);
+                if( ret < 0 ) {
+                        return [0, true];
+                }
+
+                // update statistics
+                if (to_submit > 0) {
                         __STATS__( true,
                                 if( to_submit > 0 ) {
@@ -472 +517 @@
                 }
 
-                // Memory barrier
-                __atomic_thread_fence( __ATOMIC_SEQ_CST );
+                // Release the consumed SQEs
+                __release_consumed_submission( ring );
 
                 // Drain the queue
@@ -497 +542 @@
 
                         data->result = cqe.res;
-                        if(!in_kernel) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
-                        else         { __unpark( &ring.poller.slow.id, data->thrd __cfaabi_dbg_ctx2 ); }
+                        if(!mask) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
+                        else      { __unpark( &ring.poller.slow.id, data->thrd __cfaabi_dbg_ctx2 ); }
                 }
 
@@ -546 +591 @@
                                 int count;
                                 bool again;
-                                [count, again] = __drain_io( ring, &mask, 1, true );
+                                [count, again] = __drain_io( ring, &mask );
 
                                 __atomic_store_n( &ring.poller.slow.blocked, false, __ATOMIC_SEQ_CST );
@@ -568 +613 @@
                                 int count;
                                 bool again;
-                                [count, again] = __drain_io( ring, &mask, 1, true );
+                                [count, again] = __drain_io( ring, &mask );
 
                                 // Update statistics
@@ -606 +651 @@
                         bool again;
                         disable_interrupts();
-                                [count, again] = __drain_io( *this.ring, 0p, 0, false );
+                                [count, again] = __drain_io( *this.ring, 0p );
 
                                 if(!again) reset++;
@@ -800 +845 @@
                         // We got the lock
                         unsigned to_submit = __collect_submitions( ring );
-                        int ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, 0, 0p, _NSIG / 8);
+                        int ret = __io_uring_enter( ring, to_submit, false, 0p );
                         if( ret < 0 ) {
-                                switch((int)errno) {
-                                case EAGAIN:
-                                case EINTR:
-                                        unlock(ring.submit_q.lock);
-                                        return;
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
-                                }
-                        }
-
-                        /* paranoid */ verify( ret > 0 );
+                                unlock(ring.submit_q.lock);
+                                return;
+                        }
+
+                        /* paranoid */ verify( ret > 0 || (ring.ring_flags & IORING_SETUP_SQPOLL) );
 
                         // Release the consumed SQEs
@@ -830 +869 @@
                         lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
 
+                        /* paranoid */ verifyf( ring.submit_q.sqes[ idx ].user_data != 0,
+                        /* paranoid */  "index %u already reclaimed\n"
+                        /* paranoid */  "head %u, prev %u, tail %u\n"
+                        /* paranoid */  "[-0: %u,-1: %u,-2: %u,-3: %u]\n",
+                        /* paranoid */  idx,
+                        /* paranoid */  *ring.submit_q.head, ring.submit_q.prev_head, *tail
+                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 0) & (*ring.submit_q.mask) ]
+                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 1) & (*ring.submit_q.mask) ]
+                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 2) & (*ring.submit_q.mask) ]
+                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 3) & (*ring.submit_q.mask) ]
+                        /* paranoid */ );
+
                         // Append to the list of ready entries
 
                         /* paranoid */ verify( idx <= mask );
-
-                        ring.submit_q.array[ (*tail) & mask ] = idx & mask;
+                        ring.submit_q.array[ (*tail) & mask ] = idx;
                         __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
 
-                        /* paranoid */ verify( ring.submit_q.sqes[ idx ].user_data != 0 );
-
                         // Submit however, many entries need to be submitted
-                        int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
+                        int ret = __io_uring_enter( ring, 1, false, 0p );
                         if( ret < 0 ) {
                                 switch((int)errno) {
@@ -906 +954 @@
                 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

libcfa/src/concurrency/iocall.cfa

@@ -131 +131 @@
                 #endif
         }
+
+        ssize_t cfa_preadv2_fixed(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
+                #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READV)
+                        return preadv2(fd, iov, iovcnt, offset, flags);
+                #else
+                        __submit_prelude
+
+                        (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
+                        sqe->flags |= IOSQE_FIXED_FILE;
+
+                        __submit_wait
+                #endif
+        }
 #endif
 

src/Makefile.in

@@ -626 +626 @@
 
 BUILT_SOURCES = Parser/parser.hh
-AM_YFLAGS = -d -t -v
+AM_YFLAGS = -d -t -v -Wno-yacc
 SRC_RESOLVEXPR = \
       ResolvExpr/AdjustExprType.cc \

src/Parser/module.mk

@@ -17 +17 @@
 BUILT_SOURCES = Parser/parser.hh
 
-AM_YFLAGS = -d -t -v
+AM_YFLAGS = -d -t -v -Wno-yacc
 
 SRC += \