source: libcfa/src/concurrency/io.cfa @ 426f60c

arm-ehenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 426f60c was 426f60c, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Web server seems to work

  • Property mode set to 100644
File size: 21.2 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2020 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// io.cfa --
8//
9// Author           : Thierry Delisle
10// Created On       : Thu Apr 23 17:31:00 2020
11// Last Modified By :
12// Last Modified On :
13// Update Count     :
14//
15
16#define __cforall_thread__
17
18#if defined(__CFA_DEBUG__)
19        // #define __CFA_DEBUG_PRINT_IO__
20        // #define __CFA_DEBUG_PRINT_IO_CORE__
21#endif
22
23
24#if defined(CFA_HAVE_LINUX_IO_URING_H)
25        #define _GNU_SOURCE         /* See feature_test_macros(7) */
26        #include <errno.h>
27        #include <signal.h>
28        #include <stdint.h>
29        #include <string.h>
30        #include <unistd.h>
31
32        extern "C" {
33                #include <sys/epoll.h>
34                #include <sys/eventfd.h>
35                #include <sys/syscall.h>
36
37                #include <linux/io_uring.h>
38        }
39
40        #include "stats.hfa"
41        #include "kernel.hfa"
42        #include "kernel/fwd.hfa"
43        #include "io/types.hfa"
44
45        static const char * opcodes[] = {
46                "OP_NOP",
47                "OP_READV",
48                "OP_WRITEV",
49                "OP_FSYNC",
50                "OP_READ_FIXED",
51                "OP_WRITE_FIXED",
52                "OP_POLL_ADD",
53                "OP_POLL_REMOVE",
54                "OP_SYNC_FILE_RANGE",
55                "OP_SENDMSG",
56                "OP_RECVMSG",
57                "OP_TIMEOUT",
58                "OP_TIMEOUT_REMOVE",
59                "OP_ACCEPT",
60                "OP_ASYNC_CANCEL",
61                "OP_LINK_TIMEOUT",
62                "OP_CONNECT",
63                "OP_FALLOCATE",
64                "OP_OPENAT",
65                "OP_CLOSE",
66                "OP_FILES_UPDATE",
67                "OP_STATX",
68                "OP_READ",
69                "OP_WRITE",
70                "OP_FADVISE",
71                "OP_MADVISE",
72                "OP_SEND",
73                "OP_RECV",
74                "OP_OPENAT2",
75                "OP_EPOLL_CTL",
76                "OP_SPLICE",
77                "OP_PROVIDE_BUFFERS",
78                "OP_REMOVE_BUFFERS",
79                "OP_TEE",
80                "INVALID_OP"
81        };
82
83        // returns true of acquired as leader or second leader
84        static inline bool try_lock( __leaderlock_t & this ) {
85                const uintptr_t thrd = 1z | (uintptr_t)active_thread();
86                bool block;
87                disable_interrupts();
88                for() {
89                        struct $thread * expected = this.value;
90                        if( 1p != expected && 0p != expected ) {
91                                /* paranoid */ verify( thrd != (uintptr_t)expected ); // We better not already be the next leader
92                                enable_interrupts( __cfaabi_dbg_ctx );
93                                return false;
94                        }
95                        struct $thread * desired;
96                        if( 0p == expected ) {
97                                // If the lock isn't locked acquire it, no need to block
98                                desired = 1p;
99                                block = false;
100                        }
101                        else {
102                                // If the lock is already locked try becomming the next leader
103                                desired = (struct $thread *)thrd;
104                                block = true;
105                        }
106                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
107                }
108                if( block ) {
109                        enable_interrupts( __cfaabi_dbg_ctx );
110                        park();
111                        disable_interrupts();
112                }
113                return true;
114        }
115
116        static inline bool next( __leaderlock_t & this ) {
117                /* paranoid */ verify( ! __preemption_enabled() );
118                struct $thread * nextt;
119                for() {
120                        struct $thread * expected = this.value;
121                        /* paranoid */ verify( (1 & (uintptr_t)expected) == 1 ); // The lock better be locked
122
123                        struct $thread * desired;
124                        if( 1p == expected ) {
125                                // No next leader, just unlock
126                                desired = 0p;
127                                nextt   = 0p;
128                        }
129                        else {
130                                // There is a next leader, remove but keep locked
131                                desired = 1p;
132                                nextt   = (struct $thread *)(~1z & (uintptr_t)expected);
133                        }
134                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
135                }
136
137                if(nextt) {
138                        unpark( nextt );
139                        enable_interrupts( __cfaabi_dbg_ctx );
140                        return true;
141                }
142                enable_interrupts( __cfaabi_dbg_ctx );
143                return false;
144        }
145
146//=============================================================================================
147// I/O Syscall
148//=============================================================================================
149        static int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get ) {
150                bool need_sys_to_submit = false;
151                bool need_sys_to_complete = false;
152                unsigned flags = 0;
153
154                TO_SUBMIT:
155                if( to_submit > 0 ) {
156                        if( !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
157                                need_sys_to_submit = true;
158                                break TO_SUBMIT;
159                        }
160                        if( (*ring.submit_q.flags) & IORING_SQ_NEED_WAKEUP ) {
161                                need_sys_to_submit = true;
162                                flags |= IORING_ENTER_SQ_WAKEUP;
163                        }
164                }
165
166                if( get && !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
167                        flags |= IORING_ENTER_GETEVENTS;
168                        if( (ring.ring_flags & IORING_SETUP_IOPOLL) ) {
169                                need_sys_to_complete = true;
170                        }
171                }
172
173                int ret = 0;
174                if( need_sys_to_submit || need_sys_to_complete ) {
175                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING enter %d %u %u\n", ring.fd, to_submit, flags);
176                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, (sigset_t *)0p, _NSIG / 8);
177                        if( ret < 0 ) {
178                                switch((int)errno) {
179                                case EAGAIN:
180                                case EINTR:
181                                        ret = -1;
182                                        break;
183                                default:
184                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
185                                }
186                        }
187                }
188
189                // Memory barrier
190                __atomic_thread_fence( __ATOMIC_SEQ_CST );
191                return ret;
192        }
193
194//=============================================================================================
195// I/O Polling
196//=============================================================================================
197        static unsigned __collect_submitions( struct __io_data & ring );
198        static __u32 __release_consumed_submission( struct __io_data & ring );
199
200        // Process a single completion message from the io_uring
201        // This is NOT thread-safe
202        static inline void process( volatile struct io_uring_cqe & cqe ) {
203                struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
204                __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
205
206                fulfil( *future, cqe.res );
207        }
208
209        static [int, bool] __drain_io( & struct __io_data ring ) {
210                /* paranoid */ verify( ! __preemption_enabled() );
211
212                unsigned to_submit = 0;
213                if( ring.poller_submits ) {
214                        // If the poller thread also submits, then we need to aggregate the submissions which are ready
215                        to_submit = __collect_submitions( ring );
216                }
217
218                int ret = __io_uring_enter(ring, to_submit, true);
219                if( ret < 0 ) {
220                        return [0, true];
221                }
222
223                // update statistics
224                if (to_submit > 0) {
225                        __STATS__( true,
226                                if( to_submit > 0 ) {
227                                        io.submit_q.submit_avg.rdy += to_submit;
228                                        io.submit_q.submit_avg.csm += ret;
229                                        io.submit_q.submit_avg.cnt += 1;
230                                }
231                        )
232                }
233
234                __atomic_thread_fence( __ATOMIC_SEQ_CST );
235
236                // Release the consumed SQEs
237                __release_consumed_submission( ring );
238
239                // Drain the queue
240                unsigned head = *ring.completion_q.head;
241                unsigned tail = *ring.completion_q.tail;
242                const __u32 mask = *ring.completion_q.mask;
243
244                // Nothing was new return 0
245                if (head == tail) {
246                        return [0, to_submit > 0];
247                }
248
249                __u32 count = tail - head;
250                /* paranoid */ verify( count != 0 );
251                for(i; count) {
252                        unsigned idx = (head + i) & mask;
253                        volatile struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
254
255                        /* paranoid */ verify(&cqe);
256
257                        process( cqe );
258                }
259
260                // Mark to the kernel that the cqe has been seen
261                // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
262                __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_SEQ_CST );
263
264                return [count, count > 0 || to_submit > 0];
265        }
266
267        void main( $io_ctx_thread & this ) {
268                epoll_event ev;
269                __ioctx_register( this, ev );
270
271                __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %d (%p) ready\n", this.ring->fd, &this);
272
273                const int reset_cnt = 5;
274                int reset = reset_cnt;
275                // Then loop until we need to start
276                LOOP:
277                while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
278                        // Drain the io
279                        int count;
280                        bool again;
281                        disable_interrupts();
282                                [count, again] = __drain_io( *this.ring );
283
284                                if(!again) reset--;
285
286                                // Update statistics
287                                __STATS__( true,
288                                        io.complete_q.completed_avg.val += count;
289                                        io.complete_q.completed_avg.cnt += 1;
290                                )
291                        enable_interrupts( __cfaabi_dbg_ctx );
292
293                        // If we got something, just yield and check again
294                        if(reset > 1) {
295                                yield();
296                                continue LOOP;
297                        }
298
299                        // We alread failed to find completed entries a few time.
300                        if(reset == 1) {
301                                // Rearm the context so it can block
302                                // but don't block right away
303                                // we need to retry one last time in case
304                                // something completed *just now*
305                                __ioctx_prepare_block( this, ev );
306                                continue LOOP;
307                        }
308
309                                __STATS__( false,
310                                        io.complete_q.blocks += 1;
311                                )
312                                __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %d (%p)\n", this.ring->fd, &this);
313
314                                // block this thread
315                                wait( this.sem );
316
317                                eventfd_t v;
318                                eventfd_read(this.ring->efd, &v);
319
320                        // restore counter
321                        reset = reset_cnt;
322                }
323
324                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller %d (%p) stopping\n", this.ring->fd, &this);
325        }
326
327//=============================================================================================
328// I/O Submissions
329//=============================================================================================
330
331// Submition steps :
332// 1 - Allocate a queue entry. The ring already has memory for all entries but only the ones
333//     listed in sq.array are visible by the kernel. For those not listed, the kernel does not
334//     offer any assurance that an entry is not being filled by multiple flags. Therefore, we
335//     need to write an allocator that allows allocating concurrently.
336//
337// 2 - Actually fill the submit entry, this is the only simple and straightforward step.
338//
339// 3 - Append the entry index to the array and adjust the tail accordingly. This operation
340//     needs to arrive to two concensus at the same time:
341//     A - The order in which entries are listed in the array: no two threads must pick the
342//         same index for their entries
343//     B - When can the tail be update for the kernel. EVERY entries in the array between
344//         head and tail must be fully filled and shouldn't ever be touched again.
345//
346
347        // Allocate an submit queue entry.
348        // The kernel cannot see these entries until they are submitted, but other threads must be
349        // able to see which entries can be used and which are already un used by an other thread
350        // for convenience, return both the index and the pointer to the sqe
351        // sqe == &sqes[idx]
352        [* volatile struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data ) {
353                /* paranoid */ verify( data != 0 );
354
355                // Prepare the data we need
356                __attribute((unused)) int len   = 0;
357                __attribute((unused)) int block = 0;
358                __u32 cnt = *ring.submit_q.num;
359                __u32 mask = *ring.submit_q.mask;
360
361                __u32 off = thread_rand();
362
363                // Loop around looking for an available spot
364                for() {
365                        // Look through the list starting at some offset
366                        for(i; cnt) {
367                                __u64 expected = 3;
368                                __u32 idx = (i + off) & mask; // Get an index from a random
369                                volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
370                                volatile __u64 * udata = &sqe->user_data;
371
372                                // Allocate the entry by CASing the user_data field from 0 to the future address
373                                if( *udata == expected &&
374                                        __atomic_compare_exchange_n( udata, &expected, data, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) )
375                                {
376                                        // update statistics
377                                        __STATS__( false,
378                                                io.submit_q.alloc_avg.val   += len;
379                                                io.submit_q.alloc_avg.block += block;
380                                                io.submit_q.alloc_avg.cnt   += 1;
381                                        )
382
383                                        // debug log
384                                        __cfadbg_print_safe( io, "Kernel I/O : allocated [%p, %u] for %p (%p)\n", sqe, idx, active_thread(), (void*)data );
385
386                                        // Success return the data
387                                        sqe->opcode = 0;
388                                        sqe->flags = 0;
389                                        sqe->ioprio = 0;
390                                        sqe->fd = 0;
391                                        sqe->off = 0;
392                                        sqe->addr = 0;
393                                        sqe->len = 0;
394                                        sqe->accept_flags = 0;
395                                        sqe->__pad2[0] = 0;
396                                        sqe->__pad2[1] = 0;
397                                        sqe->__pad2[2] = 0;
398                                        return [sqe, idx];
399                                }
400                                verify(expected != data);
401
402                                // This one was used
403                                len ++;
404                        }
405
406                        block++;
407
408                        abort( "Kernel I/O : all submit queue entries used, yielding\n" );
409
410                        yield();
411                }
412        }
413
414        static inline __u32 __submit_to_ready_array( struct __io_data & ring, __u32 idx, const __u32 mask ) {
415                /* paranoid */ verify( idx <= mask   );
416                /* paranoid */ verify( idx != -1ul32 );
417
418                // We need to find a spot in the ready array
419                __attribute((unused)) int len   = 0;
420                __attribute((unused)) int block = 0;
421                __u32 ready_mask = ring.submit_q.ready_cnt - 1;
422
423                __u32 off = thread_rand();
424
425                __u32 picked;
426                LOOKING: for() {
427                        for(i; ring.submit_q.ready_cnt) {
428                                picked = (i + off) & ready_mask;
429                                __u32 expected = -1ul32;
430                                if( __atomic_compare_exchange_n( &ring.submit_q.ready[picked], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
431                                        break LOOKING;
432                                }
433                                verify(expected != idx);
434
435                                len ++;
436                        }
437
438                        block++;
439
440                        __u32 released = __release_consumed_submission( ring );
441                        if( released == 0 ) {
442                                yield();
443                        }
444                }
445
446                // update statistics
447                __STATS__( false,
448                        io.submit_q.look_avg.val   += len;
449                        io.submit_q.look_avg.block += block;
450                        io.submit_q.look_avg.cnt   += 1;
451                )
452
453                return picked;
454        }
455
456        void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
457                __io_data & ring = *ctx->thrd.ring;
458
459                {
460                        __attribute__((unused)) volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
461                        __cfadbg_print_safe( io,
462                                "Kernel I/O : submitting %u (%p) for %p\n"
463                                "    data: %p\n"
464                                "    opcode: %s\n"
465                                "    fd: %d\n"
466                                "    flags: %d\n"
467                                "    prio: %d\n"
468                                "    off: %p\n"
469                                "    addr: %p\n"
470                                "    len: %d\n"
471                                "    other flags: %d\n"
472                                "    splice fd: %d\n"
473                                "    pad[0]: %llu\n"
474                                "    pad[1]: %llu\n"
475                                "    pad[2]: %llu\n",
476                                idx, sqe,
477                                active_thread(),
478                                (void*)sqe->user_data,
479                                opcodes[sqe->opcode],
480                                sqe->fd,
481                                sqe->flags,
482                                sqe->ioprio,
483                                sqe->off,
484                                sqe->addr,
485                                sqe->len,
486                                sqe->accept_flags,
487                                sqe->splice_fd_in,
488                                sqe->__pad2[0],
489                                sqe->__pad2[1],
490                                sqe->__pad2[2]
491                        );
492                }
493
494
495                // Get now the data we definetely need
496                volatile __u32 * const tail = ring.submit_q.tail;
497                const __u32 mask  = *ring.submit_q.mask;
498
499                // There are 2 submission schemes, check which one we are using
500                if( ring.poller_submits ) {
501                        // If the poller thread submits, then we just need to add this to the ready array
502                        __submit_to_ready_array( ring, idx, mask );
503
504                        post( ctx->thrd.sem );
505
506                        __cfadbg_print_safe( io, "Kernel I/O : Added %u to ready for %p\n", idx, active_thread() );
507                }
508                else if( ring.eager_submits ) {
509                        __u32 picked = __submit_to_ready_array( ring, idx, mask );
510
511                        #if defined(LEADER_LOCK)
512                                if( !try_lock(ring.submit_q.submit_lock) ) {
513                                        __STATS__( false,
514                                                io.submit_q.helped += 1;
515                                        )
516                                        return;
517                                }
518                                /* paranoid */ verify( ! __preemption_enabled() );
519                                __STATS__( true,
520                                        io.submit_q.leader += 1;
521                                )
522                        #else
523                                for() {
524                                        yield();
525
526                                        if( try_lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2) ) {
527                                                __STATS__( false,
528                                                        io.submit_q.leader += 1;
529                                                )
530                                                break;
531                                        }
532
533                                        // If some one else collected our index, we are done
534                                        #warning ABA problem
535                                        if( ring.submit_q.ready[picked] != idx ) {
536                                                __STATS__( false,
537                                                        io.submit_q.helped += 1;
538                                                )
539                                                return;
540                                        }
541
542                                        __STATS__( false,
543                                                io.submit_q.busy += 1;
544                                        )
545                                }
546                        #endif
547
548                        // We got the lock
549                        // Collect the submissions
550                        unsigned to_submit = __collect_submitions( ring );
551
552                        // Actually submit
553                        int ret = __io_uring_enter( ring, to_submit, false );
554
555                        #if defined(LEADER_LOCK)
556                                /* paranoid */ verify( ! __preemption_enabled() );
557                                next(ring.submit_q.submit_lock);
558                        #else
559                                unlock(ring.submit_q.submit_lock);
560                        #endif
561                        if( ret < 0 ) {
562                                return;
563                        }
564
565                        // Release the consumed SQEs
566                        __release_consumed_submission( ring );
567
568                        // update statistics
569                        __STATS__( false,
570                                io.submit_q.submit_avg.rdy += to_submit;
571                                io.submit_q.submit_avg.csm += ret;
572                                io.submit_q.submit_avg.cnt += 1;
573                        )
574
575                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u (among %u) for %p\n", idx, ret, active_thread() );
576                }
577                else
578                {
579                        // get mutual exclusion
580                        #if defined(LEADER_LOCK)
581                                while(!try_lock(ring.submit_q.submit_lock));
582                        #else
583                                lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2);
584                        #endif
585
586                        /* paranoid */ verifyf( ring.submit_q.sqes[ idx ].user_data != 3ul64,
587                        /* paranoid */  "index %u already reclaimed\n"
588                        /* paranoid */  "head %u, prev %u, tail %u\n"
589                        /* paranoid */  "[-0: %u,-1: %u,-2: %u,-3: %u]\n",
590                        /* paranoid */  idx,
591                        /* paranoid */  *ring.submit_q.head, ring.submit_q.prev_head, *tail
592                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 0) & (*ring.submit_q.mask) ]
593                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 1) & (*ring.submit_q.mask) ]
594                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 2) & (*ring.submit_q.mask) ]
595                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 3) & (*ring.submit_q.mask) ]
596                        /* paranoid */ );
597
598                        // Append to the list of ready entries
599
600                        /* paranoid */ verify( idx <= mask );
601                        ring.submit_q.array[ (*tail) & mask ] = idx;
602                        __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
603
604                        // Submit however, many entries need to be submitted
605                        int ret = __io_uring_enter( ring, 1, false );
606                        if( ret < 0 ) {
607                                switch((int)errno) {
608                                default:
609                                        abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
610                                }
611                        }
612
613                        /* paranoid */ verify(ret == 1);
614
615                        // update statistics
616                        __STATS__( false,
617                                io.submit_q.submit_avg.csm += 1;
618                                io.submit_q.submit_avg.cnt += 1;
619                        )
620
621                        {
622                                __attribute__((unused)) volatile __u32 * const head = ring.submit_q.head;
623                                __attribute__((unused)) __u32 last_idx = ring.submit_q.array[ ((*head) - 1) & mask ];
624                                __attribute__((unused)) volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[last_idx];
625
626                                __cfadbg_print_safe( io,
627                                        "Kernel I/O : last submitted is %u (%p)\n"
628                                        "    data: %p\n"
629                                        "    opcode: %s\n"
630                                        "    fd: %d\n"
631                                        "    flags: %d\n"
632                                        "    prio: %d\n"
633                                        "    off: %p\n"
634                                        "    addr: %p\n"
635                                        "    len: %d\n"
636                                        "    other flags: %d\n"
637                                        "    splice fd: %d\n"
638                                        "    pad[0]: %llu\n"
639                                        "    pad[1]: %llu\n"
640                                        "    pad[2]: %llu\n",
641                                        last_idx, sqe,
642                                        (void*)sqe->user_data,
643                                        opcodes[sqe->opcode],
644                                        sqe->fd,
645                                        sqe->flags,
646                                        sqe->ioprio,
647                                        sqe->off,
648                                        sqe->addr,
649                                        sqe->len,
650                                        sqe->accept_flags,
651                                        sqe->splice_fd_in,
652                                        sqe->__pad2[0],
653                                        sqe->__pad2[1],
654                                        sqe->__pad2[2]
655                                );
656                        }
657
658                        __atomic_thread_fence( __ATOMIC_SEQ_CST );
659                        // Release the consumed SQEs
660                        __release_consumed_submission( ring );
661                        // ring.submit_q.sqes[idx].user_data = 3ul64;
662
663                        #if defined(LEADER_LOCK)
664                                next(ring.submit_q.submit_lock);
665                        #else
666                                unlock(ring.submit_q.submit_lock);
667                        #endif
668
669                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u for %p\n", idx, active_thread() );
670                }
671        }
672
673        // #define PARTIAL_SUBMIT 32
674
675        // go through the list of submissions in the ready array and moved them into
676        // the ring's submit queue
677        static unsigned __collect_submitions( struct __io_data & ring ) {
678                /* paranoid */ verify( ring.submit_q.ready != 0p );
679                /* paranoid */ verify( ring.submit_q.ready_cnt > 0 );
680
681                unsigned to_submit = 0;
682                __u32 tail = *ring.submit_q.tail;
683                const __u32 mask = *ring.submit_q.mask;
684                #if defined(PARTIAL_SUBMIT)
685                        #if defined(LEADER_LOCK)
686                                #error PARTIAL_SUBMIT and LEADER_LOCK cannot co-exist
687                        #endif
688                        const __u32 cnt = ring.submit_q.ready_cnt > PARTIAL_SUBMIT ? PARTIAL_SUBMIT : ring.submit_q.ready_cnt;
689                        const __u32 offset = ring.submit_q.prev_ready;
690                        ring.submit_q.prev_ready += cnt;
691                #else
692                        const __u32 cnt = ring.submit_q.ready_cnt;
693                        const __u32 offset = 0;
694                #endif
695
696                // Go through the list of ready submissions
697                for( c; cnt ) {
698                        __u32 i = (offset + c) % ring.submit_q.ready_cnt;
699
700                        // replace any submission with the sentinel, to consume it.
701                        __u32 idx = __atomic_exchange_n( &ring.submit_q.ready[i], -1ul32, __ATOMIC_RELAXED);
702
703                        // If it was already the sentinel, then we are done
704                        if( idx == -1ul32 ) continue;
705
706                        // If we got a real submission, append it to the list
707                        ring.submit_q.array[ (tail + to_submit) & mask ] = idx & mask;
708                        to_submit++;
709                }
710
711                // Increment the tail based on how many we are ready to submit
712                __atomic_fetch_add(ring.submit_q.tail, to_submit, __ATOMIC_SEQ_CST);
713
714                return to_submit;
715        }
716
717        // Go through the ring's submit queue and release everything that has already been consumed
718        // by io_uring
719        static __u32 __release_consumed_submission( struct __io_data & ring ) {
720                const __u32 smask = *ring.submit_q.mask;
721
722                // We need to get the lock to copy the old head and new head
723                if( !try_lock(ring.submit_q.release_lock __cfaabi_dbg_ctx2) ) return 0;
724                __attribute__((unused))
725                __u32 ctail = *ring.submit_q.tail;        // get the current tail of the queue
726                __u32 chead = *ring.submit_q.head;              // get the current head of the queue
727                __u32 phead = ring.submit_q.prev_head;  // get the head the last time we were here
728                ring.submit_q.prev_head = chead;                // note up to were we processed
729                unlock(ring.submit_q.release_lock);
730
731                // the 3 fields are organized like this diagram
732                // except it's are ring
733                // ---+--------+--------+----
734                // ---+--------+--------+----
735                //    ^        ^        ^
736                // phead    chead    ctail
737
738                // make sure ctail doesn't wrap around and reach phead
739                /* paranoid */ verify(
740                           (ctail >= chead && chead >= phead)
741                        || (chead >= phead && phead >= ctail)
742                        || (phead >= ctail && ctail >= chead)
743                );
744
745                // find the range we need to clear
746                __u32 count = chead - phead;
747
748                // We acquired an previous-head/current-head range
749                // go through the range and release the sqes
750                for( i; count ) {
751                        __u32 idx = ring.submit_q.array[ (phead + i) & smask ];
752
753                        /* paranoid */ verify( 0 != ring.submit_q.sqes[ idx ].user_data );
754                        ring.submit_q.sqes[ idx ].user_data = 3ul64;
755                }
756                return count;
757        }
758#endif
Note: See TracBrowser for help on using the repository browser.