source: libcfa/src/concurrency/io.cfa@ 6c12fd28

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 6c12fd28 was dd4e2d7, checked in by Thierry Delisle <tdelisle@…>, 6 years ago

Added option to change the length of the array of ready sqe
  • 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 &, unsigned, 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, unsigned 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 = max(io_flags >> CFA_CLUSTER_IO_BUFFLEN_OFFSET, 8);
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 __cfadbg_print_safe( io, "Kernel I/O : Added %u to ready for %p\n", idx, active_thread() );
757 }
758 else {
759 // get mutual exclusion
760 lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
761
762 // Append to the list of ready entries
763
764 /* paranoid */ verify( idx <= mask );
765
766 ring.submit_q.array[ (*tail) & mask ] = idx & mask;
767 __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
768
769 // Submit however, many entries need to be submitted
770 int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
771 if( ret < 0 ) {
772 switch((int)errno) {
773 default:
774 abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
775 }
776 }
777
778 // update statistics
779 #if !defined(__CFA_NO_STATISTICS__)
780 ring.submit_q.stats.submit_avg.val += 1;
781 ring.submit_q.stats.submit_avg.cnt += 1;
782 #endif
783
784 unlock(ring.submit_q.lock);
785
786 __cfadbg_print_safe( io, "Kernel I/O : Performed io_submit for %p, returned %d\n", active_thread(), ret );
787 }
788 }
789
790 static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
791 this.opcode = opcode;
792 #if !defined(IOSQE_ASYNC)
793 this.flags = 0;
794 #else
795 this.flags = IOSQE_ASYNC;
796 #endif
797 this.ioprio = 0;
798 this.fd = fd;
799 this.off = 0;
800 this.addr = 0;
801 this.len = 0;
802 this.rw_flags = 0;
803 this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0;
804 }
805
806 static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
807 (this){ opcode, fd };
808 this.off = off;
809 this.addr = (uint64_t)addr;
810 this.len = len;
811 }
812
813
814//=============================================================================================
815// I/O Interface
816//=============================================================================================
817
818 #define __submit_prelude \
819 struct __io_data & ring = *active_cluster()->io; \
820 struct io_uring_sqe * sqe; \
821 uint32_t idx; \
822 [sqe, idx] = __submit_alloc( ring );
823
824 #define __submit_wait \
825 io_user_data data = { 0, active_thread() }; \
826 /*__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );*/ \
827 sqe->user_data = (uint64_t)&data; \
828 __submit( ring, idx ); \
829 park( __cfaabi_dbg_ctx ); \
830 return data.result;
831#endif
832
833// Some forward declarations
834extern "C" {
835 #include <unistd.h>
836 #include <sys/types.h>
837 #include <sys/socket.h>
838 #include <sys/syscall.h>
839
840#if defined(HAVE_PREADV2)
841 struct iovec;
842 extern ssize_t preadv2 (int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
843#endif
844#if defined(HAVE_PWRITEV2)
845 struct iovec;
846 extern ssize_t pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
847#endif
848
849 extern int fsync(int fd);
850 extern int sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags);
851
852 struct msghdr;
853 struct sockaddr;
854 extern ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags);
855 extern ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);
856 extern ssize_t send(int sockfd, const void *buf, size_t len, int flags);
857 extern ssize_t recv(int sockfd, void *buf, size_t len, int flags);
858 extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
859 extern int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
860
861 extern int fallocate(int fd, int mode, uint64_t offset, uint64_t len);
862 extern int posix_fadvise(int fd, uint64_t offset, uint64_t len, int advice);
863 extern int madvise(void *addr, size_t length, int advice);
864
865 extern int openat(int dirfd, const char *pathname, int flags, mode_t mode);
866 extern int close(int fd);
867
868 extern ssize_t read (int fd, void *buf, size_t count);
869}
870
871//-----------------------------------------------------------------------------
872// Asynchronous operations
873#if defined(HAVE_PREADV2)
874 ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
875 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READV)
876 return preadv2(fd, iov, iovcnt, offset, flags);
877 #else
878 __submit_prelude
879
880 (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
881
882 __submit_wait
883 #endif
884 }
885#endif
886
887#if defined(HAVE_PWRITEV2)
888 ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
889 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITEV)
890 return pwritev2(fd, iov, iovcnt, offset, flags);
891 #else
892 __submit_prelude
893
894 (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
895
896 __submit_wait
897 #endif
898 }
899#endif
900
901int cfa_fsync(int fd) {
902 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FSYNC)
903 return fsync(fd);
904 #else
905 __submit_prelude
906
907 (*sqe){ IORING_OP_FSYNC, fd };
908
909 __submit_wait
910 #endif
911}
912
913int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) {
914 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SYNC_FILE_RANGE)
915 return sync_file_range(fd, offset, nbytes, flags);
916 #else
917 __submit_prelude
918
919 (*sqe){ IORING_OP_SYNC_FILE_RANGE, fd };
920 sqe->off = offset;
921 sqe->len = nbytes;
922 sqe->sync_range_flags = flags;
923
924 __submit_wait
925 #endif
926}
927
928
929ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) {
930 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SENDMSG)
931 return sendmsg(sockfd, msg, flags);
932 #else
933 __submit_prelude
934
935 (*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 };
936 sqe->msg_flags = flags;
937
938 __submit_wait
939 #endif
940}
941
942ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) {
943 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECVMSG)
944 return recvmsg(sockfd, msg, flags);
945 #else
946 __submit_prelude
947
948 (*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 };
949 sqe->msg_flags = flags;
950
951 __submit_wait
952 #endif
953}
954
955ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) {
956 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SEND)
957 return send( sockfd, buf, len, flags );
958 #else
959 __submit_prelude
960
961 (*sqe){ IORING_OP_SEND, sockfd };
962 sqe->addr = (uint64_t)buf;
963 sqe->len = len;
964 sqe->msg_flags = flags;
965
966 __submit_wait
967 #endif
968}
969
970ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) {
971 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECV)
972 return recv( sockfd, buf, len, flags );
973 #else
974 __submit_prelude
975
976 (*sqe){ IORING_OP_RECV, sockfd };
977 sqe->addr = (uint64_t)buf;
978 sqe->len = len;
979 sqe->msg_flags = flags;
980
981 __submit_wait
982 #endif
983}
984
985int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
986 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_ACCEPT)
987 return accept4( sockfd, addr, addrlen, flags );
988 #else
989 __submit_prelude
990
991 (*sqe){ IORING_OP_ACCEPT, sockfd };
992 sqe->addr = addr;
993 sqe->addr2 = addrlen;
994 sqe->accept_flags = flags;
995
996 __submit_wait
997 #endif
998}
999
1000int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
1001 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CONNECT)
1002 return connect( sockfd, addr, addrlen );
1003 #else
1004 __submit_prelude
1005
1006 (*sqe){ IORING_OP_CONNECT, sockfd };
1007 sqe->addr = (uint64_t)addr;
1008 sqe->off = addrlen;
1009
1010 __submit_wait
1011 #endif
1012}
1013
1014int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) {
1015 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FALLOCATE)
1016 return fallocate( fd, mode, offset, len );
1017 #else
1018 __submit_prelude
1019
1020 (*sqe){ IORING_OP_FALLOCATE, fd };
1021 sqe->off = offset;
1022 sqe->len = length;
1023 sqe->mode = mode;
1024
1025 __submit_wait
1026 #endif
1027}
1028
1029int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) {
1030 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FADVISE)
1031 return posix_fadvise( fd, offset, len, advice );
1032 #else
1033 __submit_prelude
1034
1035 (*sqe){ IORING_OP_FADVISE, fd };
1036 sqe->off = (uint64_t)offset;
1037 sqe->len = length;
1038 sqe->fadvise_advice = advice;
1039
1040 __submit_wait
1041 #endif
1042}
1043
1044int cfa_madvise(void *addr, size_t length, int advice) {
1045 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_MADVISE)
1046 return madvise( addr, length, advice );
1047 #else
1048 __submit_prelude
1049
1050 (*sqe){ IORING_OP_MADVISE, 0 };
1051 sqe->addr = (uint64_t)addr;
1052 sqe->len = length;
1053 sqe->fadvise_advice = advice;
1054
1055 __submit_wait
1056 #endif
1057}
1058
1059int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) {
1060 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_OPENAT)
1061 return openat( dirfd, pathname, flags, mode );
1062 #else
1063 __submit_prelude
1064
1065 (*sqe){ IORING_OP_OPENAT, dirfd };
1066 sqe->addr = (uint64_t)pathname;
1067 sqe->open_flags = flags;
1068 sqe->mode = mode;
1069
1070 __submit_wait
1071 #endif
1072}
1073
1074int cfa_close(int fd) {
1075 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CLOSE)
1076 return close( fd );
1077 #else
1078 __submit_prelude
1079
1080 (*sqe){ IORING_OP_CLOSE, fd };
1081
1082 __submit_wait
1083 #endif
1084}
1085
1086
1087ssize_t cfa_read(int fd, void *buf, size_t count) {
1088 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READ)
1089 return read( fd, buf, count );
1090 #else
1091 __submit_prelude
1092
1093 (*sqe){ IORING_OP_READ, fd, buf, count, 0 };
1094
1095 __submit_wait
1096 #endif
1097}
1098
1099ssize_t cfa_write(int fd, void *buf, size_t count) {
1100 #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITE)
1101 return read( fd, buf, count );
1102 #else
1103 __submit_prelude
1104
1105 (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
1106
1107 __submit_wait
1108 #endif
1109}
1110
1111//-----------------------------------------------------------------------------
1112// Check if a function is asynchronous
1113
1114// Macro magic to reduce the size of the following switch case
1115#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
1116#define IS_DEFINED_SECOND(first, second, ...) second
1117#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
1118#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
1119
1120bool has_user_level_blocking( fptr_t func ) {
1121 #if defined(HAVE_LINUX_IO_URING_H)
1122 #if defined(HAVE_PREADV2)
1123 if( /*func == (fptr_t)preadv2 || */
1124 func == (fptr_t)cfa_preadv2 )
1125 #define _CFA_IO_FEATURE_IORING_OP_READV ,
1126 return IS_DEFINED(IORING_OP_READV);
1127 #endif
1128
1129 #if defined(HAVE_PWRITEV2)
1130 if( /*func == (fptr_t)pwritev2 || */
1131 func == (fptr_t)cfa_pwritev2 )
1132 #define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
1133 return IS_DEFINED(IORING_OP_WRITEV);
1134 #endif
1135
1136 if( /*func == (fptr_t)fsync || */
1137 func == (fptr_t)cfa_fsync )
1138 #define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
1139 return IS_DEFINED(IORING_OP_FSYNC);
1140
1141 if( /*func == (fptr_t)ync_file_range || */
1142 func == (fptr_t)cfa_sync_file_range )
1143 #define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
1144 return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
1145
1146 if( /*func == (fptr_t)sendmsg || */
1147 func == (fptr_t)cfa_sendmsg )
1148 #define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
1149 return IS_DEFINED(IORING_OP_SENDMSG);
1150
1151 if( /*func == (fptr_t)recvmsg || */
1152 func == (fptr_t)cfa_recvmsg )
1153 #define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
1154 return IS_DEFINED(IORING_OP_RECVMSG);
1155
1156 if( /*func == (fptr_t)send || */
1157 func == (fptr_t)cfa_send )
1158 #define _CFA_IO_FEATURE_IORING_OP_SEND ,
1159 return IS_DEFINED(IORING_OP_SEND);
1160
1161 if( /*func == (fptr_t)recv || */
1162 func == (fptr_t)cfa_recv )
1163 #define _CFA_IO_FEATURE_IORING_OP_RECV ,
1164 return IS_DEFINED(IORING_OP_RECV);
1165
1166 if( /*func == (fptr_t)accept4 || */
1167 func == (fptr_t)cfa_accept4 )
1168 #define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
1169 return IS_DEFINED(IORING_OP_ACCEPT);
1170
1171 if( /*func == (fptr_t)connect || */
1172 func == (fptr_t)cfa_connect )
1173 #define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
1174 return IS_DEFINED(IORING_OP_CONNECT);
1175
1176 if( /*func == (fptr_t)fallocate || */
1177 func == (fptr_t)cfa_fallocate )
1178 #define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
1179 return IS_DEFINED(IORING_OP_FALLOCATE);
1180
1181 if( /*func == (fptr_t)posix_fadvise || */
1182 func == (fptr_t)cfa_fadvise )
1183 #define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
1184 return IS_DEFINED(IORING_OP_FADVISE);
1185
1186 if( /*func == (fptr_t)madvise || */
1187 func == (fptr_t)cfa_madvise )
1188 #define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
1189 return IS_DEFINED(IORING_OP_MADVISE);
1190
1191 if( /*func == (fptr_t)openat || */
1192 func == (fptr_t)cfa_openat )
1193 #define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
1194 return IS_DEFINED(IORING_OP_OPENAT);
1195
1196 if( /*func == (fptr_t)close || */
1197 func == (fptr_t)cfa_close )
1198 #define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
1199 return IS_DEFINED(IORING_OP_CLOSE);
1200
1201 if( /*func == (fptr_t)read || */
1202 func == (fptr_t)cfa_read )
1203 #define _CFA_IO_FEATURE_IORING_OP_READ ,
1204 return IS_DEFINED(IORING_OP_READ);
1205
1206 if( /*func == (fptr_t)write || */
1207 func == (fptr_t)cfa_write )
1208 #define _CFA_IO_FEATURE_IORING_OP_WRITE ,
1209 return IS_DEFINED(IORING_OP_WRITE);
1210 #endif
1211
1212 return false;
1213}
Note: See TracBrowser for help on using the repository browser.