Context Navigation

io.cfa@ f90d10f

Visit:

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 f90d10f was f90d10f, checked in by Thierry Delisle <tdelisle@…>, 6 years ago
Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc
Property mode set to `100644`
File size: 31.1 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 &, 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	bool waiting;
94	$thread thrd;
95	};
96
97	void ?{}( __io_poller_fast & this, struct cluster & cltr ) {
98	this.ring = cltr.io;
99	this.waiting = true;
100	(this.thrd){ "Fast I/O Poller", cltr };
101	}
102	void ^?{}( __io_poller_fast & mutex this );
103	void main( __io_poller_fast & this );
104	static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; }
105	void ^?{}( __io_poller_fast & mutex this ) {}
106
107	struct __submition_data {
108	// Head and tail of the ring (associated with array)
109	volatile uint32_t * head;
110	volatile uint32_t * tail;
111
112	// The actual kernel ring which uses head/tail
113	// indexes into the sqes arrays
114	uint32_t * array;
115
116	// number of entries and mask to go with it
117	const uint32_t * num;
118	const uint32_t * mask;
119
120	// Submission flags (Not sure what for)
121	uint32_t * flags;
122
123	// number of sqes not submitted (whatever that means)
124	uint32_t * dropped;
125
126	// Like head/tail but not seen by the kernel
127	volatile uint32_t alloc;
128	volatile uint32_t ready;
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	unsigned long long int val;
144	unsigned long long int cnt;
145	} submit_avg;
146	} stats;
147	#endif
148	};
149
150	struct __completion_data {
151	// Head and tail of the ring
152	volatile uint32_t * head;
153	volatile uint32_t * tail;
154
155	// number of entries and mask to go with it
156	const uint32_t * mask;
157	const uint32_t * num;
158
159	// number of cqes not submitted (whatever that means)
160	uint32_t * overflow;
161
162	// the kernel ring
163	struct io_uring_cqe * cqes;
164
165	// The location and size of the mmaped area
166	void * ring_ptr;
167	size_t ring_sz;
168
169	// Statistics
170	#if !defined(__CFA_NO_STATISTICS__)
171	struct {
172	struct {
173	unsigned long long int val;
174	unsigned long long int slow_cnt;
175	unsigned long long int fast_cnt;
176	} completed_avg;
177	} stats;
178	#endif
179	};
180
181	struct __io_data {
182	struct __submition_data submit_q;
183	struct __completion_data completion_q;
184	uint32_t flags;
185	int fd;
186	semaphore submit;
187	volatile bool done;
188	struct {
189	struct {
190	void * stack;
191	pthread_t kthrd;
192	} slow;
193	__io_poller_fast fast;
194	__bin_sem_t sem;
195	} poller;
196	};
197
198	//=============================================================================================
199	// I/O Startup / Shutdown logic
200	//=============================================================================================
201	void __kernel_io_startup( cluster & this, bool main_cluster ) {
202	this.io = malloc();
203
204	// Step 1 : call to setup
205	struct io_uring_params params;
206	memset(&params, 0, sizeof(params));
207
208	uint32_t nentries = entries_per_cluster();
209
210	int fd = syscall(__NR_io_uring_setup, nentries, &params );
211	if(fd < 0) {
212	abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
213	}
214
215	// Step 2 : mmap result
216	memset( this.io, 0, sizeof(struct __io_data) );
217	struct __submition_data & sq = this.io->submit_q;
218	struct __completion_data & cq = this.io->completion_q;
219
220	// calculate the right ring size
221	sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned) );
222	cq.ring_sz = params.cq_off.cqes + (params.cq_entries * sizeof(struct io_uring_cqe));
223
224	// Requires features
225	#if defined(IORING_FEAT_SINGLE_MMAP)
226	// adjust the size according to the parameters
227	if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
228	cq->ring_sz = sq->ring_sz = max(cq->ring_sz, sq->ring_sz);
229	}
230	#endif
231
232	// mmap the Submit Queue into existence
233	sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_SQ_RING);
234	if (sq.ring_ptr == (void*)MAP_FAILED) {
235	abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
236	}
237
238	// Requires features
239	#if defined(IORING_FEAT_SINGLE_MMAP)
240	// mmap the Completion Queue into existence (may or may not be needed)
241	if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
242	cq->ring_ptr = sq->ring_ptr;
243	}
244	else
245	#endif
246	{
247	// We need multiple call to MMAP
248	cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_CQ_RING);
249	if (cq.ring_ptr == (void*)MAP_FAILED) {
250	munmap(sq.ring_ptr, sq.ring_sz);
251	abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
252	}
253	}
254
255	// mmap the submit queue entries
256	size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
257	sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_SQES);
258	if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
259	munmap(sq.ring_ptr, sq.ring_sz);
260	if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
261	abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
262	}
263
264	// Get the pointers from the kernel to fill the structure
265	// submit queue
266	sq.head = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
267	sq.tail = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
268	sq.mask = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
269	sq.num = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
270	sq.flags = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
271	sq.dropped = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
272	sq.array = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
273	sq.alloc = *sq.tail;
274	sq.ready = *sq.tail;
275
276	// completion queue
277	cq.head = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
278	cq.tail = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
279	cq.mask = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
280	cq.num = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
281	cq.overflow = ( uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
282	cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
283
284	// some paranoid checks
285	/* 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 );
286	/* paranoid / verifyf( (cq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
287	/* paranoid / verifyf( (cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
288	/* paranoid / verifyf( (cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
289
290	/* 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 );
291	/* paranoid / verifyf( (sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
292	/* paranoid / verifyf( (sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
293	/* paranoid / verifyf( (sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
294
295	// Update the global ring info
296	this.io->flags = params.flags;
297	this.io->fd = fd;
298	this.io->done = false;
299	(this.io->submit){ min(sq.num, cq.num) };
300
301	// Initialize statistics
302	#if !defined(__CFA_NO_STATISTICS__)
303	this.io->submit_q.stats.submit_avg.val = 0;
304	this.io->submit_q.stats.submit_avg.cnt = 0;
305	this.io->completion_q.stats.completed_avg.val = 0;
306	this.io->completion_q.stats.completed_avg.slow_cnt = 0;
307	this.io->completion_q.stats.completed_avg.fast_cnt = 0;
308	#endif
309
310	if(!main_cluster) {
311	__kernel_io_finish_start( this );
312	}
313	}
314
315	void __kernel_io_finish_start( cluster & this ) {
316	__cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
317	(this.io->poller.fast){ this };
318	__thrd_start( this.io->poller.fast, main );
319
320	// Create the poller thread
321	__cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
322	this.io->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );
323	}
324
325	void __kernel_io_prepare_stop( cluster & this ) {
326	__cfadbg_print_safe(io_core, "Kernel I/O : Stopping pollers for cluster\n", &this);
327	// Notify the poller thread of the shutdown
328	__atomic_store_n(&this.io->done, true, __ATOMIC_SEQ_CST);
329
330	// Stop the IO Poller
331	sigval val = { 1 };
332	pthread_sigqueue( this.io->poller.slow.kthrd, SIGUSR1, val );
333	post( this.io->poller.sem );
334
335	// Wait for the poller thread to finish
336	pthread_join( this.io->poller.slow.kthrd, 0p );
337	free( this.io->poller.slow.stack );
338
339	__cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this);
340
341	#if defined(__CFA_IO_POLLING_USER__)
342	verify( this.io->poller.fast.waiting );
343	verify( this.io->poller.fast.thrd.state == Blocked );
344
345	this.io->poller.fast.thrd.curr_cluster = mainCluster;
346
347	// unpark the fast io_poller
348	unpark( &this.io->poller.fast.thrd __cfaabi_dbg_ctx2 );
349
350	^(this.io->poller.fast){};
351
352	__cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this);
353	#endif
354	}
355
356	void __kernel_io_shutdown( cluster & this, bool main_cluster ) {
357	if(!main_cluster) {
358	__kernel_io_prepare_stop( this );
359	}
360
361	// print statistics
362	#if !defined(__CFA_NO_STATISTICS__)
363	if(this.print_stats) {
364	with(this.io->submit_q.stats, this.io->completion_q.stats) {
365	__cfaabi_bits_print_safe( STDERR_FILENO,
366	"----- I/O uRing Stats -----\n"
367	"- total submit calls : %llu\n"
368	"- avg submit : %lf\n"
369	"- total wait calls : %llu (%llu slow, %llu fast)\n"
370	"- avg completion/wait : %lf\n",
371	submit_avg.cnt,
372	((double)submit_avg.val) / submit_avg.cnt,
373	completed_avg.slow_cnt + completed_avg.fast_cnt,
374	completed_avg.slow_cnt, completed_avg.fast_cnt,
375	((double)completed_avg.val) / (completed_avg.slow_cnt + completed_avg.fast_cnt)
376	);
377	}
378	}
379	#endif
380
381	// Shutdown the io rings
382	struct __submition_data & sq = this.io->submit_q;
383	struct __completion_data & cq = this.io->completion_q;
384
385	// unmap the submit queue entries
386	munmap(sq.sqes, (sq.num) sizeof(struct io_uring_sqe));
387
388	// unmap the Submit Queue ring
389	munmap(sq.ring_ptr, sq.ring_sz);
390
391	// unmap the Completion Queue ring, if it is different
392	if (cq.ring_ptr != sq.ring_ptr) {
393	munmap(cq.ring_ptr, cq.ring_sz);
394	}
395
396	// close the file descriptor
397	close(this.io->fd);
398
399	free( this.io );
400	}
401
402	//=============================================================================================
403	// I/O Polling
404	//=============================================================================================
405	struct io_user_data {
406	int32_t result;
407	$thread * thrd;
408	};
409
410	// Process a single completion message from the io_uring
411	// This is NOT thread-safe
412	static int __drain_io( struct __io_data & ring, sigset_t * mask, int waitcnt, bool in_kernel ) {
413	int ret = syscall( __NR_io_uring_enter, ring.fd, 0, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8);
414	if( ret < 0 ) {
415	switch((int)errno) {
416	case EAGAIN:
417	case EINTR:
418	return -EAGAIN;
419	default:
420	abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
421	}
422	}
423
424	// Drain the queue
425	unsigned head = *ring.completion_q.head;
426	unsigned tail = __atomic_load_n(ring.completion_q.tail, __ATOMIC_ACQUIRE);
427
428	// Nothing was new return 0
429	if (head == tail) {
430	return 0;
431	}
432
433	uint32_t count = tail - head;
434	for(i; count) {
435	unsigned idx = (head + i) & (*ring.completion_q.mask);
436	struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
437
438	/* paranoid */ verify(&cqe);
439
440	struct io_user_data * data = (struct io_user_data *)cqe.user_data;
441	__cfadbg_print_safe( io, "Kernel I/O : Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
442
443	data->result = cqe.res;
444	if(!in_kernel) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
445	else { __unpark( data->thrd __cfaabi_dbg_ctx2 ); }
446	}
447
448	// Allow new submissions to happen
449	V(ring.submit, count);
450
451	// Mark to the kernel that the cqe has been seen
452	// Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
453	__atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED );
454
455	return count;
456	}
457
458	static void * __io_poller_slow( void * arg ) {
459	cluster * cltr = (cluster *)arg;
460	struct __io_data & ring = *cltr->io;
461
462	sigset_t mask;
463	sigfillset(&mask);
464	if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
465	abort( "KERNEL ERROR: IO_URING - pthread_sigmask" );
466	}
467
468	sigdelset( &mask, SIGUSR1 );
469
470	verify( (ring.submit_q.head) == (ring.submit_q.tail) );
471	verify( (ring.completion_q.head) == (ring.completion_q.tail) );
472
473	__cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);
474
475	while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
476	#if defined(__CFA_IO_POLLING_USER__)
477
478	// In the user-thread approach drain and if anything was drained,
479	// batton pass to the user-thread
480	int count = __drain_io( ring, &mask, 1, true );
481
482	// Update statistics
483	#if !defined(__CFA_NO_STATISTICS__)
484	ring.completion_q.stats.completed_avg.val += count;
485	ring.completion_q.stats.completed_avg.slow_cnt += 1;
486	#endif
487
488	if(count > 0) {
489	__cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to fast poller\n", &ring);
490	__unpark( &ring.poller.fast.thrd __cfaabi_dbg_ctx2 );
491	wait( ring.poller.sem );
492	}
493
494	#else
495
496	//In the naive approach, just poll the io completion queue directly
497	int count = __drain_io( ring, &mask, 1, true );
498
499	// Update statistics
500	#if !defined(__CFA_NO_STATISTICS__)
501	ring.completion_q.stats.completed_avg.val += count;
502	ring.completion_q.stats.completed_avg.slow_cnt += 1;
503	#endif
504
505	#endif
506	}
507
508	__cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p stopping\n", &ring);
509
510	return 0p;
511	}
512
513	void main( __io_poller_fast & this ) {
514	// Start parked
515	park( __cfaabi_dbg_ctx );
516
517	__cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p ready\n", &this.ring);
518
519	// Then loop until we need to start
520	while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) {
521	// Drain the io
522	this.waiting = false;
523	int count = __drain_io( *this.ring, 0p, 0, false );
524
525	// Update statistics
526	#if !defined(__CFA_NO_STATISTICS__)
527	this.ring->completion_q.stats.completed_avg.val += count;
528	this.ring->completion_q.stats.completed_avg.fast_cnt += 1;
529	#endif
530
531	this.waiting = true;
532	if(0 > count) {
533	// If we got something, just yield and check again
534	yield();
535	}
536	else {
537	// We didn't get anything baton pass to the slow poller
538	__cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring);
539	post( this.ring->poller.sem );
540	park( __cfaabi_dbg_ctx );
541	}
542	}
543
544	__cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring);
545	}
546
547	//=============================================================================================
548	// I/O Submissions
549	//=============================================================================================
550
551	// Submition steps :
552	// 1 - We need to make sure we don't overflow any of the buffer, P(ring.submit) to make sure
553	// entries are available. The semaphore make sure that there is no more operations in
554	// progress then the number of entries in the buffer. This probably limits concurrency
555	// more than necessary since submitted but not completed operations don't need any
556	// entries in user space. However, I don't know what happens if we overflow the buffers
557	// because too many requests completed at once. This is a safe approach in all cases.
558	// Furthermore, with hundreds of entries, this may be okay.
559	//
560	// 2 - Allocate a queue entry. The ring already has memory for all entries but only the ones
561	// listed in sq.array are visible by the kernel. For those not listed, the kernel does not
562	// offer any assurance that an entry is not being filled by multiple flags. Therefore, we
563	// need to write an allocator that allows allocating concurrently.
564	//
565	// 3 - Actually fill the submit entry, this is the only simple and straightforward step.
566	//
567	// 4 - Append the entry index to the array and adjust the tail accordingly. This operation
568	// needs to arrive to two concensus at the same time:
569	// A - The order in which entries are listed in the array: no two threads must pick the
570	// same index for their entries
571	// B - When can the tail be update for the kernel. EVERY entries in the array between
572	// head and tail must be fully filled and shouldn't ever be touched again.
573	//
574
575	static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data & ring ) {
576	// Wait for a spot to be available
577	P(ring.submit);
578
579	// Allocate the sqe
580	uint32_t idx = __atomic_fetch_add(&ring.submit_q.alloc, 1ul32, __ATOMIC_SEQ_CST);
581
582	// Validate that we didn't overflow anything
583	// Check that nothing overflowed
584	/* paranoid */ verify( true );
585
586	// Check that it goes head -> tail -> alloc and never head -> alloc -> tail
587	/* paranoid */ verify( true );
588
589	// Return the sqe
590	return [&ring.submit_q.sqes[ idx & (*ring.submit_q.mask)], idx];
591	}
592
593	static inline void __submit( struct __io_data & ring, uint32_t idx ) {
594	// get mutual exclusion
595	lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
596
597	// Append to the list of ready entries
598	uint32_t * tail = ring.submit_q.tail;
599	const uint32_t mask = *ring.submit_q.mask;
600
601	ring.submit_q.array[ (*tail) & mask ] = idx & mask;
602	__atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
603
604	// Submit however, many entries need to be submitted
605	int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
606	if( ret < 0 ) {
607	switch((int)errno) {
608	default:
609	abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
610	}
611	}
612
613	// update statistics
614	#if !defined(__CFA_NO_STATISTICS__)
615	ring.submit_q.stats.submit_avg.val += 1;
616	ring.submit_q.stats.submit_avg.cnt += 1;
617	#endif
618
619	unlock(ring.submit_q.lock);
620	// Make sure that idx was submitted
621	// Be careful to not get false positive if we cycled the entire list or that someone else submitted for us
622	__cfadbg_print_safe( io, "Kernel I/O : Performed io_submit for %p, returned %d\n", active_thread(), ret );
623	}
624
625	static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
626	this.opcode = opcode;
627	#if !defined(IOSQE_ASYNC)
628	this.flags = 0;
629	#else
630	this.flags = IOSQE_ASYNC;
631	#endif
632	this.ioprio = 0;
633	this.fd = fd;
634	this.off = 0;
635	this.addr = 0;
636	this.len = 0;
637	this.rw_flags = 0;
638	this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0;
639	}
640
641	static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
642	(this){ opcode, fd };
643	this.off = off;
644	this.addr = (uint64_t)addr;
645	this.len = len;
646	}
647
648
649	//=============================================================================================
650	// I/O Interface
651	//=============================================================================================
652
653	#define __submit_prelude \
654	struct __io_data & ring = *active_cluster()->io; \
655	struct io_uring_sqe * sqe; \
656	uint32_t idx; \
657	[sqe, idx] = __submit_alloc( ring );
658
659	#define __submit_wait \
660	io_user_data data = { 0, active_thread() }; \
661	/__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );/ \
662	sqe->user_data = (uint64_t)&data; \
663	__submit( ring, idx ); \
664	park( __cfaabi_dbg_ctx ); \
665	return data.result;
666	#endif
667
668	// Some forward declarations
669	extern "C" {
670	#include <unistd.h>
671	#include <sys/types.h>
672	#include <sys/socket.h>
673	#include <sys/syscall.h>
674	struct iovec;
675	extern ssize_t preadv2 (int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
676	extern ssize_t pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
677
678	extern int fsync(int fd);
679	extern int sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags);
680
681	struct msghdr;
682	struct sockaddr;
683	extern ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags);
684	extern ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);
685	extern ssize_t send(int sockfd, const void *buf, size_t len, int flags);
686	extern ssize_t recv(int sockfd, void *buf, size_t len, int flags);
687	extern int accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags);
688	extern int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
689
690	extern int fallocate(int fd, int mode, uint64_t offset, uint64_t len);
691	extern int posix_fadvise(int fd, uint64_t offset, uint64_t len, int advice);
692	extern int madvise(void *addr, size_t length, int advice);
693
694	extern int openat(int dirfd, const char *pathname, int flags, mode_t mode);
695	extern int close(int fd);
696
697	extern ssize_t read (int fd, void *buf, size_t count);
698	}
699
700	//-----------------------------------------------------------------------------
701	// Asynchronous operations
702	ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
703	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_READV)
704	return preadv2(fd, iov, iovcnt, offset, flags);
705	#else
706	__submit_prelude
707
708	(*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
709
710	__submit_wait
711	#endif
712	}
713
714	ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
715	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_WRITEV)
716	return pwritev2(fd, iov, iovcnt, offset, flags);
717	#else
718	__submit_prelude
719
720	(*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
721
722	__submit_wait
723	#endif
724	}
725
726	int cfa_fsync(int fd) {
727	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_FSYNC)
728	return fsync(fd);
729	#else
730	__submit_prelude
731
732	(*sqe){ IORING_OP_FSYNC, fd };
733
734	__submit_wait
735	#endif
736	}
737
738	int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) {
739	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_SYNC_FILE_RANGE)
740	return sync_file_range(fd, offset, nbytes, flags);
741	#else
742	__submit_prelude
743
744	(*sqe){ IORING_OP_SYNC_FILE_RANGE, fd };
745	sqe->off = offset;
746	sqe->len = nbytes;
747	sqe->sync_range_flags = flags;
748
749	__submit_wait
750	#endif
751	}
752
753
754	ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) {
755	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_SENDMSG)
756	return sendmsg(sockfd, msg, flags);
757	#else
758	__submit_prelude
759
760	(*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 };
761	sqe->msg_flags = flags;
762
763	__submit_wait
764	#endif
765	}
766
767	ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) {
768	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_RECVMSG)
769	return recvmsg(sockfd, msg, flags);
770	#else
771	__submit_prelude
772
773	(*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 };
774	sqe->msg_flags = flags;
775
776	__submit_wait
777	#endif
778	}
779
780	ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) {
781	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_SEND)
782	return send( sockfd, buf, len, flags );
783	#else
784	__submit_prelude
785
786	(*sqe){ IORING_OP_SEND, sockfd };
787	sqe->addr = (uint64_t)buf;
788	sqe->len = len;
789	sqe->msg_flags = flags;
790
791	__submit_wait
792	#endif
793	}
794
795	ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) {
796	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_RECV)
797	return recv( sockfd, buf, len, flags );
798	#else
799	__submit_prelude
800
801	(*sqe){ IORING_OP_RECV, sockfd };
802	sqe->addr = (uint64_t)buf;
803	sqe->len = len;
804	sqe->msg_flags = flags;
805
806	__submit_wait
807	#endif
808	}
809
810	int cfa_accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags) {
811	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_ACCEPT)
812	return accept4( sockfd, addr, addrlen, flags );
813	#else
814	__submit_prelude
815
816	(*sqe){ IORING_OP_ACCEPT, sockfd };
817	sqe->addr = addr;
818	sqe->addr2 = addrlen;
819	sqe->accept_flags = flags;
820
821	__submit_wait
822	#endif
823	}
824
825	int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
826	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_CONNECT)
827	return connect( sockfd, addr, addrlen );
828	#else
829	__submit_prelude
830
831	(*sqe){ IORING_OP_CONNECT, sockfd };
832	sqe->addr = (uint64_t)addr;
833	sqe->off = addrlen;
834
835	__submit_wait
836	#endif
837	}
838
839	int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) {
840	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_FALLOCATE)
841	return fallocate( fd, mode, offset, len );
842	#else
843	__submit_prelude
844
845	(*sqe){ IORING_OP_FALLOCATE, fd };
846	sqe->off = offset;
847	sqe->len = length;
848	sqe->mode = mode;
849
850	__submit_wait
851	#endif
852	}
853
854	int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) {
855	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_FADVISE)
856	return posix_fadvise( fd, offset, len, advice );
857	#else
858	__submit_prelude
859
860	(*sqe){ IORING_OP_FADVISE, fd };
861	sqe->off = (uint64_t)offset;
862	sqe->len = length;
863	sqe->fadvise_advice = advice;
864
865	__submit_wait
866	#endif
867	}
868
869	int cfa_madvise(void *addr, size_t length, int advice) {
870	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_MADVISE)
871	return madvise( addr, length, advice );
872	#else
873	__submit_prelude
874
875	(*sqe){ IORING_OP_MADVISE, 0 };
876	sqe->addr = (uint64_t)addr;
877	sqe->len = length;
878	sqe->fadvise_advice = advice;
879
880	__submit_wait
881	#endif
882	}
883
884	int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) {
885	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_OPENAT)
886	return openat( dirfd, pathname, flags, mode );
887	#else
888	__submit_prelude
889
890	(*sqe){ IORING_OP_OPENAT, dirfd };
891	sqe->addr = (uint64_t)pathname;
892	sqe->open_flags = flags;
893	sqe->mode = mode;
894
895	__submit_wait
896	#endif
897	}
898
899	int cfa_close(int fd) {
900	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_CLOSE)
901	return close( fd );
902	#else
903	__submit_prelude
904
905	(*sqe){ IORING_OP_CLOSE, fd };
906
907	__submit_wait
908	#endif
909	}
910
911
912	ssize_t cfa_read(int fd, void *buf, size_t count) {
913	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_READ)
914	return read( fd, buf, count );
915	#else
916	__submit_prelude
917
918	(*sqe){ IORING_OP_READ, fd, buf, count, 0 };
919
920	__submit_wait
921	#endif
922	}
923
924	ssize_t cfa_write(int fd, void *buf, size_t count) {
925	#if !defined(HAVE_LINUX_IO_URING_H) \|\| !defined(IORING_OP_WRITE)
926	return read( fd, buf, count );
927	#else
928	__submit_prelude
929
930	(*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
931
932	__submit_wait
933	#endif
934	}
935
936	//-----------------------------------------------------------------------------
937	// Check if a function is asynchronous
938
939	// Macro magic to reduce the size of the following switch case
940	#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
941	#define IS_DEFINED_SECOND(first, second, ...) second
942	#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
943	#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
944
945	bool has_user_level_blocking( fptr_t func ) {
946	#if defined(HAVE_LINUX_IO_URING_H)
947	if( /func == (fptr_t)preadv2 \|\| /
948	func == (fptr_t)cfa_preadv2 )
949	#define _CFA_IO_FEATURE_IORING_OP_READV ,
950	return IS_DEFINED(IORING_OP_READV);
951
952	if( /func == (fptr_t)pwritev2 \|\| /
953	func == (fptr_t)cfa_pwritev2 )
954	#define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
955	return IS_DEFINED(IORING_OP_WRITEV);
956
957	if( /func == (fptr_t)fsync \|\| /
958	func == (fptr_t)cfa_fsync )
959	#define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
960	return IS_DEFINED(IORING_OP_FSYNC);
961
962	if( /func == (fptr_t)ync_file_range \|\| /
963	func == (fptr_t)cfa_sync_file_range )
964	#define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
965	return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
966
967	if( /func == (fptr_t)sendmsg \|\| /
968	func == (fptr_t)cfa_sendmsg )
969	#define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
970	return IS_DEFINED(IORING_OP_SENDMSG);
971
972	if( /func == (fptr_t)recvmsg \|\| /
973	func == (fptr_t)cfa_recvmsg )
974	#define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
975	return IS_DEFINED(IORING_OP_RECVMSG);
976
977	if( /func == (fptr_t)send \|\| /
978	func == (fptr_t)cfa_send )
979	#define _CFA_IO_FEATURE_IORING_OP_SEND ,
980	return IS_DEFINED(IORING_OP_SEND);
981
982	if( /func == (fptr_t)recv \|\| /
983	func == (fptr_t)cfa_recv )
984	#define _CFA_IO_FEATURE_IORING_OP_RECV ,
985	return IS_DEFINED(IORING_OP_RECV);
986
987	if( /func == (fptr_t)accept4 \|\| /
988	func == (fptr_t)cfa_accept4 )
989	#define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
990	return IS_DEFINED(IORING_OP_ACCEPT);
991
992	if( /func == (fptr_t)connect \|\| /
993	func == (fptr_t)cfa_connect )
994	#define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
995	return IS_DEFINED(IORING_OP_CONNECT);
996
997	if( /func == (fptr_t)fallocate \|\| /
998	func == (fptr_t)cfa_fallocate )
999	#define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
1000	return IS_DEFINED(IORING_OP_FALLOCATE);
1001
1002	if( /func == (fptr_t)posix_fadvise \|\| /
1003	func == (fptr_t)cfa_fadvise )
1004	#define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
1005	return IS_DEFINED(IORING_OP_FADVISE);
1006
1007	if( /func == (fptr_t)madvise \|\| /
1008	func == (fptr_t)cfa_madvise )
1009	#define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
1010	return IS_DEFINED(IORING_OP_MADVISE);
1011
1012	if( /func == (fptr_t)openat \|\| /
1013	func == (fptr_t)cfa_openat )
1014	#define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
1015	return IS_DEFINED(IORING_OP_OPENAT);
1016
1017	if( /func == (fptr_t)close \|\| /
1018	func == (fptr_t)cfa_close )
1019	#define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
1020	return IS_DEFINED(IORING_OP_CLOSE);
1021
1022	if( /func == (fptr_t)read \|\| /
1023	func == (fptr_t)cfa_read )
1024	#define _CFA_IO_FEATURE_IORING_OP_READ ,
1025	return IS_DEFINED(IORING_OP_READ);
1026
1027	if( /func == (fptr_t)write \|\| /
1028	func == (fptr_t)cfa_write )
1029	#define _CFA_IO_FEATURE_IORING_OP_WRITE ,
1030	return IS_DEFINED(IORING_OP_WRITE);
1031	#endif
1032
1033	return false;
1034	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats:

Original Format