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

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

some clean-up in io.cfa

  • Property mode set to 100644
File size: 23.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#include "kernel.hfa"
17
18#if !defined(HAVE_LINUX_IO_URING_H)
19        void __kernel_io_startup( cluster & this ) {
20                // Nothing to do without io_uring
21        }
22
23        void __kernel_io_shutdown( cluster & this ) {
24                // Nothing to do without io_uring
25        }
26
27#else
28        extern "C" {
29                #define _GNU_SOURCE         /* See feature_test_macros(7) */
30                #include <errno.h>
31                #include <stdint.h>
32                #include <string.h>
33                #include <unistd.h>
34                #include <sys/mman.h>
35                #include <sys/syscall.h>
36
37                #include <linux/io_uring.h>
38        }
39
40        #include "bits/signal.hfa"
41        #include "kernel_private.hfa"
42        #include "thread.hfa"
43
44        uint32_t entries_per_cluster() {
45                return 256;
46        }
47
48        static void * __io_poller( void * arg );
49
50       // Weirdly, some systems that do support io_uring don't actually define these
51       #ifdef __alpha__
52       /*
53       * alpha is the only exception, all other architectures
54       * have common numbers for new system calls.
55       */
56       # ifndef __NR_io_uring_setup
57       #  define __NR_io_uring_setup           535
58       # endif
59       # ifndef __NR_io_uring_enter
60       #  define __NR_io_uring_enter           536
61       # endif
62       # ifndef __NR_io_uring_register
63       #  define __NR_io_uring_register        537
64       # endif
65       #else /* !__alpha__ */
66       # ifndef __NR_io_uring_setup
67       #  define __NR_io_uring_setup           425
68       # endif
69       # ifndef __NR_io_uring_enter
70       #  define __NR_io_uring_enter           426
71       # endif
72       # ifndef __NR_io_uring_register
73       #  define __NR_io_uring_register        427
74       # endif
75       #endif
76
77
78//=============================================================================================
79// I/O Startup / Shutdown logic
80//=============================================================================================
81        void __kernel_io_startup( cluster & this ) {
82                // Step 1 : call to setup
83                struct io_uring_params params;
84                memset(&params, 0, sizeof(params));
85
86                uint32_t nentries = entries_per_cluster();
87
88                int fd = syscall(__NR_io_uring_setup, nentries, &params );
89                if(fd < 0) {
90                        abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
91                }
92
93                // Step 2 : mmap result
94                memset(&this.io, 0, sizeof(struct io_ring));
95                struct io_uring_sq & sq = this.io.submit_q;
96                struct io_uring_cq & cq = this.io.completion_q;
97
98                // calculate the right ring size
99                sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned)           );
100                cq.ring_sz = params.cq_off.cqes  + (params.cq_entries * sizeof(struct io_uring_cqe));
101
102                // Requires features
103                // // adjust the size according to the parameters
104                // if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
105                //      cq->ring_sz = sq->ring_sz = max(cq->ring_sz, sq->ring_sz);
106                // }
107
108                // mmap the Submit Queue into existence
109                sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING);
110                if (sq.ring_ptr == (void*)MAP_FAILED) {
111                        abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
112                }
113
114                // mmap the Completion Queue into existence (may or may not be needed)
115                // Requires features
116                // if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
117                //      cq->ring_ptr = sq->ring_ptr;
118                // }
119                // else {
120                        // We need multiple call to MMAP
121                        cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING);
122                        if (cq.ring_ptr == (void*)MAP_FAILED) {
123                                munmap(sq.ring_ptr, sq.ring_sz);
124                                abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
125                        }
126                // }
127
128                // mmap the submit queue entries
129                size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
130                sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES);
131                if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
132                        munmap(sq.ring_ptr, sq.ring_sz);
133                        if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
134                        abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
135                }
136
137                // Get the pointers from the kernel to fill the structure
138                // submit queue
139                sq.head    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
140                sq.tail    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
141                sq.mask    = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
142                sq.num     = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
143                sq.flags   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
144                sq.dropped = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
145                sq.array   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
146                sq.alloc = *sq.tail;
147
148                // completion queue
149                cq.head     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
150                cq.tail     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
151                cq.mask     = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
152                cq.num      = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
153                cq.overflow = (         uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
154                cq.cqes   = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
155
156                // some paranoid checks
157                /* 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  );
158                /* paranoid */ verifyf( (*cq.num)  >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
159                /* paranoid */ verifyf( (*cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
160                /* paranoid */ verifyf( (*cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
161
162                /* 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 );
163                /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
164                /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
165                /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
166
167                // Update the global ring info
168                this.io.flags = params.flags;
169                this.io.fd    = fd;
170                this.io.done  = false;
171                (this.io.submit){ min(*sq.num, *cq.num) };
172
173                // Create the poller thread
174                this.io.stack = __create_pthread( &this.io.poller, __io_poller, &this );
175        }
176
177        void __kernel_io_shutdown( cluster & this ) {
178                // Stop the IO Poller
179                // Notify the poller thread of the shutdown
180                __atomic_store_n(&this.io.done, true, __ATOMIC_SEQ_CST);
181                sigval val = { 1 };
182                pthread_sigqueue( this.io.poller, SIGUSR1, val );
183
184                // Wait for the poller thread to finish
185                pthread_join( this.io.poller, 0p );
186                free( this.io.stack );
187
188                // Shutdown the io rings
189                struct io_uring_sq & sq = this.io.submit_q;
190                struct io_uring_cq & cq = this.io.completion_q;
191
192                // unmap the submit queue entries
193                munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe));
194
195                // unmap the Submit Queue ring
196                munmap(sq.ring_ptr, sq.ring_sz);
197
198                // unmap the Completion Queue ring, if it is different
199                if (cq.ring_ptr != sq.ring_ptr) {
200                        munmap(cq.ring_ptr, cq.ring_sz);
201                }
202
203                // close the file descriptor
204                close(this.io.fd);
205        }
206
207//=============================================================================================
208// I/O Polling
209//=============================================================================================
210        struct io_user_data {
211                int32_t result;
212                $thread * thrd;
213        };
214
215        // Process a single completion message from the io_uring
216        // This is NOT thread-safe
217        static bool __io_process(struct io_ring & ring) {
218                unsigned head = *ring.completion_q.head;
219                unsigned tail = __atomic_load_n(ring.completion_q.tail, __ATOMIC_ACQUIRE);
220
221                if (head == tail) return false;
222
223                unsigned idx = head & (*ring.completion_q.mask);
224                struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
225
226                /* paranoid */ verify(&cqe);
227
228                struct io_user_data * data = (struct io_user_data *)cqe.user_data;
229                // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
230
231                data->result = cqe.res;
232                __unpark( data->thrd __cfaabi_dbg_ctx2 );
233
234                // Allow new submissions to happen
235                V(ring.submit);
236
237                // Mark to the kernel that the cqe has been seen
238                // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
239                __atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED );
240
241                return true;
242        }
243
244        static void * __io_poller( void * arg ) {
245                cluster * cltr = (cluster *)arg;
246                struct io_ring & ring = cltr->io;
247
248                sigset_t mask;
249                sigfillset(&mask);
250                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
251                        abort( "KERNEL ERROR: IO_URING - pthread_sigmask" );
252                }
253
254                sigdelset( &mask, SIGUSR1 );
255
256                verify( (*ring.submit_q.head) == (*ring.submit_q.tail) );
257                verify( (*ring.completion_q.head) == (*ring.completion_q.tail) );
258
259                LOOP: while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
260                        int ret = syscall( __NR_io_uring_enter, ring.fd, 0, 1, IORING_ENTER_GETEVENTS, &mask, _NSIG / 8);
261                        if( ret < 0 ) {
262                                switch((int)errno) {
263                                case EAGAIN:
264                                case EINTR:
265                                        continue LOOP;
266                                default:
267                                        abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
268                                }
269                        }
270
271                        // Drain the queue
272                        while(__io_process(ring)) {}
273                }
274
275                return 0p;
276        }
277
278//=============================================================================================
279// I/O Submissions
280//=============================================================================================
281
282// Submition steps :
283// 1 - We need to make sure we don't overflow any of the buffer, P(ring.submit) to make sure
284//     entries are available. The semaphore make sure that there is no more operations in
285//     progress then the number of entries in the buffer. This probably limits concurrency
286//     more than necessary since submitted but not completed operations don't need any
287//     entries in user space. However, I don't know what happens if we overflow the buffers
288//     because too many requests completed at once. This is a safe approach in all cases.
289//     Furthermore, with hundreds of entries, this may be okay.
290//
291// 2 - Allocate a queue entry. The ring already has memory for all entries but only the ones
292//     listed in sq.array are visible by the kernel. For those not listed, the kernel does not
293//     offer any assurance that an entry is not being filled by multiple flags. Therefore, we
294//     need to write an allocator that allows allocating concurrently.
295//
296// 3 - Actually fill the submit entry, this is the only simple and straightforward step.
297//
298// 4 - Append the entry index to the array and adjust the tail accordingly. This operation
299//     needs to arrive to two concensus at the same time:
300//     A - The order in which entries are listed in the array: no two threads must pick the
301//         same index for their entries
302//     B - When can the tail be update for the kernel. EVERY entries in the array between
303//         head and tail must be fully filled and shouldn't ever be touched again.
304//
305
306        static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct io_ring & ring ) {
307                // Wait for a spot to be available
308                P(ring.submit);
309
310                // Allocate the sqe
311                uint32_t idx = __atomic_fetch_add(&ring.submit_q.alloc, 1ul32, __ATOMIC_SEQ_CST);
312
313                // Validate that we didn't overflow anything
314                // Check that nothing overflowed
315                /* paranoid */ verify( true );
316
317                // Check that it goes head -> tail -> alloc and never head -> alloc -> tail
318                /* paranoid */ verify( true );
319
320                // Return the sqe
321                return [&ring.submit_q.sqes[ idx & (*ring.submit_q.mask)], idx];
322        }
323
324        static inline void __submit( struct io_ring & ring, uint32_t idx ) {
325                // get mutual exclusion
326                lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
327
328                // Append to the list of ready entries
329                uint32_t * tail = ring.submit_q.tail;
330                const uint32_t mask = *ring.submit_q.mask;
331
332                ring.submit_q.array[ (*tail) & mask ] = idx & mask;
333                __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
334
335                // Submit however, many entries need to be submitted
336                int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
337                // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_submit, returned %d\n", ret );
338                if( ret < 0 ) {
339                        switch((int)errno) {
340                        default:
341                                abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
342                        }
343                }
344
345                unlock(ring.submit_q.lock);
346                // Make sure that idx was submitted
347                // Be careful to not get false positive if we cycled the entire list or that someone else submitted for us
348        }
349
350        static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
351                this.opcode = opcode;
352                #if !defined(IOSQE_ASYNC)
353                        this.flags = 0;
354                #else
355                        this.flags = IOSQE_ASYNC;
356                #endif
357                this.ioprio = 0;
358                this.fd = fd;
359                this.off = 0;
360                this.addr = 0;
361                this.len = 0;
362                this.rw_flags = 0;
363                this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0;
364        }
365
366        static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
367                (this){ opcode, fd };
368                this.off = off;
369                this.addr = (uint64_t)addr;
370                this.len = len;
371        }
372#endif
373
374//=============================================================================================
375// I/O Interface
376//=============================================================================================
377extern "C" {
378        #define __USE_GNU
379        #define _GNU_SOURCE
380        #include <fcntl.h>
381        #include <sys/uio.h>
382        #include <sys/socket.h>
383        #include <sys/stat.h>
384}
385
386#if defined(HAVE_LINUX_IO_URING_H)
387        #define __submit_prelude \
388                struct io_ring & ring = active_cluster()->io; \
389                struct io_uring_sqe * sqe; \
390                uint32_t idx; \
391                [sqe, idx] = __submit_alloc( ring );
392
393        #define __submit_wait \
394                io_user_data data = { 0, active_thread() }; \
395                /*__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );*/ \
396                sqe->user_data = (uint64_t)&data; \
397                __submit( ring, idx ); \
398                park( __cfaabi_dbg_ctx ); \
399                return data.result;
400#endif
401
402//-----------------------------------------------------------------------------
403// Asynchronous operations
404ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
405        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READV)
406                return preadv2(fd, iov, iovcnt, offset, flags);
407        #else
408                __submit_prelude
409
410                (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
411
412                __submit_wait
413        #endif
414}
415
416ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
417        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITEV)
418                return pwritev2(fd, iov, iovcnt, offset, flags);
419        #else
420                __submit_prelude
421
422                (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
423
424                __submit_wait
425        #endif
426}
427
428int cfa_fsync(int fd) {
429        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FSYNC)
430                return fsync(fd);
431        #else
432                __submit_prelude
433
434                (*sqe){ IORING_OP_FSYNC, fd };
435
436                __submit_wait
437        #endif
438}
439
440int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) {
441        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SYNC_FILE_RANGE)
442                return sync_file_range(fd, offset, nbytes, flags);
443        #else
444                __submit_prelude
445
446                (*sqe){ IORING_OP_SYNC_FILE_RANGE, fd };
447                sqe->off = offset;
448                sqe->len = nbytes;
449                sqe->sync_range_flags = flags;
450
451                __submit_wait
452        #endif
453}
454
455
456ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) {
457        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SENDMSG)
458                return recv(sockfd, msg, flags);
459        #else
460                __submit_prelude
461
462                (*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 };
463                sqe->msg_flags = flags;
464
465                __submit_wait
466        #endif
467}
468
469ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) {
470        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECVMSG)
471                return recv(sockfd, msg, flags);
472        #else
473                __submit_prelude
474
475                (*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 };
476                sqe->msg_flags = flags;
477
478                __submit_wait
479        #endif
480}
481
482ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) {
483        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SEND)
484                return send( sockfd, buf, len, flags );
485        #else
486                __submit_prelude
487
488                (*sqe){ IORING_OP_SEND, sockfd };
489                sqe->addr = (uint64_t)buf;
490                sqe->len = len;
491                sqe->msg_flags = flags;
492
493                __submit_wait
494        #endif
495}
496
497ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) {
498        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECV)
499                return recv( sockfd, buf, len, flags );
500        #else
501                __submit_prelude
502
503                (*sqe){ IORING_OP_RECV, sockfd };
504                sqe->addr = (uint64_t)buf;
505                sqe->len = len;
506                sqe->msg_flags = flags;
507
508                __submit_wait
509        #endif
510}
511
512int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
513        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_ACCEPT)
514                __SOCKADDR_ARG _addr;
515                _addr.__sockaddr__ = addr;
516                return accept4( sockfd, _addr, addrlen, flags );
517        #else
518                __submit_prelude
519
520                (*sqe){ IORING_OP_ACCEPT, sockfd };
521                sqe->addr = addr;
522                sqe->addr2 = addrlen;
523                sqe->accept_flags = flags;
524
525                __submit_wait
526        #endif
527}
528
529int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
530        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CONNECT)
531                __CONST_SOCKADDR_ARG _addr;
532                _addr.__sockaddr__ = addr;
533                return connect( sockfd, _addr, addrlen );
534        #else
535                __submit_prelude
536
537                (*sqe){ IORING_OP_CONNECT, sockfd };
538                sqe->addr = (uint64_t)addr;
539                sqe->off = addrlen;
540
541                __submit_wait
542        #endif
543}
544
545int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) {
546        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FALLOCATE)
547                return fallocate( fd, mode, offset, len );
548        #else
549                __submit_prelude
550
551                (*sqe){ IORING_OP_FALLOCATE, fd };
552                sqe->off = offset;
553                sqe->len = length;
554                sqe->mode = mode;
555
556                __submit_wait
557        #endif
558}
559
560int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) {
561        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FADVISE)
562                return posix_fadvise( fd, offset, len, advice );
563        #else
564                __submit_prelude
565
566                (*sqe){ IORING_OP_FADVISE, fd };
567                sqe->off = (uint64_t)offset;
568                sqe->len = length;
569                sqe->fadvise_advice = advice;
570
571                __submit_wait
572        #endif
573}
574
575int cfa_madvise(void *addr, size_t length, int advice) {
576        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_MADVISE)
577                return madvise( addr, length, advice );
578        #else
579                __submit_prelude
580
581                (*sqe){ IORING_OP_MADVISE, 0 };
582                sqe->addr = (uint64_t)addr;
583                sqe->len = length;
584                sqe->fadvise_advice = advice;
585
586                __submit_wait
587        #endif
588}
589
590int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) {
591        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_OPENAT)
592                return openat( dirfd, pathname, flags, mode );
593        #else
594                __submit_prelude
595
596                (*sqe){ IORING_OP_OPENAT, dirfd };
597                sqe->addr = (uint64_t)pathname;
598                sqe->open_flags = flags;
599                sqe->mode = mode;
600
601                __submit_wait
602        #endif
603}
604
605int cfa_close(int fd) {
606        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CLOSE)
607                return close( fd );
608        #else
609                __submit_prelude
610
611                (*sqe){ IORING_OP_CLOSE, fd };
612
613                __submit_wait
614        #endif
615}
616
617int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf) {
618        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_STATX)
619                //return statx( dirfd, pathname, flags, mask, statxbuf );
620                return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf );
621        #else
622                __submit_prelude
623
624                (*sqe){ IORING_OP_STATX, dirfd };
625                sqe->addr = (uint64_t)pathname;
626                sqe->statx_flags = flags;
627                sqe->len = mask;
628                sqe->off = (uint64_t)statxbuf;
629
630                __submit_wait
631        #endif
632}
633
634
635ssize_t cfa_read(int fd, void *buf, size_t count) {
636        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READ)
637                return read( fd, buf, count );
638        #else
639                __submit_prelude
640
641                (*sqe){ IORING_OP_READ, fd, buf, count, 0 };
642
643                __submit_wait
644        #endif
645}
646
647ssize_t cfa_write(int fd, void *buf, size_t count) {
648        #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITE)
649                return read( fd, buf, count );
650        #else
651                __submit_prelude
652
653                (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
654
655                __submit_wait
656        #endif
657}
658
659//-----------------------------------------------------------------------------
660// Check if a function is asynchronous
661
662// Macro magic to reduce the size of the following switch case
663#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
664#define IS_DEFINED_SECOND(first, second, ...) second
665#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
666#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
667
668bool has_user_level_blocking( fptr_t func ) {
669        #if defined(HAVE_LINUX_IO_URING_H)
670                if( /*func == (fptr_t)preadv2 || */
671                        func == (fptr_t)cfa_preadv2 )
672                        #define _CFA_IO_FEATURE_IORING_OP_READV ,
673                        return IS_DEFINED(IORING_OP_READV);
674
675                if( /*func == (fptr_t)pwritev2 || */
676                        func == (fptr_t)cfa_pwritev2 )
677                        #define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
678                        return IS_DEFINED(IORING_OP_WRITEV);
679
680                if( /*func == (fptr_t)fsync || */
681                        func == (fptr_t)cfa_fsync )
682                        #define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
683                        return IS_DEFINED(IORING_OP_FSYNC);
684
685                if( /*func == (fptr_t)ync_file_range || */
686                        func == (fptr_t)cfa_sync_file_range )
687                        #define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
688                        return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
689
690                if( /*func == (fptr_t)sendmsg || */
691                        func == (fptr_t)cfa_sendmsg )
692                        #define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
693                        return IS_DEFINED(IORING_OP_SENDMSG);
694
695                if( /*func == (fptr_t)recvmsg || */
696                        func == (fptr_t)cfa_recvmsg )
697                        #define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
698                        return IS_DEFINED(IORING_OP_RECVMSG);
699
700                if( /*func == (fptr_t)send || */
701                        func == (fptr_t)cfa_send )
702                        #define _CFA_IO_FEATURE_IORING_OP_SEND ,
703                        return IS_DEFINED(IORING_OP_SEND);
704
705                if( /*func == (fptr_t)recv || */
706                        func == (fptr_t)cfa_recv )
707                        #define _CFA_IO_FEATURE_IORING_OP_RECV ,
708                        return IS_DEFINED(IORING_OP_RECV);
709
710                if( /*func == (fptr_t)accept4 || */
711                        func == (fptr_t)cfa_accept4 )
712                        #define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
713                        return IS_DEFINED(IORING_OP_ACCEPT);
714
715                if( /*func == (fptr_t)connect || */
716                        func == (fptr_t)cfa_connect )
717                        #define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
718                        return IS_DEFINED(IORING_OP_CONNECT);
719
720                if( /*func == (fptr_t)fallocate || */
721                        func == (fptr_t)cfa_fallocate )
722                        #define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
723                        return IS_DEFINED(IORING_OP_FALLOCATE);
724
725                if( /*func == (fptr_t)fadvise || */
726                        func == (fptr_t)cfa_fadvise )
727                        #define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
728                        return IS_DEFINED(IORING_OP_FADVISE);
729
730                if( /*func == (fptr_t)madvise || */
731                        func == (fptr_t)cfa_madvise )
732                        #define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
733                        return IS_DEFINED(IORING_OP_MADVISE);
734
735                if( /*func == (fptr_t)openat || */
736                        func == (fptr_t)cfa_openat )
737                        #define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
738                        return IS_DEFINED(IORING_OP_OPENAT);
739
740                if( /*func == (fptr_t)close || */
741                        func == (fptr_t)cfa_close )
742                        #define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
743                        return IS_DEFINED(IORING_OP_CLOSE);
744
745                if( /*func == (fptr_t)statx || */
746                        func == (fptr_t)cfa_statx )
747                        #define _CFA_IO_FEATURE_IORING_OP_STATX ,
748                        return IS_DEFINED(IORING_OP_STATX);
749
750                if( /*func == (fptr_t)read || */
751                        func == (fptr_t)cfa_read )
752                        #define _CFA_IO_FEATURE_IORING_OP_READ ,
753                        return IS_DEFINED(IORING_OP_READ);
754
755                if( /*func == (fptr_t)write || */
756                        func == (fptr_t)cfa_write )
757                        #define _CFA_IO_FEATURE_IORING_OP_WRITE ,
758                        return IS_DEFINED(IORING_OP_WRITE);
759        #endif
760
761        return false;
762}
Note: See TracBrowser for help on using the repository browser.