Changeset 7affcda for libcfa/src/concurrency

Timestamp:

Jun 3, 2022, 6:53:57 PM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, ast-experimental, master, pthread-emulation, qualifiedEnum

Children:

153570d

Parents:

90a8125

Message:

Fixed a bug where io wouldn't drain everything if the completion queue is filled

File:

• libcfa/src/concurrency/io.cfa (modified) (1 diff)

libcfa/src/concurrency/io.cfa

@@ -159 +159 @@
 
                 const __u32 mask = *ctx->cq.mask;
+                const __u32 num  = *ctx->cq.num;
                 unsigned long long ts_prev = ctx->cq.ts;
-
-                // re-read the head and tail in case it already changed.
-                const __u32 head = *ctx->cq.head;
-                const __u32 tail = *ctx->cq.tail;
-                const __u32 count = tail - head;
-                __STATS__( false, io.calls.drain++; io.calls.completed += count; )
-
-                for(i; count) {
-                        unsigned idx = (head + i) & mask;
-                        volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
-
-                        /* paranoid */ verify(&cqe);
-
-                        struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
-                        // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
-
-                        __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL );
-                }
-
-                unsigned long long ts_next = ctx->cq.ts = rdtscl();
-
-                // 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_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
-                ctx->proc->idle_wctx.drain_time = ts_next;
+                unsigned long long ts_next;
+
+                // We might need to do this multiple times if more events completed than can fit in the queue.
+                for() {
+                        // re-read the head and tail in case it already changed.
+                        const __u32 head = *ctx->cq.head;
+                        const __u32 tail = *ctx->cq.tail;
+                        const __u32 count = tail - head;
+                        __STATS__( false, io.calls.drain++; io.calls.completed += count; )
+
+                        for(i; count) {
+                                unsigned idx = (head + i) & mask;
+                                volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
+
+                                /* paranoid */ verify(&cqe);
+
+                                struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
+                                // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
+
+                                __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL );
+                        }
+
+                        ts_next = ctx->cq.ts = rdtscl();
+
+                        // 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_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
+                        ctx->proc->idle_wctx.drain_time = ts_next;
+
+                        if(likely(count < num)) break;
+
+                        ioring_syscsll( *ctx, 0, IORING_ENTER_GETEVENTS);
+                }
 
                 __cfadbg_print_safe(io, "Kernel I/O : %u completed age %llu\n", count, ts_next);