Context Navigation

io.cfa @ 3c039b0

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

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

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

Download in other formats:

Original Format