Context Navigation

io.cfa@ 6091b88a

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 6091b88a was 8962722, checked in by Thierry Delisle <tdelisle@…>, 5 years ago
Replace statx call with syscall since it is not on all machines
Property mode set to `100644`
File size: 22.6 KB

Line
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
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
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
75	uint32_t nentries = entries_per_cluster();
76
77	int fd = syscall(__NR_io_uring_setup, nentries, &params );
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));
84	struct io_uring_sq & sq = this.io.submit_q;
85	struct io_uring_cq & cq = this.io.completion_q;
86
87	// calculate the right ring size
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));
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
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) {
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
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);
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);
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);
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
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;
136
137	// completion queue
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 );
155
156	// Update the global ring info
157	this.io.flags = params.flags;
158	this.io.fd = fd;
159	this.io.done = false;
160	(this.io.submit){ min(sq.num, cq.num) };
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
182	munmap(sq.sqes, (sq.num) sizeof(struct io_uring_sqe));
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;
218	// __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd );
219
220	data->result = cqe.res;
221	__unpark( data->thrd __cfaabi_dbg_ctx2 );
222
223	// Allow new submissions to happen
224	V(ring.submit);
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.
228	__atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED );
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
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);
326	// __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_submit, returned %d\n", ret );
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;
360	}
361
362	//=============================================================================================
363	// I/O Interface
364	//=============================================================================================
365	extern "C" {
366	#define __USE_GNU
367	#define _GNU_SOURCE
368	#include <fcntl.h>
369	#include <sys/uio.h>
370	#include <sys/socket.h>
371	#include <sys/stat.h>
372	}
373
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() }; \
382	/__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );/ \
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
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
394	__submit_prelude
395
396	(*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset };
397
398	__submit_wait
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
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
423	#endif
424	}
425
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)
605	//return statx( dirfd, pathname, flags, mask, statxbuf );
606	return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf );
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;
747	}
748
749	#endif

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

Download in other formats:

Original Format