Context Navigation

io.cfa@ ecf6b46

Visit:

ADT arm-eh ast-experimental enum forall-pointer-decay jacob/cs343-translation new-ast new-ast-unique-expr pthread-emulation qualifiedEnum stuck-waitfor-destruct

Last change on this file since ecf6b46 was ecf6b46, checked in by Thierry Delisle <tdelisle@…>, 6 years ago
some clean-up in io.cfa
Property mode set to `100644`
File size: 23.3 KB

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

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

Download in other formats:

Original Format