source: libcfa/src/concurrency/io.cfa@ 70ac8d0

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 70ac8d0 was 5c581cc, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

Fixed spinning of slow poller and added random offset to submit to avoid clashing

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