source: libcfa/src/concurrency/io.cfa @ ece0e80

arm-ehjacob/cs343-translationnew-ast-unique-expr
Last change on this file since ece0e80 was ece0e80, checked in by Thierry Delisle <tdelisle@…>, 9 months ago

Added prints.
Naive implementation of cancel.
Server now shutdown cleanly.

  • Property mode set to 100644
File size: 16.8 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/syscall.h>
35
36                #include <linux/io_uring.h>
37        }
38
39        #include "stats.hfa"
40        #include "kernel.hfa"
41        #include "kernel/fwd.hfa"
42        #include "io/types.hfa"
43
44        // returns true of acquired as leader or second leader
45        static inline bool try_lock( __leaderlock_t & this ) {
46                const uintptr_t thrd = 1z | (uintptr_t)active_thread();
47                bool block;
48                disable_interrupts();
49                for() {
50                        struct $thread * expected = this.value;
51                        if( 1p != expected && 0p != expected ) {
52                                /* paranoid */ verify( thrd != (uintptr_t)expected ); // We better not already be the next leader
53                                enable_interrupts( __cfaabi_dbg_ctx );
54                                return false;
55                        }
56                        struct $thread * desired;
57                        if( 0p == expected ) {
58                                // If the lock isn't locked acquire it, no need to block
59                                desired = 1p;
60                                block = false;
61                        }
62                        else {
63                                // If the lock is already locked try becomming the next leader
64                                desired = (struct $thread *)thrd;
65                                block = true;
66                        }
67                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
68                }
69                if( block ) {
70                        enable_interrupts( __cfaabi_dbg_ctx );
71                        park();
72                        disable_interrupts();
73                }
74                return true;
75        }
76
77        static inline bool next( __leaderlock_t & this ) {
78                /* paranoid */ verify( ! __preemption_enabled() );
79                struct $thread * nextt;
80                for() {
81                        struct $thread * expected = this.value;
82                        /* paranoid */ verify( (1 & (uintptr_t)expected) == 1 ); // The lock better be locked
83
84                        struct $thread * desired;
85                        if( 1p == expected ) {
86                                // No next leader, just unlock
87                                desired = 0p;
88                                nextt   = 0p;
89                        }
90                        else {
91                                // There is a next leader, remove but keep locked
92                                desired = 1p;
93                                nextt   = (struct $thread *)(~1z & (uintptr_t)expected);
94                        }
95                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
96                }
97
98                if(nextt) {
99                        unpark( nextt );
100                        enable_interrupts( __cfaabi_dbg_ctx );
101                        return true;
102                }
103                enable_interrupts( __cfaabi_dbg_ctx );
104                return false;
105        }
106
107//=============================================================================================
108// I/O Syscall
109//=============================================================================================
110        static int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get ) {
111                bool need_sys_to_submit = false;
112                bool need_sys_to_complete = false;
113                unsigned flags = 0;
114
115                TO_SUBMIT:
116                if( to_submit > 0 ) {
117                        if( !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
118                                need_sys_to_submit = true;
119                                break TO_SUBMIT;
120                        }
121                        if( (*ring.submit_q.flags) & IORING_SQ_NEED_WAKEUP ) {
122                                need_sys_to_submit = true;
123                                flags |= IORING_ENTER_SQ_WAKEUP;
124                        }
125                }
126
127                if( get && !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
128                        flags |= IORING_ENTER_GETEVENTS;
129                        if( (ring.ring_flags & IORING_SETUP_IOPOLL) ) {
130                                need_sys_to_complete = true;
131                        }
132                }
133
134                int ret = 0;
135                if( need_sys_to_submit || need_sys_to_complete ) {
136                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING enter %d %u %u\n", ring.fd, to_submit, flags);
137                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, (sigset_t *)0p, _NSIG / 8);
138                        if( ret < 0 ) {
139                                switch((int)errno) {
140                                case EAGAIN:
141                                case EINTR:
142                                        ret = -1;
143                                        break;
144                                default:
145                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
146                                }
147                        }
148                }
149
150                // Memory barrier
151                __atomic_thread_fence( __ATOMIC_SEQ_CST );
152                return ret;
153        }
154
155//=============================================================================================
156// I/O Polling
157//=============================================================================================
158        static unsigned __collect_submitions( struct __io_data & ring );
159        static __u32 __release_consumed_submission( struct __io_data & ring );
160
161        static inline void process(struct io_uring_cqe & cqe ) {
162                struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
163                __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
164
165                fulfil( *future, cqe.res );
166        }
167
168        // Process a single completion message from the io_uring
169        // This is NOT thread-safe
170        static [int, bool] __drain_io( & struct __io_data ring ) {
171                /* paranoid */ verify( ! __preemption_enabled() );
172
173                unsigned to_submit = 0;
174                if( ring.poller_submits ) {
175                        // If the poller thread also submits, then we need to aggregate the submissions which are ready
176                        to_submit = __collect_submitions( ring );
177                }
178
179                int ret = __io_uring_enter(ring, to_submit, true);
180                if( ret < 0 ) {
181                        return [0, true];
182                }
183
184                // update statistics
185                if (to_submit > 0) {
186                        __STATS__( true,
187                                if( to_submit > 0 ) {
188                                        io.submit_q.submit_avg.rdy += to_submit;
189                                        io.submit_q.submit_avg.csm += ret;
190                                        io.submit_q.submit_avg.cnt += 1;
191                                }
192                        )
193                }
194
195                // Release the consumed SQEs
196                __release_consumed_submission( ring );
197
198                // Drain the queue
199                unsigned head = *ring.completion_q.head;
200                unsigned tail = *ring.completion_q.tail;
201                const __u32 mask = *ring.completion_q.mask;
202
203                // Nothing was new return 0
204                if (head == tail) {
205                        return [0, to_submit > 0];
206                }
207
208                __u32 count = tail - head;
209                /* paranoid */ verify( count != 0 );
210                for(i; count) {
211                        unsigned idx = (head + i) & mask;
212                        struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
213
214                        /* paranoid */ verify(&cqe);
215
216                        process( cqe );
217                }
218
219                // Mark to the kernel that the cqe has been seen
220                // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
221                __atomic_thread_fence( __ATOMIC_SEQ_CST );
222                __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED );
223
224                return [count, count > 0 || to_submit > 0];
225        }
226
227        void main( $io_ctx_thread & this ) {
228                epoll_event ev;
229                __ioctx_register( this, ev );
230
231                __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %p for ring %p ready\n", &this, &this.ring);
232
233                const int reset_cnt = 5;
234                int reset = reset_cnt;
235                // Then loop until we need to start
236                LOOP:
237                while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
238                        // Drain the io
239                        int count;
240                        bool again;
241                        disable_interrupts();
242                                [count, again] = __drain_io( *this.ring );
243
244                                if(!again) reset--;
245
246                                // Update statistics
247                                __STATS__( true,
248                                        io.complete_q.completed_avg.val += count;
249                                        io.complete_q.completed_avg.cnt += 1;
250                                )
251                        enable_interrupts( __cfaabi_dbg_ctx );
252
253                        // If we got something, just yield and check again
254                        if(reset > 1) {
255                                yield();
256                                continue LOOP;
257                        }
258
259                        // We alread failed to find events a few time.
260                        if(reset == 1) {
261                                // Rearm the context so it can block
262                                // but don't block right away
263                                // we need to retry one last time in case
264                                // something completed *just now*
265                                __ioctx_prepare_block( this, ev );
266                                continue LOOP;
267                        }
268
269                                __STATS__( false,
270                                        io.complete_q.blocks += 1;
271                                )
272                                __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %p\n", &this.self);
273
274                                // block this thread
275                                wait( this.sem );
276
277                        // restore counter
278                        reset = reset_cnt;
279                }
280
281                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring);
282        }
283
284//=============================================================================================
285// I/O Submissions
286//=============================================================================================
287
288// Submition steps :
289// 1 - Allocate a queue entry. The ring already has memory for all entries but only the ones
290//     listed in sq.array are visible by the kernel. For those not listed, the kernel does not
291//     offer any assurance that an entry is not being filled by multiple flags. Therefore, we
292//     need to write an allocator that allows allocating concurrently.
293//
294// 2 - Actually fill the submit entry, this is the only simple and straightforward step.
295//
296// 3 - Append the entry index to the array and adjust the tail accordingly. This operation
297//     needs to arrive to two concensus at the same time:
298//     A - The order in which entries are listed in the array: no two threads must pick the
299//         same index for their entries
300//     B - When can the tail be update for the kernel. EVERY entries in the array between
301//         head and tail must be fully filled and shouldn't ever be touched again.
302//
303
304        [* struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data ) {
305                /* paranoid */ verify( data != 0 );
306
307                // Prepare the data we need
308                __attribute((unused)) int len   = 0;
309                __attribute((unused)) int block = 0;
310                __u32 cnt = *ring.submit_q.num;
311                __u32 mask = *ring.submit_q.mask;
312
313                __u32 off = thread_rand();
314
315                // Loop around looking for an available spot
316                for() {
317                        // Look through the list starting at some offset
318                        for(i; cnt) {
319                                __u64 expected = 0;
320                                __u32 idx = (i + off) & mask;
321                                struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
322                                volatile __u64 * udata = &sqe->user_data;
323
324                                if( *udata == expected &&
325                                        __atomic_compare_exchange_n( udata, &expected, data, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) )
326                                {
327                                        // update statistics
328                                        __STATS__( false,
329                                                io.submit_q.alloc_avg.val   += len;
330                                                io.submit_q.alloc_avg.block += block;
331                                                io.submit_q.alloc_avg.cnt   += 1;
332                                        )
333
334                                        __cfadbg_print_safe( io, "Kernel I/O : allocated [%p, %u] for %p (%p)\n", sqe, idx, active_thread(), (void*)data );
335
336                                        // Success return the data
337                                        return [sqe, idx];
338                                }
339                                verify(expected != data);
340
341                                len ++;
342                        }
343
344                        block++;
345                        yield();
346                }
347        }
348
349        static inline __u32 __submit_to_ready_array( struct __io_data & ring, __u32 idx, const __u32 mask ) {
350                /* paranoid */ verify( idx <= mask   );
351                /* paranoid */ verify( idx != -1ul32 );
352
353                // We need to find a spot in the ready array
354                __attribute((unused)) int len   = 0;
355                __attribute((unused)) int block = 0;
356                __u32 ready_mask = ring.submit_q.ready_cnt - 1;
357
358                __u32 off = thread_rand();
359
360                __u32 picked;
361                LOOKING: for() {
362                        for(i; ring.submit_q.ready_cnt) {
363                                picked = (i + off) & ready_mask;
364                                __u32 expected = -1ul32;
365                                if( __atomic_compare_exchange_n( &ring.submit_q.ready[picked], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
366                                        break LOOKING;
367                                }
368                                verify(expected != idx);
369
370                                len ++;
371                        }
372
373                        block++;
374
375                        __u32 released = __release_consumed_submission( ring );
376                        if( released == 0 ) {
377                                yield();
378                        }
379                }
380
381                // update statistics
382                __STATS__( false,
383                        io.submit_q.look_avg.val   += len;
384                        io.submit_q.look_avg.block += block;
385                        io.submit_q.look_avg.cnt   += 1;
386                )
387
388                return picked;
389        }
390
391        void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
392                __cfadbg_print_safe( io, "Kernel I/O : submitting %u for %p\n", idx, active_thread() );
393
394                __io_data & ring = *ctx->thrd.ring;
395                // Get now the data we definetely need
396                volatile __u32 * const tail = ring.submit_q.tail;
397                const __u32 mask  = *ring.submit_q.mask;
398
399                // There are 2 submission schemes, check which one we are using
400                if( ring.poller_submits ) {
401                        // If the poller thread submits, then we just need to add this to the ready array
402                        __submit_to_ready_array( ring, idx, mask );
403
404                        post( ctx->thrd.sem );
405
406                        __cfadbg_print_safe( io, "Kernel I/O : Added %u to ready for %p\n", idx, active_thread() );
407                }
408                else if( ring.eager_submits ) {
409                        __u32 picked = __submit_to_ready_array( ring, idx, mask );
410
411                        #if defined(LEADER_LOCK)
412                                if( !try_lock(ring.submit_q.submit_lock) ) {
413                                        __STATS__( false,
414                                                io.submit_q.helped += 1;
415                                        )
416                                        return;
417                                }
418                                /* paranoid */ verify( ! __preemption_enabled() );
419                                __STATS__( true,
420                                        io.submit_q.leader += 1;
421                                )
422                        #else
423                                for() {
424                                        yield();
425
426                                        if( try_lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2) ) {
427                                                __STATS__( false,
428                                                        io.submit_q.leader += 1;
429                                                )
430                                                break;
431                                        }
432
433                                        // If some one else collected our index, we are done
434                                        #warning ABA problem
435                                        if( ring.submit_q.ready[picked] != idx ) {
436                                                __STATS__( false,
437                                                        io.submit_q.helped += 1;
438                                                )
439                                                return;
440                                        }
441
442                                        __STATS__( false,
443                                                io.submit_q.busy += 1;
444                                        )
445                                }
446                        #endif
447
448                        // We got the lock
449                        // Collect the submissions
450                        unsigned to_submit = __collect_submitions( ring );
451
452                        // Actually submit
453                        int ret = __io_uring_enter( ring, to_submit, false );
454
455                        #if defined(LEADER_LOCK)
456                                /* paranoid */ verify( ! __preemption_enabled() );
457                                next(ring.submit_q.submit_lock);
458                        #else
459                                unlock(ring.submit_q.submit_lock);
460                        #endif
461                        if( ret < 0 ) {
462                                return;
463                        }
464
465                        // Release the consumed SQEs
466                        __release_consumed_submission( ring );
467
468                        // update statistics
469                        __STATS__( false,
470                                io.submit_q.submit_avg.rdy += to_submit;
471                                io.submit_q.submit_avg.csm += ret;
472                                io.submit_q.submit_avg.cnt += 1;
473                        )
474
475                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u (among %u) for %p\n", idx, ret, active_thread() );
476                }
477                else {
478                        // get mutual exclusion
479                        #if defined(LEADER_LOCK)
480                                while(!try_lock(ring.submit_q.submit_lock));
481                        #else
482                                lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2);
483                        #endif
484
485                        /* paranoid */ verifyf( ring.submit_q.sqes[ idx ].user_data != 0,
486                        /* paranoid */  "index %u already reclaimed\n"
487                        /* paranoid */  "head %u, prev %u, tail %u\n"
488                        /* paranoid */  "[-0: %u,-1: %u,-2: %u,-3: %u]\n",
489                        /* paranoid */  idx,
490                        /* paranoid */  *ring.submit_q.head, ring.submit_q.prev_head, *tail
491                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 0) & (*ring.submit_q.mask) ]
492                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 1) & (*ring.submit_q.mask) ]
493                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 2) & (*ring.submit_q.mask) ]
494                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 3) & (*ring.submit_q.mask) ]
495                        /* paranoid */ );
496
497                        // Append to the list of ready entries
498
499                        /* paranoid */ verify( idx <= mask );
500                        ring.submit_q.array[ (*tail) & mask ] = idx;
501                        __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
502
503                        // Submit however, many entries need to be submitted
504                        int ret = __io_uring_enter( ring, 1, false );
505                        if( ret < 0 ) {
506                                switch((int)errno) {
507                                default:
508                                        abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
509                                }
510                        }
511
512                        // update statistics
513                        __STATS__( false,
514                                io.submit_q.submit_avg.csm += 1;
515                                io.submit_q.submit_avg.cnt += 1;
516                        )
517
518                        // Release the consumed SQEs
519                        __release_consumed_submission( ring );
520
521                        #if defined(LEADER_LOCK)
522                                next(ring.submit_q.submit_lock);
523                        #else
524                                unlock(ring.submit_q.submit_lock);
525                        #endif
526
527                        __cfadbg_print_safe( io, "Kernel I/O : Performed io_submit for %p, returned %d\n", active_thread(), ret );
528                }
529        }
530
531        // #define PARTIAL_SUBMIT 32
532        static unsigned __collect_submitions( struct __io_data & ring ) {
533                /* paranoid */ verify( ring.submit_q.ready != 0p );
534                /* paranoid */ verify( ring.submit_q.ready_cnt > 0 );
535
536                unsigned to_submit = 0;
537                __u32 tail = *ring.submit_q.tail;
538                const __u32 mask = *ring.submit_q.mask;
539                #if defined(PARTIAL_SUBMIT)
540                        #if defined(LEADER_LOCK)
541                                #error PARTIAL_SUBMIT and LEADER_LOCK cannot co-exist
542                        #endif
543                        const __u32 cnt = ring.submit_q.ready_cnt > PARTIAL_SUBMIT ? PARTIAL_SUBMIT : ring.submit_q.ready_cnt;
544                        const __u32 offset = ring.submit_q.prev_ready;
545                        ring.submit_q.prev_ready += cnt;
546                #else
547                        const __u32 cnt = ring.submit_q.ready_cnt;
548                        const __u32 offset = 0;
549                #endif
550
551                // Go through the list of ready submissions
552                for( c; cnt ) {
553                        __u32 i = (offset + c) % ring.submit_q.ready_cnt;
554
555                        // replace any submission with the sentinel, to consume it.
556                        __u32 idx = __atomic_exchange_n( &ring.submit_q.ready[i], -1ul32, __ATOMIC_RELAXED);
557
558                        // If it was already the sentinel, then we are done
559                        if( idx == -1ul32 ) continue;
560
561                        // If we got a real submission, append it to the list
562                        ring.submit_q.array[ (tail + to_submit) & mask ] = idx & mask;
563                        to_submit++;
564                }
565
566                // Increment the tail based on how many we are ready to submit
567                __atomic_fetch_add(ring.submit_q.tail, to_submit, __ATOMIC_SEQ_CST);
568
569                return to_submit;
570        }
571
572        static __u32 __release_consumed_submission( struct __io_data & ring ) {
573                const __u32 smask = *ring.submit_q.mask;
574
575                if( !try_lock(ring.submit_q.release_lock __cfaabi_dbg_ctx2) ) return 0;
576                __u32 chead = *ring.submit_q.head;
577                __u32 phead = ring.submit_q.prev_head;
578                ring.submit_q.prev_head = chead;
579                unlock(ring.submit_q.release_lock);
580
581                __u32 count = chead - phead;
582                for( i; count ) {
583                        __u32 idx = ring.submit_q.array[ (phead + i) & smask ];
584                        ring.submit_q.sqes[ idx ].user_data = 0;
585                }
586                return count;
587        }
588#endif
Note: See TracBrowser for help on using the repository browser.