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

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

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