source: libcfa/src/concurrency/io.cfa@ 9edf835

ADT arm-eh ast-experimental enum forall-pointer-decay jacob/cs343-translation new-ast new-ast-unique-expr pthread-emulation qualifiedEnum
Last change on this file since 9edf835 was 8962722, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

Replace statx call with syscall since it is not on all machines
  • Property mode set to 100644
File size: 22.6 KB
RevLine 
[92976d9]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
[185efe6]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
[92976d9]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
[2d8f7b0]75 uint32_t nentries = entries_per_cluster();
76
77 int fd = syscall(__NR_io_uring_setup, nentries, &params );
[92976d9]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));
[2d8f7b0]84 struct io_uring_sq & sq = this.io.submit_q;
85 struct io_uring_cq & cq = this.io.completion_q;
[92976d9]86
87 // calculate the right ring size
[2d8f7b0]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));
[92976d9]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
[2d8f7b0]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) {
[92976d9]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
[2d8f7b0]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);
[92976d9]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);
[2d8f7b0]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);
[92976d9]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
[2d8f7b0]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;
[92976d9]136
137 // completion queue
[2d8f7b0]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 );
[92976d9]155
156 // Update the global ring info
157 this.io.flags = params.flags;
158 this.io.fd = fd;
159 this.io.done = false;
[2d8f7b0]160 (this.io.submit){ min(*sq.num, *cq.num) };
[92976d9]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
[2d8f7b0]182 munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe));
[92976d9]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;
[185efe6]218 // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
[2d8f7b0]219
[92976d9]220 data->result = cqe.res;
[2d8f7b0]221 __unpark( data->thrd __cfaabi_dbg_ctx2 );
222
223 // Allow new submissions to happen
224 V(ring.submit);
[92976d9]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.
[2d8f7b0]228 __atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED );
[92976d9]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
[2d8f7b0]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);
[185efe6]326 // __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_submit, returned %d\n", ret );
[2d8f7b0]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;
[92976d9]360}
361
362//=============================================================================================
363// I/O Interface
364//=============================================================================================
365 extern "C" {
366 #define __USE_GNU
[2d8f7b0]367 #define _GNU_SOURCE
368 #include <fcntl.h>
[92976d9]369 #include <sys/uio.h>
[2d8f7b0]370 #include <sys/socket.h>
371 #include <sys/stat.h>
[92976d9]372 }
373
[2d8f7b0]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() }; \
[185efe6]382 /*__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );*/ \
[2d8f7b0]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
[92976d9]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
[2d8f7b0]394 __submit_prelude
395
396 (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
397
398 __submit_wait
[92976d9]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
[2d8f7b0]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
[92976d9]423 #endif
424 }
425
[2d8f7b0]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)
[8962722]605 //return statx( dirfd, pathname, flags, mask, statxbuf );
606 return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf );
[2d8f7b0]607 #else
608 __submit_prelude
609
610 (*sqe){ IORING_OP_STATX, dirfd };
611 sqe->addr = (uint64_t)pathname;
612 sqe->statx_flags = flags;
613 sqe->len = mask;
614 sqe->off = (uint64_t)statxbuf;
615
616 __submit_wait
617 #endif
618 }
619
620
621 ssize_t async_read(int fd, void *buf, size_t count) {
622 #if !defined(IORING_OP_READ)
623 return read( fd, buf, count );
624 #else
625 __submit_prelude
626
627 (*sqe){ IORING_OP_READ, fd, buf, count, 0 };
628
629 __submit_wait
630 #endif
631 }
632
633 ssize_t async_write(int fd, void *buf, size_t count) {
634 #if !defined(IORING_OP_WRITE)
635 return read( fd, buf, count );
636 #else
637 __submit_prelude
638
639 (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
640
641 __submit_wait
642 #endif
643 }
644
645//-----------------------------------------------------------------------------
646// Check if a function is asynchronous
647
648// Macro magic to reduce the size of the following switch case
649 #define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
650 #define IS_DEFINED_SECOND(first, second, ...) second
651 #define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
652 #define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
653
654 bool is_async( fptr_t func ) {
655
656 if( /*func == (fptr_t)preadv2 || */
657 func == (fptr_t)async_preadv2 )
658 #define _CFA_IO_FEATURE_IORING_OP_READV ,
659 return IS_DEFINED(IORING_OP_READV);
660
661 if( /*func == (fptr_t)pwritev2 || */
662 func == (fptr_t)async_pwritev2 )
663 #define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
664 return IS_DEFINED(IORING_OP_WRITEV);
665
666 if( /*func == (fptr_t)fsync || */
667 func == (fptr_t)async_fsync )
668 #define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
669 return IS_DEFINED(IORING_OP_FSYNC);
670
671 if( /*func == (fptr_t)ync_file_range || */
672 func == (fptr_t)async_sync_file_range )
673 #define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
674 return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
675
676 if( /*func == (fptr_t)sendmsg || */
677 func == (fptr_t)async_sendmsg )
678 #define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
679 return IS_DEFINED(IORING_OP_SENDMSG);
680
681 if( /*func == (fptr_t)recvmsg || */
682 func == (fptr_t)async_recvmsg )
683 #define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
684 return IS_DEFINED(IORING_OP_RECVMSG);
685
686 if( /*func == (fptr_t)send || */
687 func == (fptr_t)async_send )
688 #define _CFA_IO_FEATURE_IORING_OP_SEND ,
689 return IS_DEFINED(IORING_OP_SEND);
690
691 if( /*func == (fptr_t)recv || */
692 func == (fptr_t)async_recv )
693 #define _CFA_IO_FEATURE_IORING_OP_RECV ,
694 return IS_DEFINED(IORING_OP_RECV);
695
696 if( /*func == (fptr_t)accept4 || */
697 func == (fptr_t)async_accept4 )
698 #define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
699 return IS_DEFINED(IORING_OP_ACCEPT);
700
701 if( /*func == (fptr_t)connect || */
702 func == (fptr_t)async_connect )
703 #define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
704 return IS_DEFINED(IORING_OP_CONNECT);
705
706 if( /*func == (fptr_t)fallocate || */
707 func == (fptr_t)async_fallocate )
708 #define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
709 return IS_DEFINED(IORING_OP_FALLOCATE);
710
711 if( /*func == (fptr_t)fadvise || */
712 func == (fptr_t)async_fadvise )
713 #define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
714 return IS_DEFINED(IORING_OP_FADVISE);
715
716 if( /*func == (fptr_t)madvise || */
717 func == (fptr_t)async_madvise )
718 #define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
719 return IS_DEFINED(IORING_OP_MADVISE);
720
721 if( /*func == (fptr_t)openat || */
722 func == (fptr_t)async_openat )
723 #define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
724 return IS_DEFINED(IORING_OP_OPENAT);
725
726 if( /*func == (fptr_t)close || */
727 func == (fptr_t)async_close )
728 #define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
729 return IS_DEFINED(IORING_OP_CLOSE);
730
731 if( /*func == (fptr_t)statx || */
732 func == (fptr_t)async_statx )
733 #define _CFA_IO_FEATURE_IORING_OP_STATX ,
734 return IS_DEFINED(IORING_OP_STATX);
735
736 if( /*func == (fptr_t)read || */
737 func == (fptr_t)async_read )
738 #define _CFA_IO_FEATURE_IORING_OP_READ ,
739 return IS_DEFINED(IORING_OP_READ);
740
741 if( /*func == (fptr_t)write || */
742 func == (fptr_t)async_write )
743 #define _CFA_IO_FEATURE_IORING_OP_WRITE ,
744 return IS_DEFINED(IORING_OP_WRITE);
745
746 return false;
[92976d9]747 }
748
749#endif
Note: See TracBrowser for help on using the repository browser.