Context Navigation

io.cfa @ 8962722

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

Last change on this file since 8962722 was 8962722, checked in by Thierry Delisle <tdelisle@…>, 4 years ago
Replace statx call with syscall since it is not on all machines
Property mode set to `100644`
File size: 22.6 KB

Rev	Line
[92976d9]	1	#include "kernel.hfa"
	2
	3	#if !defined(HAVE_LINUX_IO_URING_H)
	4	void __kernel_io_startup( cluster & this ) {
	5	// Nothing to do without io_uring
	6	}
	7
	8	void __kernel_io_shutdown( cluster & this ) {
	9	// Nothing to do without io_uring
	10	}
	11
	12	bool is_async( void (*)() ) {
	13	return false;
	14	}
	15
	16	#else
	17	extern "C" {
	18	#define _GNU_SOURCE /* See feature_test_macros(7) */
	19	#include <errno.h>
	20	#include <stdint.h>
	21	#include <string.h>
	22	#include <unistd.h>
	23	#include <sys/mman.h>
	24	#include <sys/syscall.h>
	25
	26	#include <linux/io_uring.h>
	27	}
	28
	29	#include "bits/signal.hfa"
	30	#include "kernel_private.hfa"
	31	#include "thread.hfa"
	32
	33	uint32_t entries_per_cluster() {
	34	return 256;
	35	}
	36
	37	static void * __io_poller( void * arg );
	38
[185efe6]	39	// Weirdly, some systems that do support io_uring don't actually define these
	40	#ifdef __alpha__
	41	/*
	42	* alpha is the only exception, all other architectures
	43	* have common numbers for new system calls.
	44	*/
	45	# ifndef __NR_io_uring_setup
	46	# define __NR_io_uring_setup 535
	47	# endif
	48	# ifndef __NR_io_uring_enter
	49	# define __NR_io_uring_enter 536
	50	# endif
	51	# ifndef __NR_io_uring_register
	52	# define __NR_io_uring_register 537
	53	# endif
	54	#else /* !__alpha__ */
	55	# ifndef __NR_io_uring_setup
	56	# define __NR_io_uring_setup 425
	57	# endif
	58	# ifndef __NR_io_uring_enter
	59	# define __NR_io_uring_enter 426
	60	# endif
	61	# ifndef __NR_io_uring_register
	62	# define __NR_io_uring_register 427
	63	# endif
	64	#endif
	65
	66
[92976d9]	67	//=============================================================================================
	68	// I/O Startup / Shutdown logic
	69	//=============================================================================================
	70	void __kernel_io_startup( cluster & this ) {
	71	// Step 1 : call to setup
	72	struct io_uring_params params;
	73	memset(&params, 0, sizeof(params));
	74
[2d8f7b0]	75	uint32_t nentries = entries_per_cluster();
	76
	77	int fd = syscall(__NR_io_uring_setup, nentries, &params );
[92976d9]	78	if(fd < 0) {
	79	abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
	80	}
	81
	82	// Step 2 : mmap result
	83	memset(&this.io, 0, sizeof(struct io_ring));
[2d8f7b0]	84	struct io_uring_sq & sq = this.io.submit_q;
	85	struct io_uring_cq & cq = this.io.completion_q;
[92976d9]	86
	87	// calculate the right ring size
[2d8f7b0]	88	sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned) );
	89	cq.ring_sz = params.cq_off.cqes + (params.cq_entries * sizeof(struct io_uring_cqe));
[92976d9]	90
	91	// Requires features
	92	// // adjust the size according to the parameters
	93	// if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
	94	// cq->ring_sz = sq->ring_sz = max(cq->ring_sz, sq->ring_sz);
	95	// }
	96
	97	// mmap the Submit Queue into existence
[2d8f7b0]	98	sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_SQ_RING);
	99	if (sq.ring_ptr == (void*)MAP_FAILED) {
[92976d9]	100	abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
	101	}
	102
	103	// mmap the Completion Queue into existence (may or may not be needed)
	104	// Requires features
	105	// if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
	106	// cq->ring_ptr = sq->ring_ptr;
	107	// }
	108	// else {
	109	// We need multiple call to MMAP
[2d8f7b0]	110	cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_CQ_RING);
	111	if (cq.ring_ptr == (void*)MAP_FAILED) {
	112	munmap(sq.ring_ptr, sq.ring_sz);
[92976d9]	113	abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
	114	}
	115	// }
	116
	117	// mmap the submit queue entries
	118	size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
[2d8f7b0]	119	sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd, IORING_OFF_SQES);
	120	if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
	121	munmap(sq.ring_ptr, sq.ring_sz);
	122	if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
[92976d9]	123	abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
	124	}
	125
	126	// Get the pointers from the kernel to fill the structure
	127	// submit queue
[2d8f7b0]	128	sq.head = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
	129	sq.tail = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
	130	sq.mask = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
	131	sq.num = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
	132	sq.flags = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
	133	sq.dropped = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
	134	sq.array = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
	135	sq.alloc = *sq.tail;
[92976d9]	136
	137	// completion queue
[2d8f7b0]	138	cq.head = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
	139	cq.tail = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
	140	cq.mask = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
	141	cq.num = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
	142	cq.overflow = ( uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
	143	cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
	144
	145	// some paranoid checks
	146	/* 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 );
	147	/* paranoid / verifyf( (cq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
	148	/* paranoid / verifyf( (cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
	149	/* paranoid / verifyf( (cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
	150
	151	/* 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 );
	152	/* paranoid / verifyf( (sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
	153	/* paranoid / verifyf( (sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
	154	/* paranoid / verifyf( (sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
[92976d9]	155
	156	// Update the global ring info
	157	this.io.flags = params.flags;
	158	this.io.fd = fd;
	159	this.io.done = false;
[2d8f7b0]	160	(this.io.submit){ min(sq.num, cq.num) };
[92976d9]	161
	162	// Create the poller thread
	163	this.io.stack = __create_pthread( &this.io.poller, __io_poller, &this );
	164	}
	165
	166	void __kernel_io_shutdown( cluster & this ) {
	167	// Stop the IO Poller
	168	// Notify the poller thread of the shutdown
	169	__atomic_store_n(&this.io.done, true, __ATOMIC_SEQ_CST);
	170	sigval val = { 1 };
	171	pthread_sigqueue( this.io.poller, SIGUSR1, val );
	172
	173	// Wait for the poller thread to finish
	174	pthread_join( this.io.poller, 0p );
	175	free( this.io.stack );
	176
	177	// Shutdown the io rings
	178	struct io_uring_sq & sq = this.io.submit_q;
	179	struct io_uring_cq & cq = this.io.completion_q;
	180
	181	// unmap the submit queue entries
[2d8f7b0]	182	munmap(sq.sqes, (sq.num) sizeof(struct io_uring_sqe));
[92976d9]	183
	184	// unmap the Submit Queue ring
	185	munmap(sq.ring_ptr, sq.ring_sz);
	186
	187	// unmap the Completion Queue ring, if it is different
	188	if (cq.ring_ptr != sq.ring_ptr) {
	189	munmap(cq.ring_ptr, cq.ring_sz);
	190	}
	191
	192	// close the file descriptor
	193	close(this.io.fd);
	194	}
	195
	196	//=============================================================================================
	197	// I/O Polling
	198	//=============================================================================================
	199	struct io_user_data {
	200	int32_t result;
	201	$thread * thrd;
	202	};
	203
	204	// Process a single completion message from the io_uring
	205	// This is NOT thread-safe
	206	static bool __io_process(struct io_ring & ring) {
	207	unsigned head = *ring.completion_q.head;
	208	unsigned tail = __atomic_load_n(ring.completion_q.tail, __ATOMIC_ACQUIRE);
	209
	210	if (head == tail) return false;
	211
	212	unsigned idx = head & (*ring.completion_q.mask);
	213	struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
	214
	215	/* paranoid */ verify(&cqe);
	216
	217	struct io_user_data * data = (struct io_user_data *)cqe.user_data;
[185efe6]	218	// __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
[2d8f7b0]	219
[92976d9]	220	data->result = cqe.res;
[2d8f7b0]	221	__unpark( data->thrd __cfaabi_dbg_ctx2 );
	222
	223	// Allow new submissions to happen
	224	V(ring.submit);
[92976d9]	225
	226	// Mark to the kernel that the cqe has been seen
	227	// Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
[2d8f7b0]	228	__atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED );
[92976d9]	229
	230	return true;
	231	}
	232
	233	static void * __io_poller( void * arg ) {
	234	cluster * cltr = (cluster *)arg;
	235	struct io_ring & ring = cltr->io;
	236
	237	sigset_t mask;
	238	sigfillset(&mask);
	239	if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
	240	abort( "KERNEL ERROR: IO_URING - pthread_sigmask" );
	241	}
	242
	243	sigdelset( &mask, SIGUSR1 );
	244
	245	verify( (ring.submit_q.head) == (ring.submit_q.tail) );
	246	verify( (ring.completion_q.head) == (ring.completion_q.tail) );
	247
	248	LOOP: while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
	249	int ret = syscall( __NR_io_uring_enter, ring.fd, 0, 1, IORING_ENTER_GETEVENTS, &mask, _NSIG / 8);
	250	if( ret < 0 ) {
	251	switch((int)errno) {
	252	case EAGAIN:
	253	case EINTR:
	254	continue LOOP;
	255	default:
	256	abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
	257	}
	258	}
	259
	260	// Drain the queue
	261	while(__io_process(ring)) {}
	262	}
	263
	264	return 0p;
	265	}
	266
	267	//=============================================================================================
	268	// I/O Submissions
	269	//=============================================================================================
	270
[2d8f7b0]	271	// Submition steps :
	272	// 1 - We need to make sure we don't overflow any of the buffer, P(ring.submit) to make sure
	273	// entries are available. The semaphore make sure that there is no more operations in
	274	// progress then the number of entries in the buffer. This probably limits concurrency
	275	// more than necessary since submitted but not completed operations don't need any
	276	// entries in user space. However, I don't know what happens if we overflow the buffers
	277	// because too many requests completed at once. This is a safe approach in all cases.
	278	// Furthermore, with hundreds of entries, this may be okay.
	279	//
	280	// 2 - Allocate a queue entry. The ring already has memory for all entries but only the ones
	281	// listed in sq.array are visible by the kernel. For those not listed, the kernel does not
	282	// offer any assurance that an entry is not being filled by multiple flags. Therefore, we
	283	// need to write an allocator that allows allocating concurrently.
	284	//
	285	// 3 - Actually fill the submit entry, this is the only simple and straightforward step.
	286	//
	287	// 4 - Append the entry index to the array and adjust the tail accordingly. This operation
	288	// needs to arrive to two concensus at the same time:
	289	// A - The order in which entries are listed in the array: no two threads must pick the
	290	// same index for their entries
	291	// B - When can the tail be update for the kernel. EVERY entries in the array between
	292	// head and tail must be fully filled and shouldn't ever be touched again.
	293	//
	294
	295	static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct io_ring & ring ) {
	296	// Wait for a spot to be available
	297	P(ring.submit);
	298
	299	// Allocate the sqe
	300	uint32_t idx = __atomic_fetch_add(&ring.submit_q.alloc, 1ul32, __ATOMIC_SEQ_CST);
	301
	302	// Validate that we didn't overflow anything
	303	// Check that nothing overflowed
	304	/* paranoid */ verify( true );
	305
	306	// Check that it goes head -> tail -> alloc and never head -> alloc -> tail
	307	/* paranoid */ verify( true );
	308
	309	// Return the sqe
	310	return [&ring.submit_q.sqes[ idx & (*ring.submit_q.mask)], idx];
	311	}
	312
	313	static inline void __submit( struct io_ring & ring, uint32_t idx ) {
	314	// get mutual exclusion
	315	lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
	316
	317	// Append to the list of ready entries
	318	uint32_t * tail = ring.submit_q.tail;
	319	const uint32_t mask = *ring.submit_q.mask;
	320
	321	ring.submit_q.array[ (*tail) & mask ] = idx & mask;
	322	__atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
	323
	324	// Submit however, many entries need to be submitted
	325	int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
[185efe6]	326	// __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_submit, returned %d\n", ret );
[2d8f7b0]	327	if( ret < 0 ) {
	328	switch((int)errno) {
	329	default:
	330	abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
	331	}
	332	}
	333
	334	unlock(ring.submit_q.lock);
	335	// Make sure that idx was submitted
	336	// Be careful to not get false positive if we cycled the entire list or that someone else submitted for us
	337	}
	338
	339	static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
	340	this.opcode = opcode;
	341	#if !defined(IOSQE_ASYNC)
	342	this.flags = 0;
	343	#else
	344	this.flags = IOSQE_ASYNC;
	345	#endif
	346	this.ioprio = 0;
	347	this.fd = fd;
	348	this.off = 0;
	349	this.addr = 0;
	350	this.len = 0;
	351	this.rw_flags = 0;
	352	this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0;
	353	}
	354
	355	static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
	356	(this){ opcode, fd };
	357	this.off = off;
	358	this.addr = (uint64_t)addr;
	359	this.len = len;
[92976d9]	360	}
	361
	362	//=============================================================================================
	363	// I/O Interface
	364	//=============================================================================================
	365	extern "C" {
	366	#define __USE_GNU
[2d8f7b0]	367	#define _GNU_SOURCE
	368	#include <fcntl.h>
[92976d9]	369	#include <sys/uio.h>
[2d8f7b0]	370	#include <sys/socket.h>
	371	#include <sys/stat.h>
[92976d9]	372	}
	373
[2d8f7b0]	374	#define __submit_prelude \
	375	struct io_ring & ring = active_cluster()->io; \
	376	struct io_uring_sqe * sqe; \
	377	uint32_t idx; \
	378	[sqe, idx] = __submit_alloc( ring );
	379
	380	#define __submit_wait \
	381	io_user_data data = { 0, active_thread() }; \
[185efe6]	382	/__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );/ \
[2d8f7b0]	383	sqe->user_data = (uint64_t)&data; \
	384	__submit( ring, idx ); \
	385	park( __cfaabi_dbg_ctx ); \
	386	return data.result;
	387
	388	//-----------------------------------------------------------------------------
	389	// Asynchronous operations
[92976d9]	390	ssize_t async_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
	391	#if !defined(IORING_OP_READV)
	392	return preadv2(fd, iov, iovcnt, offset, flags);
	393	#else
[2d8f7b0]	394	__submit_prelude
	395
	396	(*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
	397
	398	__submit_wait
[92976d9]	399	#endif
	400	}
	401
	402	ssize_t async_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) {
	403	#if !defined(IORING_OP_WRITEV)
	404	return pwritev2(fd, iov, iovcnt, offset, flags);
	405	#else
[2d8f7b0]	406	__submit_prelude
	407
	408	(*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
	409
	410	__submit_wait
	411	#endif
	412	}
	413
	414	int async_fsync(int fd) {
	415	#if !defined(IORING_OP_FSYNC)
	416	return fsync(fd);
	417	#else
	418	__submit_prelude
	419
	420	(*sqe){ IORING_OP_FSYNC, fd };
	421
	422	__submit_wait
[92976d9]	423	#endif
	424	}
	425
[2d8f7b0]	426	int async_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) {
	427	#if !defined(IORING_OP_SYNC_FILE_RANGE)
	428	return sync_file_range(fd, offset, nbytes, flags);
	429	#else
	430	__submit_prelude
	431
	432	(*sqe){ IORING_OP_SYNC_FILE_RANGE, fd };
	433	sqe->off = offset;
	434	sqe->len = nbytes;
	435	sqe->sync_range_flags = flags;
	436
	437	__submit_wait
	438	#endif
	439	}
	440
	441
	442	ssize_t async_sendmsg(int sockfd, const struct msghdr *msg, int flags) {
	443	#if !defined(IORING_OP_SENDMSG)
	444	return recv(sockfd, msg, flags);
	445	#else
	446	__submit_prelude
	447
	448	(*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 };
	449	sqe->msg_flags = flags;
	450
	451	__submit_wait
	452	#endif
	453	}
	454
	455	ssize_t async_recvmsg(int sockfd, struct msghdr *msg, int flags) {
	456	#if !defined(IORING_OP_RECVMSG)
	457	return recv(sockfd, msg, flags);
	458	#else
	459	__submit_prelude
	460
	461	(*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 };
	462	sqe->msg_flags = flags;
	463
	464	__submit_wait
	465	#endif
	466	}
	467
	468	ssize_t async_send(int sockfd, const void *buf, size_t len, int flags) {
	469	#if !defined(IORING_OP_SEND)
	470	return send( sockfd, buf, len, flags );
	471	#else
	472	__submit_prelude
	473
	474	(*sqe){ IORING_OP_SEND, sockfd };
	475	sqe->addr = (uint64_t)buf;
	476	sqe->len = len;
	477	sqe->msg_flags = flags;
	478
	479	__submit_wait
	480	#endif
	481	}
	482
	483	ssize_t async_recv(int sockfd, void *buf, size_t len, int flags) {
	484	#if !defined(IORING_OP_RECV)
	485	return recv( sockfd, buf, len, flags );
	486	#else
	487	__submit_prelude
	488
	489	(*sqe){ IORING_OP_RECV, sockfd };
	490	sqe->addr = (uint64_t)buf;
	491	sqe->len = len;
	492	sqe->msg_flags = flags;
	493
	494	__submit_wait
	495	#endif
	496	}
	497
	498	int async_accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags) {
	499	#if !defined(IORING_OP_ACCEPT)
	500	__SOCKADDR_ARG _addr;
	501	_addr.__sockaddr__ = addr;
	502	return accept4( sockfd, _addr, addrlen, flags );
	503	#else
	504	__submit_prelude
	505
	506	(*sqe){ IORING_OP_ACCEPT, sockfd };
	507	sqe->addr = addr;
	508	sqe->addr2 = addrlen;
	509	sqe->accept_flags = flags;
	510
	511	__submit_wait
	512	#endif
	513	}
	514
	515	int async_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) {
	516	#if !defined(IORING_OP_CONNECT)
	517	__CONST_SOCKADDR_ARG _addr;
	518	_addr.__sockaddr__ = addr;
	519	return connect( sockfd, _addr, addrlen );
	520	#else
	521	__submit_prelude
	522
	523	(*sqe){ IORING_OP_CONNECT, sockfd };
	524	sqe->addr = (uint64_t)addr;
	525	sqe->off = addrlen;
	526
	527	__submit_wait
	528	#endif
	529	}
	530
	531	int async_fallocate(int fd, int mode, uint64_t offset, uint64_t len) {
	532	#if !defined(IORING_OP_FALLOCATE)
	533	return fallocate( fd, mode, offset, len );
	534	#else
	535	__submit_prelude
	536
	537	(*sqe){ IORING_OP_FALLOCATE, fd };
	538	sqe->off = offset;
	539	sqe->len = length;
	540	sqe->mode = mode;
	541
	542	__submit_wait
	543	#endif
	544	}
	545
	546	int async_fadvise(int fd, uint64_t offset, uint64_t len, int advice) {
	547	#if !defined(IORING_OP_FADVISE)
	548	return posix_fadvise( fd, offset, len, advice );
	549	#else
	550	__submit_prelude
	551
	552	(*sqe){ IORING_OP_FADVISE, fd };
	553	sqe->off = (uint64_t)offset;
	554	sqe->len = length;
	555	sqe->fadvise_advice = advice;
	556
	557	__submit_wait
	558	#endif
	559	}
	560
	561	int async_madvise(void *addr, size_t length, int advice) {
	562	#if !defined(IORING_OP_MADVISE)
	563	return madvise( addr, length, advice );
	564	#else
	565	__submit_prelude
	566
	567	(*sqe){ IORING_OP_MADVISE, 0 };
	568	sqe->addr = (uint64_t)addr;
	569	sqe->len = length;
	570	sqe->fadvise_advice = advice;
	571
	572	__submit_wait
	573	#endif
	574	}
	575
	576	int async_openat(int dirfd, const char *pathname, int flags, mode_t mode) {
	577	#if !defined(IORING_OP_OPENAT)
	578	return openat( dirfd, pathname, flags, mode );
	579	#else
	580	__submit_prelude
	581
	582	(*sqe){ IORING_OP_OPENAT, dirfd };
	583	sqe->addr = (uint64_t)pathname;
	584	sqe->open_flags = flags;
	585	sqe->mode = mode;
	586
	587	__submit_wait
	588	#endif
	589	}
	590
	591	int async_close(int fd) {
	592	#if !defined(IORING_OP_CLOSE)
	593	return close( fd );
	594	#else
	595	__submit_prelude
	596
	597	(*sqe){ IORING_OP_CLOSE, fd };
	598
	599	__submit_wait
	600	#endif
	601	}
	602
	603	int async_statx(int dirfd, const char pathname, int flags, unsigned int mask, struct statx statxbuf) {
	604	#if !defined(IORING_OP_STATX)
[8962722]	605	//return statx( dirfd, pathname, flags, mask, statxbuf );
	606	return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf );
[2d8f7b0]	607	#else
	608	__submit_prelude
	609
	610	(*sqe){ IORING_OP_STATX, dirfd };
	611	sqe->addr = (uint64_t)pathname;
	612	sqe->statx_flags = flags;
	613	sqe->len = mask;
	614	sqe->off = (uint64_t)statxbuf;
	615
	616	__submit_wait
	617	#endif
	618	}
	619
	620
	621	ssize_t async_read(int fd, void *buf, size_t count) {
	622	#if !defined(IORING_OP_READ)
	623	return read( fd, buf, count );
	624	#else
	625	__submit_prelude
	626
	627	(*sqe){ IORING_OP_READ, fd, buf, count, 0 };
	628
	629	__submit_wait
	630	#endif
	631	}
	632
	633	ssize_t async_write(int fd, void *buf, size_t count) {
	634	#if !defined(IORING_OP_WRITE)
	635	return read( fd, buf, count );
	636	#else
	637	__submit_prelude
	638
	639	(*sqe){ IORING_OP_WRITE, fd, buf, count, 0 };
	640
	641	__submit_wait
	642	#endif
	643	}
	644
	645	//-----------------------------------------------------------------------------
	646	// Check if a function is asynchronous
	647
	648	// Macro magic to reduce the size of the following switch case
	649	#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__)
	650	#define IS_DEFINED_SECOND(first, second, ...) second
	651	#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion
	652	#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true)
	653
	654	bool is_async( fptr_t func ) {
	655
	656	if( /func == (fptr_t)preadv2 \|\| /
	657	func == (fptr_t)async_preadv2 )
	658	#define _CFA_IO_FEATURE_IORING_OP_READV ,
	659	return IS_DEFINED(IORING_OP_READV);
	660
	661	if( /func == (fptr_t)pwritev2 \|\| /
	662	func == (fptr_t)async_pwritev2 )
	663	#define _CFA_IO_FEATURE_IORING_OP_WRITEV ,
	664	return IS_DEFINED(IORING_OP_WRITEV);
	665
	666	if( /func == (fptr_t)fsync \|\| /
	667	func == (fptr_t)async_fsync )
	668	#define _CFA_IO_FEATURE_IORING_OP_FSYNC ,
	669	return IS_DEFINED(IORING_OP_FSYNC);
	670
	671	if( /func == (fptr_t)ync_file_range \|\| /
	672	func == (fptr_t)async_sync_file_range )
	673	#define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE ,
	674	return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE);
	675
	676	if( /func == (fptr_t)sendmsg \|\| /
	677	func == (fptr_t)async_sendmsg )
	678	#define _CFA_IO_FEATURE_IORING_OP_SENDMSG ,
	679	return IS_DEFINED(IORING_OP_SENDMSG);
	680
	681	if( /func == (fptr_t)recvmsg \|\| /
	682	func == (fptr_t)async_recvmsg )
	683	#define _CFA_IO_FEATURE_IORING_OP_RECVMSG ,
	684	return IS_DEFINED(IORING_OP_RECVMSG);
	685
	686	if( /func == (fptr_t)send \|\| /
	687	func == (fptr_t)async_send )
	688	#define _CFA_IO_FEATURE_IORING_OP_SEND ,
	689	return IS_DEFINED(IORING_OP_SEND);
	690
	691	if( /func == (fptr_t)recv \|\| /
	692	func == (fptr_t)async_recv )
	693	#define _CFA_IO_FEATURE_IORING_OP_RECV ,
	694	return IS_DEFINED(IORING_OP_RECV);
	695
	696	if( /func == (fptr_t)accept4 \|\| /
	697	func == (fptr_t)async_accept4 )
	698	#define _CFA_IO_FEATURE_IORING_OP_ACCEPT ,
	699	return IS_DEFINED(IORING_OP_ACCEPT);
	700
	701	if( /func == (fptr_t)connect \|\| /
	702	func == (fptr_t)async_connect )
	703	#define _CFA_IO_FEATURE_IORING_OP_CONNECT ,
	704	return IS_DEFINED(IORING_OP_CONNECT);
	705
	706	if( /func == (fptr_t)fallocate \|\| /
	707	func == (fptr_t)async_fallocate )
	708	#define _CFA_IO_FEATURE_IORING_OP_FALLOCATE ,
	709	return IS_DEFINED(IORING_OP_FALLOCATE);
	710
	711	if( /func == (fptr_t)fadvise \|\| /
	712	func == (fptr_t)async_fadvise )
	713	#define _CFA_IO_FEATURE_IORING_OP_FADVISE ,
	714	return IS_DEFINED(IORING_OP_FADVISE);
	715
	716	if( /func == (fptr_t)madvise \|\| /
	717	func == (fptr_t)async_madvise )
	718	#define _CFA_IO_FEATURE_IORING_OP_MADVISE ,
	719	return IS_DEFINED(IORING_OP_MADVISE);
	720
	721	if( /func == (fptr_t)openat \|\| /
	722	func == (fptr_t)async_openat )
	723	#define _CFA_IO_FEATURE_IORING_OP_OPENAT ,
	724	return IS_DEFINED(IORING_OP_OPENAT);
	725
	726	if( /func == (fptr_t)close \|\| /
	727	func == (fptr_t)async_close )
	728	#define _CFA_IO_FEATURE_IORING_OP_CLOSE ,
	729	return IS_DEFINED(IORING_OP_CLOSE);
	730
	731	if( /func == (fptr_t)statx \|\| /
	732	func == (fptr_t)async_statx )
	733	#define _CFA_IO_FEATURE_IORING_OP_STATX ,
	734	return IS_DEFINED(IORING_OP_STATX);
	735
	736	if( /func == (fptr_t)read \|\| /
	737	func == (fptr_t)async_read )
	738	#define _CFA_IO_FEATURE_IORING_OP_READ ,
	739	return IS_DEFINED(IORING_OP_READ);
	740
	741	if( /func == (fptr_t)write \|\| /
	742	func == (fptr_t)async_write )
	743	#define _CFA_IO_FEATURE_IORING_OP_WRITE ,
	744	return IS_DEFINED(IORING_OP_WRITE);
	745
	746	return false;
[92976d9]	747	}
	748
	749	#endif

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

Download in other formats:

Original Format