source: libcfa/src/concurrency/io.cfa @ 185efe6

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

Some small fixes for the io code

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