source: libcfa/src/concurrency/io.cfa @ 05cfa4d

arm-ehenumforall-pointer-decayjacob/cs343-translationnew-astnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 05cfa4d was 05cfa4d, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
  • Added 'fix' of the deadlock on cluster destruction by taking preempted threads and pretending they were blocked.
  • Added statistics for when submition blocked to obtain an sqe.
  • Property mode set to 100644
File size: 33.3 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 __CFA_DEBUG_PRINT_IO__
17// #define __CFA_DEBUG_PRINT_IO_CORE__
18
19#include "kernel.hfa"
20
21#if !defined(HAVE_LINUX_IO_URING_H)
22        void __kernel_io_startup( cluster &, int, bool ) {
23                // Nothing to do without io_uring
24        }
25
26        void __kernel_io_finish_start( cluster & ) {
27                // Nothing to do without io_uring
28        }
29
30        void __kernel_io_prepare_stop( cluster & ) {
31                // Nothing to do without io_uring
32        }
33
34        void __kernel_io_shutdown( cluster &, bool ) {
35                // Nothing to do without io_uring
36        }
37
38#else
39        extern "C" {
40                #define _GNU_SOURCE         /* See feature_test_macros(7) */
41                #include <errno.h>
42                #include <stdint.h>
43                #include <string.h>
44                #include <unistd.h>
45                #include <sys/mman.h>
46                #include <sys/syscall.h>
47
48                #include <linux/io_uring.h>
49        }
50
51        #include "bits/signal.hfa"
52        #include "kernel_private.hfa"
53        #include "thread.hfa"
54
55        uint32_t entries_per_cluster() {
56                return 256;
57        }
58
59        static void * __io_poller_slow( void * arg );
60
61        // Weirdly, some systems that do support io_uring don't actually define these
62        #ifdef __alpha__
63                /*
64                * alpha is the only exception, all other architectures
65                * have common numbers for new system calls.
66                */
67                #ifndef __NR_io_uring_setup
68                        #define __NR_io_uring_setup           535
69                #endif
70                #ifndef __NR_io_uring_enter
71                        #define __NR_io_uring_enter           536
72                #endif
73                #ifndef __NR_io_uring_register
74                        #define __NR_io_uring_register        537
75                #endif
76        #else /* !__alpha__ */
77                #ifndef __NR_io_uring_setup
78                        #define __NR_io_uring_setup           425
79                #endif
80                #ifndef __NR_io_uring_enter
81                        #define __NR_io_uring_enter           426
82                #endif
83                #ifndef __NR_io_uring_register
84                        #define __NR_io_uring_register        427
85                #endif
86        #endif
87
88        // Fast poller user-thread
89        // Not using the "thread" keyword because we want to control
90        // more carefully when to start/stop it
91        struct __io_poller_fast {
92                struct __io_data * ring;
93                bool waiting;
94                $thread thrd;
95        };
96
97        void ?{}( __io_poller_fast & this, struct cluster & cltr ) {
98                this.ring = cltr.io;
99                this.waiting = true;
100                (this.thrd){ "Fast I/O Poller", cltr };
101        }
102        void ^?{}( __io_poller_fast & mutex this );
103        void main( __io_poller_fast & this );
104        static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; }
105        void ^?{}( __io_poller_fast & mutex this ) {}
106
107        struct __submition_data {
108                // Head and tail of the ring (associated with array)
109                volatile uint32_t * head;
110                volatile uint32_t * tail;
111
112                // The actual kernel ring which uses head/tail
113                // indexes into the sqes arrays
114                uint32_t * array;
115
116                // number of entries and mask to go with it
117                const uint32_t * num;
118                const uint32_t * mask;
119
120                // Submission flags (Not sure what for)
121                uint32_t * flags;
122
123                // number of sqes not submitted (whatever that means)
124                uint32_t * dropped;
125
126                // Like head/tail but not seen by the kernel
127                volatile uint32_t alloc;
128                volatile uint32_t ready;
129
130                __spinlock_t lock;
131
132                // A buffer of sqes (not the actual ring)
133                struct io_uring_sqe * sqes;
134
135                // The location and size of the mmaped area
136                void * ring_ptr;
137                size_t ring_sz;
138
139                // Statistics
140                #if !defined(__CFA_NO_STATISTICS__)
141                        struct {
142                                struct {
143                                        volatile unsigned long long int val;
144                                        volatile unsigned long long int cnt;
145                                        volatile unsigned long long int block;
146                                } submit_avg;
147                        } stats;
148                #endif
149        };
150
151        struct __completion_data {
152                // Head and tail of the ring
153                volatile uint32_t * head;
154                volatile uint32_t * tail;
155
156                // number of entries and mask to go with it
157                const uint32_t * mask;
158                const uint32_t * num;
159
160                // number of cqes not submitted (whatever that means)
161                uint32_t * overflow;
162
163                // the kernel ring
164                struct io_uring_cqe * cqes;
165
166                // The location and size of the mmaped area
167                void * ring_ptr;
168                size_t ring_sz;
169
170                // Statistics
171                #if !defined(__CFA_NO_STATISTICS__)
172                        struct {
173                                struct {
174                                        unsigned long long int val;
175                                        unsigned long long int slow_cnt;
176                                        unsigned long long int fast_cnt;
177                                } completed_avg;
178                        } stats;
179                #endif
180        };
181
182        struct __io_data {
183                struct __submition_data submit_q;
184                struct __completion_data completion_q;
185                uint32_t ring_flags;
186                int cltr_flags;
187                int fd;
188                semaphore submit;
189                volatile bool done;
190                struct {
191                        struct {
192                                void * stack;
193                                pthread_t kthrd;
194                        } slow;
195                        __io_poller_fast fast;
196                        __bin_sem_t sem;
197                } poller;
198        };
199
200//=============================================================================================
201// I/O Startup / Shutdown logic
202//=============================================================================================
203        void __kernel_io_startup( cluster & this, int io_flags, bool main_cluster ) {
204                this.io = malloc();
205
206                // Step 1 : call to setup
207                struct io_uring_params params;
208                memset(&params, 0, sizeof(params));
209
210                uint32_t nentries = entries_per_cluster();
211
212                int fd = syscall(__NR_io_uring_setup, nentries, &params );
213                if(fd < 0) {
214                        abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
215                }
216
217                // Step 2 : mmap result
218                memset( this.io, 0, sizeof(struct __io_data) );
219                struct __submition_data  & sq = this.io->submit_q;
220                struct __completion_data & cq = this.io->completion_q;
221
222                // calculate the right ring size
223                sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned)           );
224                cq.ring_sz = params.cq_off.cqes  + (params.cq_entries * sizeof(struct io_uring_cqe));
225
226                // Requires features
227                #if defined(IORING_FEAT_SINGLE_MMAP)
228                        // adjust the size according to the parameters
229                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
230                                cq->ring_sz = sq->ring_sz = max(cq->ring_sz, sq->ring_sz);
231                        }
232                #endif
233
234                // mmap the Submit Queue into existence
235                sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING);
236                if (sq.ring_ptr == (void*)MAP_FAILED) {
237                        abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
238                }
239
240                // Requires features
241                #if defined(IORING_FEAT_SINGLE_MMAP)
242                        // mmap the Completion Queue into existence (may or may not be needed)
243                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
244                                cq->ring_ptr = sq->ring_ptr;
245                        }
246                        else
247                #endif
248                {
249                        // We need multiple call to MMAP
250                        cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING);
251                        if (cq.ring_ptr == (void*)MAP_FAILED) {
252                                munmap(sq.ring_ptr, sq.ring_sz);
253                                abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
254                        }
255                }
256
257                // mmap the submit queue entries
258                size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
259                sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES);
260                if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
261                        munmap(sq.ring_ptr, sq.ring_sz);
262                        if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
263                        abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
264                }
265
266                // Get the pointers from the kernel to fill the structure
267                // submit queue
268                sq.head    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
269                sq.tail    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
270                sq.mask    = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
271                sq.num     = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
272                sq.flags   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
273                sq.dropped = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
274                sq.array   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
275                sq.alloc = *sq.tail;
276                sq.ready = *sq.tail;
277
278                // completion queue
279                cq.head     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
280                cq.tail     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
281                cq.mask     = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
282                cq.num      = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
283                cq.overflow = (         uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
284                cq.cqes   = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
285
286                // some paranoid checks
287                /* paranoid */ verifyf( (*cq.mask) == ((*cq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*cq.num) - 1ul32, *cq.num, *cq.mask  );
288                /* paranoid */ verifyf( (*cq.num)  >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
289                /* paranoid */ verifyf( (*cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
290                /* paranoid */ verifyf( (*cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
291
292                /* paranoid */ verifyf( (*sq.mask) == ((*sq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*sq.num) - 1ul32, *sq.num, *sq.mask );
293                /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
294                /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
295                /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
296
297                // Update the global ring info
298                this.io->ring_flags = params.flags;
299                this.io->cltr_flags = io_flags;
300                this.io->fd         = fd;
301                this.io->done       = false;
302                (this.io->submit){ min(*sq.num, *cq.num) };
303
304                // Initialize statistics
305                #if !defined(__CFA_NO_STATISTICS__)
306                        this.io->submit_q.stats.submit_avg.val   = 0;
307                        this.io->submit_q.stats.submit_avg.cnt   = 0;
308                        this.io->submit_q.stats.submit_avg.block = 0;
309                        this.io->completion_q.stats.completed_avg.val = 0;
310                        this.io->completion_q.stats.completed_avg.slow_cnt = 0;
311                        this.io->completion_q.stats.completed_avg.fast_cnt = 0;
312                #endif
313
314                if(!main_cluster) {
315                        __kernel_io_finish_start( this );
316                }
317        }
318
319        void __kernel_io_finish_start( cluster & this ) {
320                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
321                        __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
322                        (this.io->poller.fast){ this };
323                        __thrd_start( this.io->poller.fast, main );
324                }
325
326                // Create the poller thread
327                __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
328                this.io->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );
329        }
330
331        void __kernel_io_prepare_stop( cluster & this ) {
332                __cfadbg_print_safe(io_core, "Kernel I/O : Stopping pollers for cluster\n", &this);
333                // Notify the poller thread of the shutdown
334                __atomic_store_n(&this.io->done, true, __ATOMIC_SEQ_CST);
335
336                // Stop the IO Poller
337                sigval val = { 1 };
338                pthread_sigqueue( this.io->poller.slow.kthrd, SIGUSR1, val );
339                post( this.io->poller.sem );
340
341                // Wait for the poller thread to finish
342                pthread_join( this.io->poller.slow.kthrd, 0p );
343                free( this.io->poller.slow.stack );
344
345                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this);
346
347                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
348                        with( this.io->poller.fast ) {
349                                /* paranoid */ verify( waiting ); // The thread shouldn't be in a system call
350                                /* paranoid */ verify( this.procs.head == 0p || &this == mainCluster );
351                                /* paranoid */ verify( this.idles.head == 0p || &this == mainCluster );
352
353                                // We need to adjust the clean-up based on where the thread is
354                                if( thrd.preempted != __NO_PREEMPTION ) {
355
356                                        // This is the tricky case
357                                        // The thread was preempted and now it is on the ready queue
358                                        /* paranoid */ verify( thrd.state == Active );           // The thread better be in this state
359                                        /* paranoid */ verify( thrd.next == 1p );                // The thread should be the last on the list
360                                        /* paranoid */ verify( this.ready_queue.head == &thrd ); // The thread should be the only thing on the list
361
362                                        // Remove the thread from the ready queue of this cluster
363                                        this.ready_queue.head = 1p;
364                                        thrd.next = 0p;
365
366                                        // Fixup the thread state
367                                        thrd.state = Blocked;
368                                        thrd.preempted = __NO_PREEMPTION;
369
370                                        // Pretend like the thread was blocked all along
371                                }
372                                // !!! This is not an else if !!!
373                                if( thrd.state == Blocked ) {
374
375                                        // This is the "easy case"
376                                        // The thread is parked and can easily be moved to active cluster
377                                        verify( thrd.curr_cluster != active_cluster() || thrd.curr_cluster == mainCluster );
378                                        thrd.curr_cluster = active_cluster();
379
380                        // unpark the fast io_poller
381                                        unpark( &thrd __cfaabi_dbg_ctx2 );
382                                }
383                                else {
384
385                                        // The thread is in a weird state
386                                        // I don't know what to do here
387                                        abort("Fast poller thread is in unexpected state, cannot clean-up correctly\n");
388                                }
389
390                        }
391
392                        ^(this.io->poller.fast){};
393
394                        __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this);
395                }
396        }
397
398        void __kernel_io_shutdown( cluster & this, bool main_cluster ) {
399                if(!main_cluster) {
400                        __kernel_io_prepare_stop( this );
401                }
402
403                // print statistics
404                #if !defined(__CFA_NO_STATISTICS__)
405                        if(this.print_stats) {
406                                with(this.io->submit_q.stats, this.io->completion_q.stats) {
407                                        __cfaabi_bits_print_safe( STDERR_FILENO,
408                                                "----- I/O uRing Stats -----\n"
409                                                "- total submit calls  : %'llu\n"
410                                                "- avg submit          : %'.2lf\n"
411                                                "- pre-submit block %%  : %'.2lf\n"
412                                                "- total wait calls    : %'llu (%'llu slow, %'llu fast)\n"
413                                                "- avg completion/wait : %'.2lf\n",
414                                                submit_avg.cnt,
415                                                ((double)submit_avg.val) / submit_avg.cnt,
416                                                (100.0 * submit_avg.block) / submit_avg.cnt,
417                                                completed_avg.slow_cnt + completed_avg.fast_cnt,
418                                                completed_avg.slow_cnt,  completed_avg.fast_cnt,
419                                                ((double)completed_avg.val) / (completed_avg.slow_cnt + completed_avg.fast_cnt)
420                                        );
421                                }
422                        }
423                #endif
424
425                // Shutdown the io rings
426                struct __submition_data  & sq = this.io->submit_q;
427                struct __completion_data & cq = this.io->completion_q;
428
429                // unmap the submit queue entries
430                munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe));
431
432                // unmap the Submit Queue ring
433                munmap(sq.ring_ptr, sq.ring_sz);
434
435                // unmap the Completion Queue ring, if it is different
436                if (cq.ring_ptr != sq.ring_ptr) {
437                        munmap(cq.ring_ptr, cq.ring_sz);
438                }
439
440                // close the file descriptor
441                close(this.io->fd);
442
443                free( this.io );
444        }
445
446//=============================================================================================
447// I/O Polling
448//=============================================================================================
449        struct io_user_data {
450                int32_t result;
451                $thread * thrd;
452        };
453
454        // Process a single completion message from the io_uring
455        // This is NOT thread-safe
456        static int __drain_io( struct __io_data & ring, sigset_t * mask, int waitcnt, bool in_kernel ) {
457                int ret = syscall( __NR_io_uring_enter, ring.fd, 0, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8);
458                if( ret < 0 ) {
459                        switch((int)errno) {
460                        case EAGAIN:
461                        case EINTR:
462                                return -EAGAIN;
463                        default:
464                                abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
465                        }
466                }
467
468                // Drain the queue
469                unsigned head = *ring.completion_q.head;
470                unsigned tail = __atomic_load_n(ring.completion_q.tail, __ATOMIC_ACQUIRE);
471
472                // Nothing was new return 0
473                if (head == tail) {
474                        return 0;
475                }
476
477                uint32_t count = tail - head;
478                for(i; count) {
479                        unsigned idx = (head + i) & (*ring.completion_q.mask);
480                        struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
481
482                        /* paranoid */ verify(&cqe);
483
484                        struct io_user_data * data = (struct io_user_data *)cqe.user_data;
485                        __cfadbg_print_safe( io, "Kernel I/O : Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
486
487                        data->result = cqe.res;
488                        if(!in_kernel) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
489                        else         { __unpark( data->thrd __cfaabi_dbg_ctx2 ); }
490                }
491
492                // Allow new submissions to happen
493                V(ring.submit, count);
494
495                // Mark to the kernel that the cqe has been seen
496                // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
497                __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED );
498
499                return count;
500        }
501
502        static void * __io_poller_slow( void * arg ) {
503                cluster * cltr = (cluster *)arg;
504                struct __io_data & ring = *cltr->io;
505
506                sigset_t mask;
507                sigfillset(&mask);
508                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
509                        abort( "KERNEL ERROR: IO_URING - pthread_sigmask" );
510                }
511
512                sigdelset( &mask, SIGUSR1 );
513
514                verify( (*ring.submit_q.head) == (*ring.submit_q.tail) );
515                verify( (*ring.completion_q.head) == (*ring.completion_q.tail) );
516
517                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);
518
519                if( ring.cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
520                        while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
521                                // In the user-thread approach drain and if anything was drained,
522                                // batton pass to the user-thread
523                                int count = __drain_io( ring, &mask, 1, true );
524
525                                // Update statistics
526                                #if !defined(__CFA_NO_STATISTICS__)
527                                        ring.completion_q.stats.completed_avg.val += count;
528                                        ring.completion_q.stats.completed_avg.slow_cnt += 1;
529                                #endif
530
531                                if(count > 0) {
532                                        __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to fast poller\n", &ring);
533                                        __unpark( &ring.poller.fast.thrd __cfaabi_dbg_ctx2 );
534                                        wait( ring.poller.sem );
535                                }
536                        }
537                }
538                else {
539                        while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
540                                //In the naive approach, just poll the io completion queue directly
541                                int count = __drain_io( ring, &mask, 1, true );
542
543                                // Update statistics
544                                #if !defined(__CFA_NO_STATISTICS__)
545                                        ring.completion_q.stats.completed_avg.val += count;
546                                        ring.completion_q.stats.completed_avg.slow_cnt += 1;
547                                #endif
548                        }
549                }
550
551                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p stopping\n", &ring);
552
553                return 0p;
554        }
555
556        void main( __io_poller_fast & this ) {
557                verify( this.ring->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD );
558
559                // Start parked
560                park( __cfaabi_dbg_ctx );
561
562                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p ready\n", &this.ring);
563
564                int reset = 0;
565
566                // Then loop until we need to start
567                while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) {
568                        // Drain the io
569                        this.waiting = false;
570                        int count = __drain_io( *this.ring, 0p, 0, false );
571                        reset += count > 0 ? 1 : 0;
572
573                        // Update statistics
574                        #if !defined(__CFA_NO_STATISTICS__)
575                                this.ring->completion_q.stats.completed_avg.val += count;
576                                this.ring->completion_q.stats.completed_avg.fast_cnt += 1;
577                        #endif
578
579                        this.waiting = true;
580                        if(reset < 5) {
581                                // If we got something, just yield and check again
582                                yield();
583                        }
584                        else {
585                                // We didn't get anything baton pass to the slow poller
586                                __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring);
587                                post( this.ring->poller.sem );
588                                park( __cfaabi_dbg_ctx );
589                                reset = 0;
590                        }
591                }
592
593                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring);
594        }
595
596//=============================================================================================
597// I/O Submissions
598//=============================================================================================
599
600// Submition steps :
601// 1 - We need to make sure we don't overflow any of the buffer, P(ring.submit) to make sure
602//     entries are available. The semaphore make sure that there is no more operations in
603//     progress then the number of entries in the buffer. This probably limits concurrency
604//     more than necessary since submitted but not completed operations don't need any
605//     entries in user space. However, I don't know what happens if we overflow the buffers
606//     because too many requests completed at once. This is a safe approach in all cases.
607//     Furthermore, with hundreds of entries, this may be okay.
608//
609// 2 - Allocate a queue entry. The ring already has memory for all entries but only the ones
610//     listed in sq.array are visible by the kernel. For those not listed, the kernel does not
611//     offer any assurance that an entry is not being filled by multiple flags. Therefore, we
612//     need to write an allocator that allows allocating concurrently.
613//
614// 3 - Actually fill the submit entry, this is the only simple and straightforward step.
615//
616// 4 - Append the entry index to the array and adjust the tail accordingly. This operation
617//     needs to arrive to two concensus at the same time:
618//     A - The order in which entries are listed in the array: no two threads must pick the
619//         same index for their entries
620//     B - When can the tail be update for the kernel. EVERY entries in the array between
621//         head and tail must be fully filled and shouldn't ever be touched again.
622//
623
624        static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data & ring ) {
625                // Wait for a spot to be available
626                __attribute__((unused)) bool blocked = P(ring.submit);
627                #if !defined(__CFA_NO_STATISTICS__)
628                        __atomic_fetch_add( &ring.submit_q.stats.submit_avg.block, blocked ? 1ul64 : 0ul64, __ATOMIC_RELAXED );
629                #endif
630
631                // Allocate the sqe
632                uint32_t idx = __atomic_fetch_add(&ring.submit_q.alloc, 1ul32, __ATOMIC_SEQ_CST);
633
634                // Validate that we didn't overflow anything
635                // Check that nothing overflowed
636                /* paranoid */ verify( true );
637
638                // Check that it goes head -> tail -> alloc and never head -> alloc -> tail
639                /* paranoid */ verify( true );
640
641                // Return the sqe
642                return [&ring.submit_q.sqes[ idx & (*ring.submit_q.mask)], idx];
643        }
644
645        static inline void __submit( struct __io_data & ring, uint32_t idx ) {
646                // get mutual exclusion
647                lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
648
649                // Append to the list of ready entries
650                uint32_t * tail = ring.submit_q.tail;
651                const uint32_t mask = *ring.submit_q.mask;
652
653                ring.submit_q.array[ (*tail) & mask ] = idx & mask;
654                __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
655
656                // Submit however, many entries need to be submitted
657                int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
658                if( ret < 0 ) {
659                        switch((int)errno) {
660                        default:
661                                abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
662                        }
663                }
664
665                // update statistics
666                #if !defined(__CFA_NO_STATISTICS__)
667                        ring.submit_q.stats.submit_avg.val += 1;
668                        ring.submit_q.stats.submit_avg.cnt += 1;
669                #endif
670
671                unlock(ring.submit_q.lock);
672                // Make sure that idx was submitted
673                // Be careful to not get false positive if we cycled the entire list or that someone else submitted for us
674                __cfadbg_print_safe( io, "Kernel I/O : Performed io_submit for %p, returned %d\n", active_thread(), ret );
675        }
676
677        static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
678                this.opcode = opcode;
679                #if !defined(IOSQE_ASYNC)
680                        this.flags = 0;
681                #else
682                        this.flags = IOSQE_ASYNC;
683                #endif
684                this.ioprio = 0;
685                this.fd = fd;
686                this.off = 0;
687                this.addr = 0;
688                this.len = 0;
689                this.rw_flags = 0;
690                this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0;
691        }
692
693        static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
694                (this){ opcode, fd };
695                this.off = off;
696                this.addr = (uint64_t)addr;
697                this.len = len;
698        }
699
700
701//=============================================================================================
702// I/O Interface
703//=============================================================================================
704
705        #define __submit_prelude \
706                struct __io_data & ring = *active_cluster()->io; \
707                struct io_uring_sqe * sqe; \
708                uint32_t idx; \
709                [sqe, idx] = __submit_alloc( ring );
710
711        #define __submit_wait \
712                io_user_data data = { 0, active_thread() }; \
713                /*__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );*/ \
714                sqe->user_data = (uint64_t)&data; \
715                __submit( ring, idx ); \
716                park( __cfaabi_dbg_ctx ); \
717                return data.result;
718#endif
719
720// Some forward declarations
721extern "C" {
722        #include <unistd.h>
723        #include <sys/types.h>
724        #include <sys/socket.h>
725        #include <sys/syscall.h>
726        struct iovec;
727        extern ssize_t preadv2 (int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
728        extern ssize_t pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
729
730        extern int fsync(int fd);
731        extern int sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags);
732
733        struct msghdr;
734        struct sockaddr;
735        extern ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags);
736        extern ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);
737        extern ssize_t send(int sockfd, const void *buf, size_t len, int flags);
738        extern ssize_t recv(int sockfd, void *buf, size_t len, int flags);
739        extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
740        extern int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
741
742        extern int fallocate(int fd, int mode, uint64_t offset, uint64_t len);
743        extern int posix_fadvise(int fd, uint64_t offset, uint64_t len, int advice);
744        extern int madvise(void *addr, size_t length, int advice);
745
746        extern int openat(int dirfd, const char *pathname, int flags, mode_t mode);
747        extern int close(int fd);
748
749        extern ssize_t read (int fd, void *buf, size_t count);
750}
751
752//-----------------------------------------------------------------------------
753// Asynchronous operations
754ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
755        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READV)
756                return preadv2(fd, iov, iovcnt, offset, flags);
757        #else
758                __submit_prelude
759
760                (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
761
762                __submit_wait
763        #endif
764}
765
766ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
767        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITEV)
768                return pwritev2(fd, iov, iovcnt, offset, flags);
769        #else
770                __submit_prelude
771
772                (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
773
774                __submit_wait
775        #endif
776}
777
778int cfa_fsync(int fd) {
779        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FSYNC)
780                return fsync(fd);
781        #else
782                __submit_prelude
783
784                (*sqe){ IORING_OP_FSYNC, fd };
785
786                __submit_wait
787        #endif
788}
789
790int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) {
791        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SYNC_FILE_RANGE)
792                return sync_file_range(fd, offset, nbytes, flags);
793        #else
794                __submit_prelude
795
796                (*sqe){ IORING_OP_SYNC_FILE_RANGE, fd };
797                sqe->off = offset;
798                sqe->len = nbytes;
799                sqe->sync_range_flags = flags;
800
801                __submit_wait
802        #endif
803}
804
805
806ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) {
807        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SENDMSG)
808                return sendmsg(sockfd, msg, flags);
809        #else
810                __submit_prelude
811
812                (*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 };
813                sqe->msg_flags = flags;
814
815                __submit_wait
816        #endif
817}
818
819ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) {
820        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECVMSG)
821                return recvmsg(sockfd, msg, flags);
822        #else
823                __submit_prelude
824
825                (*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 };
826                sqe->msg_flags = flags;
827
828                __submit_wait
829        #endif
830}
831
832ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) {
833        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SEND)
834                return send( sockfd, buf, len, flags );
835        #else
836                __submit_prelude
837
838                (*sqe){ IORING_OP_SEND, sockfd };
839                sqe->addr = (uint64_t)buf;
840                sqe->len = len;
841                sqe->msg_flags = flags;
842
843                __submit_wait
844        #endif
845}
846
847ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) {
848        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECV)
849                return recv( sockfd, buf, len, flags );
850        #else
851                __submit_prelude
852
853                (*sqe){ IORING_OP_RECV, sockfd };
854                sqe->addr = (uint64_t)buf;
855                sqe->len = len;
856                sqe->msg_flags = flags;
857
858                __submit_wait
859        #endif
860}
861
862int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
863        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_ACCEPT)
864                return accept4( sockfd, addr, addrlen, flags );
865        #else
866                __submit_prelude
867
868                (*sqe){ IORING_OP_ACCEPT, sockfd };
869                sqe->addr = addr;
870                sqe->addr2 = addrlen;
871                sqe->accept_flags = flags;
872
873                __submit_wait
874        #endif
875}
876
877int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
878        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CONNECT)
879                return connect( sockfd, addr, addrlen );
880        #else
881                __submit_prelude
882
883                (*sqe){ IORING_OP_CONNECT, sockfd };
884                sqe->addr = (uint64_t)addr;
885                sqe->off = addrlen;
886
887                __submit_wait
888        #endif
889}
890
891int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) {
892        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FALLOCATE)
893                return fallocate( fd, mode, offset, len );
894        #else
895                __submit_prelude
896
897                (*sqe){ IORING_OP_FALLOCATE, fd };
898                sqe->off = offset;
899                sqe->len = length;
900                sqe->mode = mode;
901
902                __submit_wait
903        #endif
904}
905
906int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) {
907        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FADVISE)
908                return posix_fadvise( fd, offset, len, advice );
909        #else
910                __submit_prelude
911
912                (*sqe){ IORING_OP_FADVISE, fd };
913                sqe->off = (uint64_t)offset;
914                sqe->len = length;
915                sqe->fadvise_advice = advice;
916
917                __submit_wait
918        #endif
919}
920
921int cfa_madvise(void *addr, size_t length, int advice) {
922        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_MADVISE)
923                return madvise( addr, length, advice );
924        #else
925                __submit_prelude
926
927                (*sqe){ IORING_OP_MADVISE, 0 };
928                sqe->addr = (uint64_t)addr;
929                sqe->len = length;
930                sqe->fadvise_advice = advice;
931
932                __submit_wait
933        #endif
934}
935
936int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) {
937        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_OPENAT)
938                return openat( dirfd, pathname, flags, mode );
939        #else
940                __submit_prelude
941
942                (*sqe){ IORING_OP_OPENAT, dirfd };
943                sqe->addr = (uint64_t)pathname;
944                sqe->open_flags = flags;
945                sqe->mode = mode;
946
947                __submit_wait
948        #endif
949}
950
951int cfa_close(int fd) {
952        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CLOSE)
953                return close( fd );
954        #else
955                __submit_prelude
956
957                (*sqe){ IORING_OP_CLOSE, fd };
958
959                __submit_wait
960        #endif
961}
962
963
964ssize_t cfa_read(int fd, void *buf, size_t count) {
965        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READ)
966                return read( fd, buf, count );
967        #else
968                __submit_prelude
969
970                (*sqe){ IORING_OP_READ, fd, buf, count, 0 };
971
972                __submit_wait
973        #endif
974}
975
976ssize_t cfa_write(int fd, void *buf, size_t count) {
977        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITE)
978                return read( fd, buf, count );
979        #else
980                __submit_prelude
981
982                (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
983
984                __submit_wait
985        #endif
986}
987
988//-----------------------------------------------------------------------------
989// Check if a function is asynchronous
990
991// Macro magic to reduce the size of the following switch case
992#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
993#define IS_DEFINED_SECOND(first, second, ...) second
994#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
995#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
996
997bool has_user_level_blocking( fptr_t func ) {
998        #if defined(HAVE_LINUX_IO_URING_H)
999                if( /*func == (fptr_t)preadv2 || */
1000                        func == (fptr_t)cfa_preadv2 )
1001                        #define _CFA_IO_FEATURE_IORING_OP_READV ,
1002                        return IS_DEFINED(IORING_OP_READV);
1003
1004                if( /*func == (fptr_t)pwritev2 || */
1005                        func == (fptr_t)cfa_pwritev2 )
1006                        #define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
1007                        return IS_DEFINED(IORING_OP_WRITEV);
1008
1009                if( /*func == (fptr_t)fsync || */
1010                        func == (fptr_t)cfa_fsync )
1011                        #define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
1012                        return IS_DEFINED(IORING_OP_FSYNC);
1013
1014                if( /*func == (fptr_t)ync_file_range || */
1015                        func == (fptr_t)cfa_sync_file_range )
1016                        #define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
1017                        return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
1018
1019                if( /*func == (fptr_t)sendmsg || */
1020                        func == (fptr_t)cfa_sendmsg )
1021                        #define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
1022                        return IS_DEFINED(IORING_OP_SENDMSG);
1023
1024                if( /*func == (fptr_t)recvmsg || */
1025                        func == (fptr_t)cfa_recvmsg )
1026                        #define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
1027                        return IS_DEFINED(IORING_OP_RECVMSG);
1028
1029                if( /*func == (fptr_t)send || */
1030                        func == (fptr_t)cfa_send )
1031                        #define _CFA_IO_FEATURE_IORING_OP_SEND ,
1032                        return IS_DEFINED(IORING_OP_SEND);
1033
1034                if( /*func == (fptr_t)recv || */
1035                        func == (fptr_t)cfa_recv )
1036                        #define _CFA_IO_FEATURE_IORING_OP_RECV ,
1037                        return IS_DEFINED(IORING_OP_RECV);
1038
1039                if( /*func == (fptr_t)accept4 || */
1040                        func == (fptr_t)cfa_accept4 )
1041                        #define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
1042                        return IS_DEFINED(IORING_OP_ACCEPT);
1043
1044                if( /*func == (fptr_t)connect || */
1045                        func == (fptr_t)cfa_connect )
1046                        #define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
1047                        return IS_DEFINED(IORING_OP_CONNECT);
1048
1049                if( /*func == (fptr_t)fallocate || */
1050                        func == (fptr_t)cfa_fallocate )
1051                        #define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
1052                        return IS_DEFINED(IORING_OP_FALLOCATE);
1053
1054                if( /*func == (fptr_t)posix_fadvise || */
1055                        func == (fptr_t)cfa_fadvise )
1056                        #define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
1057                        return IS_DEFINED(IORING_OP_FADVISE);
1058
1059                if( /*func == (fptr_t)madvise || */
1060                        func == (fptr_t)cfa_madvise )
1061                        #define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
1062                        return IS_DEFINED(IORING_OP_MADVISE);
1063
1064                if( /*func == (fptr_t)openat || */
1065                        func == (fptr_t)cfa_openat )
1066                        #define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
1067                        return IS_DEFINED(IORING_OP_OPENAT);
1068
1069                if( /*func == (fptr_t)close || */
1070                        func == (fptr_t)cfa_close )
1071                        #define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
1072                        return IS_DEFINED(IORING_OP_CLOSE);
1073
1074                if( /*func == (fptr_t)read || */
1075                        func == (fptr_t)cfa_read )
1076                        #define _CFA_IO_FEATURE_IORING_OP_READ ,
1077                        return IS_DEFINED(IORING_OP_READ);
1078
1079                if( /*func == (fptr_t)write || */
1080                        func == (fptr_t)cfa_write )
1081                        #define _CFA_IO_FEATURE_IORING_OP_WRITE ,
1082                        return IS_DEFINED(IORING_OP_WRITE);
1083        #endif
1084
1085        return false;
1086}
Note: See TracBrowser for help on using the repository browser.