Context Navigation

io.cfa@ 8ad5752

Visit:

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 8ad5752 was 068a202, checked in by Thierry Delisle <tdelisle@…>, 5 years ago
Cluster io stats now print to STDOUT
Property mode set to `100644`
File size: 37.4 KB

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

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

Download in other formats:

Original Format