Changeset d384787

Timestamp:

Apr 24, 2020, 1:54:29 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:

0ea6c5a

Parents:

ecf6b46

Message:

Added basic statistics to io.cfa

Location:

libcfa/src/concurrency

Files:

• io.cfa (modified) (10 diffs)
• iofwd.hfa (added)
• kernel.cfa (modified) (1 diff)
• kernel.hfa (modified) (3 diffs)

libcfa/src/concurrency/io.cfa

@@ -101 +101 @@
 
                 // Requires features
-                // // 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);
-                // }
+                #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
@@ -112 +114 @@
                 }
 
-                // mmap the Completion Queue into existence (may or may not be needed)
                 // Requires features
-                // if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
-                //      cq->ring_ptr = sq->ring_ptr;
-                // }
-                // else {
+                #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);
@@ -124 +129 @@
                                 abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
                         }
-                // }
+                }
 
                 // mmap the submit queue entries
@@ -171 +176 @@
                 (this.io.submit){ min(*sq.num, *cq.num) };
 
+                // Initialize statistics
+                this.io.submit_q.stats.submit_avg.val = 0;
+                this.io.submit_q.stats.submit_avg.cnt = 0;
+                this.io.completion_q.stats.completed_avg.val = 0;
+                this.io.completion_q.stats.completed_avg.cnt = 0;
+
                 // Create the poller thread
                 this.io.stack = __create_pthread( &this.io.poller, __io_poller, &this );
@@ -185 +196 @@
                 pthread_join( this.io.poller, 0p );
                 free( this.io.stack );
+
+                // print statistics
+                __cfaabi_bits_print_safe( STDERR_FILENO,
+                        "----- I/O uRing Stats -----\n"
+                        "- total submit calls  : %llu\n"
+                        "- avg submit          : %lf\n"
+                        "- total wait calls    : %llu\n"
+                        "- avg completion/wait : %lf\n",
+                        this.io.submit_q.stats.submit_avg.cnt,
+                        ((double)this.io.submit_q.stats.submit_avg.val) / this.io.submit_q.stats.submit_avg.cnt,
+                        this.io.completion_q.stats.completed_avg.cnt,
+                        ((double)this.io.completion_q.stats.completed_avg.val) / this.io.completion_q.stats.completed_avg.cnt
+                );
 
                 // Shutdown the io rings
@@ -215 +239 @@
         // Process a single completion message from the io_uring
         // This is NOT thread-safe
-        static bool __io_process(struct io_ring & ring) {
+        int __drain_io( struct io_ring & ring, sigset_t & mask, int waitcnt ) {
+                int ret = syscall( __NR_io_uring_enter, ring.fd, 0, waitcnt, IORING_ENTER_GETEVENTS, &mask, _NSIG / 8);
+                if( ret < 0 ) {
+                        switch((int)errno) {
+                        case EAGAIN:
+                        case EINTR:
+                                return -EAGAIN;
+                        default:
+                                abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
+                        }
+                }
+
+                // Drain the queue
                 unsigned head = *ring.completion_q.head;
                 unsigned tail = __atomic_load_n(ring.completion_q.tail, __ATOMIC_ACQUIRE);
 
-                if (head == tail) return false;
-
-                unsigned idx = head & (*ring.completion_q.mask);
-                struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
-
-                /* paranoid */ verify(&cqe);
-
-                struct io_user_data * data = (struct io_user_data *)cqe.user_data;
-                // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
-
-                data->result = cqe.res;
-                __unpark( data->thrd __cfaabi_dbg_ctx2 );
+                // Nothing was new return 0
+                if (head == tail) {
+                        ring.completion_q.stats.completed_avg.cnt += 1;
+                        return 0;
+                }
+
+                uint32_t count = tail - head;
+                for(i; count) {
+                        unsigned idx = (head + i) & (*ring.completion_q.mask);
+                        struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
+
+                        /* paranoid */ verify(&cqe);
+
+                        struct io_user_data * data = (struct io_user_data *)cqe.user_data;
+                        // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
+
+                        data->result = cqe.res;
+                        __unpark( data->thrd __cfaabi_dbg_ctx2 );
+                }
 
                 // Allow new submissions to happen
-                V(ring.submit);
+                V(ring.submit, count);
 
                 // Mark to the kernel that the cqe has been seen
                 // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
-                __atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED );
-
-                return true;
+                __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED );
+
+                ring.completion_q.stats.completed_avg.val += count;
+                ring.completion_q.stats.completed_avg.cnt += 1;
+
+                return count;
         }
 
@@ -257 +303 @@
                 verify( (*ring.completion_q.head) == (*ring.completion_q.tail) );
 
-                LOOP: while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
-                        int ret = syscall( __NR_io_uring_enter, ring.fd, 0, 1, IORING_ENTER_GETEVENTS, &mask, _NSIG / 8);
-                        if( ret < 0 ) {
-                                switch((int)errno) {
-                                case EAGAIN:
-                                case EINTR:
-                                        continue LOOP;
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
-                                }
-                        }
-
-                        // Drain the queue
-                        while(__io_process(ring)) {}
+                while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
+                        __drain_io( ring, mask, 1 );
                 }
 
@@ -343 +377 @@
                 }
 
+                ring.submit_q.stats.submit_avg.val += 1;
+                ring.submit_q.stats.submit_avg.cnt += 1;
+
                 unlock(ring.submit_q.lock);
                 // Make sure that idx was submitted
@@ -512 +549 @@
 int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
         #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_ACCEPT)
-                __SOCKADDR_ARG _addr;
-                _addr.__sockaddr__ = addr;
-                return accept4( sockfd, _addr, addrlen, flags );
+                #pragma GCC diagnostic push
+                #pragma GCC diagnostic ignored "-Wattributes"
+                        __SOCKADDR_ARG _addr;
+                        _addr.__sockaddr__ = addr;
+                        return accept4( sockfd, _addr, addrlen, flags );
+                #pragma GCC diagnostic pop
         #else
                 __submit_prelude
@@ -529 +569 @@
 int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
         #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CONNECT)
-                __CONST_SOCKADDR_ARG _addr;
-                _addr.__sockaddr__ = addr;
-                return connect( sockfd, _addr, addrlen );
+                #pragma GCC diagnostic push
+                #pragma GCC diagnostic ignored "-Wattributes"
+                        __CONST_SOCKADDR_ARG _addr;
+                        _addr.__sockaddr__ = addr;
+                        return connect( sockfd, _addr, addrlen );
+                #pragma GCC diagnostic pop
         #else
                 __submit_prelude

libcfa/src/concurrency/kernel.cfa

@@ -1004 +1004 @@
 }
 
+bool V(semaphore & this, unsigned diff) with( this ) {
+        $thread * thrd = 0p;
+        lock( lock __cfaabi_dbg_ctx2 );
+        int release = max(-count, (int)diff);
+        count += diff;
+        for(release) {
+                unpark( pop_head( waiting ) __cfaabi_dbg_ctx2 );
+        }
+
+        unlock( lock );
+
+        return thrd != 0p;
+}
+
 //-----------------------------------------------------------------------------
 // Global Queues

libcfa/src/concurrency/kernel.hfa

@@ -40 +40 @@
 void   P (semaphore & this);
 bool   V (semaphore & this);
+bool   V (semaphore & this, unsigned count);
 
 
@@ -144 +145 @@
         void * ring_ptr;
         size_t ring_sz;
+
+        // Statistics
+        struct {
+                struct {
+                        unsigned long long int val;
+                        unsigned long long int cnt;
+                } submit_avg;
+        } stats;
 };
 
@@ -164 +173 @@
         void * ring_ptr;
         size_t ring_sz;
+
+        // Statistics
+        struct {
+                struct {
+                        unsigned long long int val;
+                        unsigned long long int cnt;
+                } completed_avg;
+        } stats;
 };