Context Navigation

io.cfa @ 1539bbd

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

Last change on this file since 1539bbd was 1539bbd, checked in by Thierry Delisle <tdelisle@…>, 4 years ago
Added some debugging prints for fast poller
Property mode set to `100644`
File size: 29.2 KB

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

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

Download in other formats:

Original Format