source: libcfa/src/concurrency/io.cfa @ 61dd73d

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-astnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 61dd73d was 61dd73d, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Moved io_uring data to io.cfa and create it using dynamic allocation.

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