Context Navigation

io.cfa @ 05cfa4d

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-astnew-ast-unique-exprpthread-emulationqualifiedEnum

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

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

Download in other formats:

Original Format