Context Navigation

io.cfa@ 06bdba4

Visit:

ADT ast-experimental pthread-emulation qualifiedEnum

Last change on this file since 06bdba4 was 7affcda, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
Fixed a bug where io wouldn't drain everything if the completion queue is filled
Property mode set to `100644`
File size: 21.6 KB

Rev	Line
[ecf6b46]	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
[3e2b9c9]	16	#define __cforall_thread__
[43784ac]	17	#define _GNU_SOURCE
[3e2b9c9]	18
[20ab637]	19	#if defined(__CFA_DEBUG__)
[d60d30e]	20	// #define __CFA_DEBUG_PRINT_IO__
	21	// #define __CFA_DEBUG_PRINT_IO_CORE__
[20ab637]	22	#endif
[4069faad]	23
[f6660520]	24
[3e2b9c9]	25	#if defined(CFA_HAVE_LINUX_IO_URING_H)
[31bb2e1]	26	#include <errno.h>
[3e2b9c9]	27	#include <signal.h>
[31bb2e1]	28	#include <stdint.h>
	29	#include <string.h>
	30	#include <unistd.h>
	31
[92976d9]	32	extern "C" {
	33	#include <sys/syscall.h>
[dddb3dd0]	34	#include <sys/eventfd.h>
[d3605f8]	35	#include <sys/uio.h>
[92976d9]	36
	37	#include <linux/io_uring.h>
	38	}
	39
[3e2b9c9]	40	#include "stats.hfa"
	41	#include "kernel.hfa"
	42	#include "kernel/fwd.hfa"
[708ae38]	43	#include "kernel/private.hfa"
[78a580d]	44	#include "kernel/cluster.hfa"
[3e2b9c9]	45	#include "io/types.hfa"
[185efe6]	46
[2fab24e3]	47	__attribute__((unused)) static const char * opcodes[] = {
[426f60c]	48	"OP_NOP",
	49	"OP_READV",
	50	"OP_WRITEV",
	51	"OP_FSYNC",
	52	"OP_READ_FIXED",
	53	"OP_WRITE_FIXED",
	54	"OP_POLL_ADD",
	55	"OP_POLL_REMOVE",
	56	"OP_SYNC_FILE_RANGE",
	57	"OP_SENDMSG",
	58	"OP_RECVMSG",
	59	"OP_TIMEOUT",
	60	"OP_TIMEOUT_REMOVE",
	61	"OP_ACCEPT",
	62	"OP_ASYNC_CANCEL",
	63	"OP_LINK_TIMEOUT",
	64	"OP_CONNECT",
	65	"OP_FALLOCATE",
	66	"OP_OPENAT",
	67	"OP_CLOSE",
	68	"OP_FILES_UPDATE",
	69	"OP_STATX",
	70	"OP_READ",
	71	"OP_WRITE",
	72	"OP_FADVISE",
	73	"OP_MADVISE",
	74	"OP_SEND",
	75	"OP_RECV",
	76	"OP_OPENAT2",
	77	"OP_EPOLL_CTL",
	78	"OP_SPLICE",
	79	"OP_PROVIDE_BUFFERS",
	80	"OP_REMOVE_BUFFERS",
	81	"OP_TEE",
	82	"INVALID_OP"
	83	};
	84
[11054eb]	85	static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want );
	86	static void __ioarbiter_submit( $io_context * , __u32 idxs[], __u32 have, bool lazy );
	87	static void __ioarbiter_flush ( $io_context & );
[dddb3dd0]	88	static inline void __ioarbiter_notify( $io_context & ctx );
[92976d9]	89	//=============================================================================================
	90	// I/O Polling
	91	//=============================================================================================
[78da4ab]	92	static inline unsigned __flush( struct $io_context & );
	93	static inline __u32 __release_sqes( struct $io_context & );
[24e321c]	94	extern void __kernel_unpark( thread$ * thrd, unpark_hint );
[1d5e4711]	95
[18f7858]	96	static void ioring_syscsll( struct $io_context & ctx, unsigned int min_comp, unsigned int flags ) {
	97	__STATS__( true, io.calls.flush++; )
[bdfd0bd]	98	int ret;
	99	for() {
	100	ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, flags, (sigset_t *)0p, _NSIG / 8);
	101	if( ret < 0 ) {
	102	switch((int)errno) {
	103	case EINTR:
	104	continue;
	105	case EAGAIN:
	106	case EBUSY:
	107	// Update statistics
	108	__STATS__( false, io.calls.errors.busy ++; )
	109	return false;
	110	default:
	111	abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
	112	}
[18f7858]	113	}
[bdfd0bd]	114	break;
[18f7858]	115	}
	116
	117	__cfadbg_print_safe(io, "Kernel I/O : %u submitted to io_uring %d\n", ret, ctx.fd);
	118	__STATS__( true, io.calls.submitted += ret; )
	119	/* paranoid / verify( ctx.sq.to_submit <= ctx.sq.num );
	120	/* paranoid */ verify( ctx.sq.to_submit >= ret );
	121
	122	ctx.sq.to_submit -= ret;
	123
	124	/* paranoid / verify( ctx.sq.to_submit <= ctx.sq.num );
	125
	126	// Release the consumed SQEs
	127	__release_sqes( ctx );
	128
[dddb3dd0]	129	/* paranoid */ verify( ! __preemption_enabled() );
[6f121b8]	130
[18f7858]	131	__atomic_store_n(&ctx.proc->io.pending, false, __ATOMIC_RELAXED);
	132	}
	133
	134	static bool try_acquire( $io_context * ctx ) __attribute__((nonnull(1))) {
	135	/* paranoid */ verify( ! __preemption_enabled() );
	136	/* paranoid */ verify( ready_schedule_islocked() );
[92976d9]	137
[d60d30e]	138
[3caf5e3]	139	{
	140	const __u32 head = *ctx->cq.head;
	141	const __u32 tail = *ctx->cq.tail;
	142
	143	if(head == tail) return false;
	144	}
[c1c95b1]	145
[3caf5e3]	146	// Drain the queue
[4ecc35a]	147	if(!__atomic_try_acquire(&ctx->cq.lock)) {
[54c1196]	148	__STATS__( false, io.calls.locked++; )
[4ecc35a]	149	return false;
	150	}
	151
[18f7858]	152	return true;
	153	}
	154
	155	static bool __cfa_do_drain( $io_context * ctx, cluster * cltr ) __attribute__((nonnull(1, 2))) {
	156	/* paranoid */ verify( ! __preemption_enabled() );
	157	/* paranoid */ verify( ready_schedule_islocked() );
	158	/* paranoid */ verify( ctx->cq.lock == true );
	159
	160	const __u32 mask = *ctx->cq.mask;
[7affcda]	161	const __u32 num = *ctx->cq.num;
[78a580d]	162	unsigned long long ts_prev = ctx->cq.ts;
[7affcda]	163	unsigned long long ts_next;
[78a580d]	164
[7affcda]	165	// We might need to do this multiple times if more events completed than can fit in the queue.
	166	for() {
	167	// re-read the head and tail in case it already changed.
	168	const __u32 head = *ctx->cq.head;
	169	const __u32 tail = *ctx->cq.tail;
	170	const __u32 count = tail - head;
	171	__STATS__( false, io.calls.drain++; io.calls.completed += count; )
[3caf5e3]	172
[7affcda]	173	for(i; count) {
	174	unsigned idx = (head + i) & mask;
	175	volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
[92976d9]	176
[7affcda]	177	/* paranoid */ verify(&cqe);
[92976d9]	178
[7affcda]	179	struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
	180	// __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
[78da4ab]	181
[7affcda]	182	__kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL );
	183	}
	184
	185	ts_next = ctx->cq.ts = rdtscl();
[78da4ab]	186
[7affcda]	187	// Mark to the kernel that the cqe has been seen
	188	// Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
	189	__atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
	190	ctx->proc->idle_wctx.drain_time = ts_next;
[2d8f7b0]	191
[7affcda]	192	if(likely(count < num)) break;
	193
	194	ioring_syscsll( *ctx, 0, IORING_ENTER_GETEVENTS);
	195	}
[92976d9]	196
[1e6ffb44]	197	__cfadbg_print_safe(io, "Kernel I/O : %u completed age %llu\n", count, ts_next);
[e9c0b4c]	198	/* paranoid */ verify( ready_schedule_islocked() );
[dddb3dd0]	199	/* paranoid */ verify( ! __preemption_enabled() );
	200
[4ecc35a]	201	__atomic_unlock(&ctx->cq.lock);
	202
[78a580d]	203	touch_tsc( cltr->sched.io.tscs, ctx->cq.id, ts_prev, ts_next );
	204
[c1c95b1]	205	return true;
[92976d9]	206	}
	207
[4479890]	208	bool __cfa_io_drain( processor * proc ) {
	209	bool local = false;
	210	bool remote = false;
	211
[18f7858]	212	ready_schedule_lock();
	213
[4479890]	214	cluster * const cltr = proc->cltr;
	215	$io_context * const ctx = proc->io.ctx;
	216	/* paranoid */ verify( cltr );
	217	/* paranoid */ verify( ctx );
	218
	219	with(cltr->sched) {
	220	const size_t ctxs_count = io.count;
	221
	222	/* paranoid */ verify( ready_schedule_islocked() );
	223	/* paranoid */ verify( ! __preemption_enabled() );
	224	/* paranoid */ verify( active_processor() == proc );
	225	/* paranoid */ verify( __shard_factor.io > 0 );
	226	/* paranoid */ verify( ctxs_count > 0 );
	227	/* paranoid */ verify( ctx->cq.id < ctxs_count );
	228
	229	const unsigned this_cache = cache_id(cltr, ctx->cq.id / __shard_factor.io);
	230	const unsigned long long ctsc = rdtscl();
	231
[b035046]	232	if(proc->io.target == UINT_MAX) {
[4479890]	233	uint64_t chaos = __tls_rand();
	234	unsigned ext = chaos & 0xff;
	235	unsigned other = (chaos >> 8) % (ctxs_count);
	236
	237	if(ext < 3 \|\| __atomic_load_n(&caches[other / __shard_factor.io].id, __ATOMIC_RELAXED) == this_cache) {
	238	proc->io.target = other;
	239	}
	240	}
	241	else {
	242	const unsigned target = proc->io.target;
[b035046]	243	/* paranoid */ verify( io.tscs[target].tv != ULLONG_MAX );
[18f7858]	244	HELP: if(target < ctxs_count) {
[4479890]	245	const unsigned long long cutoff = calc_cutoff(ctsc, ctx->cq.id, ctxs_count, io.data, io.tscs, __shard_factor.io);
	246	const unsigned long long age = moving_average(ctsc, io.tscs[target].tv, io.tscs[target].ma);
[edf247b]	247	__cfadbg_print_safe(io, "Kernel I/O: Help attempt on %u from %u, age %'llu vs cutoff %'llu, %s\n", target, ctx->cq.id, age, cutoff, age > cutoff ? "yes" : "no");
[18f7858]	248	if(age <= cutoff) break HELP;
	249
	250	if(!try_acquire(io.data[target])) break HELP;
	251
	252	if(!__cfa_do_drain( io.data[target], cltr )) break HELP;
	253
	254	remote = true;
[8a5e357]	255	__STATS__( true, io.calls.helped++; )
[4479890]	256	}
[b035046]	257	proc->io.target = UINT_MAX;
[4479890]	258	}
	259	}
	260
	261
	262	// Drain the local queue
[18f7858]	263	if(try_acquire( proc->io.ctx )) {
	264	local = __cfa_do_drain( proc->io.ctx, cltr );
	265	}
[4479890]	266
	267	/* paranoid */ verify( ready_schedule_islocked() );
	268	/* paranoid */ verify( ! __preemption_enabled() );
	269	/* paranoid */ verify( active_processor() == proc );
[18f7858]	270
	271	ready_schedule_unlock();
[4479890]	272	return local \|\| remote;
	273	}
	274
[18f7858]	275	bool __cfa_io_flush( processor * proc ) {
[dddb3dd0]	276	/* paranoid */ verify( ! __preemption_enabled() );
	277	/* paranoid */ verify( proc );
	278	/* paranoid */ verify( proc->io.ctx );
[1539bbd]	279
[dddb3dd0]	280	$io_context & ctx = *proc->io.ctx;
[78da4ab]	281
[11054eb]	282	__ioarbiter_flush( ctx );
[3c039b0]	283
[18f7858]	284	if(ctx.sq.to_submit != 0) {
	285	ioring_syscsll(ctx, 0, 0);
[21a5bfb7]	286
	287	}
[61dd73d]	288
[18f7858]	289	return __cfa_io_drain( proc );
[61dd73d]	290	}
[f6660520]	291
[92976d9]	292	//=============================================================================================
	293	// I/O Submissions
	294	//=============================================================================================
	295
[2d8f7b0]	296	// Submition steps :
[e46c753]	297	// 1 - Allocate a queue entry. The ring already has memory for all entries but only the ones
[2d8f7b0]	298	// listed in sq.array are visible by the kernel. For those not listed, the kernel does not
	299	// offer any assurance that an entry is not being filled by multiple flags. Therefore, we
	300	// need to write an allocator that allows allocating concurrently.
	301	//
[e46c753]	302	// 2 - Actually fill the submit entry, this is the only simple and straightforward step.
[2d8f7b0]	303	//
[e46c753]	304	// 3 - Append the entry index to the array and adjust the tail accordingly. This operation
[2d8f7b0]	305	// needs to arrive to two concensus at the same time:
	306	// A - The order in which entries are listed in the array: no two threads must pick the
	307	// same index for their entries
	308	// B - When can the tail be update for the kernel. EVERY entries in the array between
	309	// head and tail must be fully filled and shouldn't ever be touched again.
	310	//
[78da4ab]	311	//=============================================================================================
	312	// Allocation
	313	// for user's convenience fill the sqes from the indexes
	314	static inline void __fill(struct io_uring_sqe * out_sqes[], __u32 want, __u32 idxs[], struct $io_context * ctx) {
	315	struct io_uring_sqe * sqes = ctx->sq.sqes;
	316	for(i; want) {
[1e6ffb44]	317	// __cfadbg_print_safe(io, "Kernel I/O : filling loop\n");
[78da4ab]	318	out_sqes[i] = &sqes[idxs[i]];
	319	}
	320	}
[2489d31]	321
[78da4ab]	322	// Try to directly allocate from the a given context
	323	// Not thread-safe
	324	static inline bool __alloc(struct $io_context * ctx, __u32 idxs[], __u32 want) {
	325	__sub_ring_t & sq = ctx->sq;
	326	const __u32 mask = *sq.mask;
	327	__u32 fhead = sq.free_ring.head; // get the current head of the queue
	328	__u32 ftail = sq.free_ring.tail; // get the current tail of the queue
[2489d31]	329
[78da4ab]	330	// If we don't have enough sqes, fail
	331	if((ftail - fhead) < want) { return false; }
[426f60c]	332
[78da4ab]	333	// copy all the indexes we want from the available list
	334	for(i; want) {
[1e6ffb44]	335	// __cfadbg_print_safe(io, "Kernel I/O : allocating loop\n");
[78da4ab]	336	idxs[i] = sq.free_ring.array[(fhead + i) & mask];
[6f121b8]	337	}
[2489d31]	338
[78da4ab]	339	// Advance the head to mark the indexes as consumed
	340	__atomic_store_n(&sq.free_ring.head, fhead + want, __ATOMIC_RELEASE);
[df40a56]	341
[78da4ab]	342	// return success
	343	return true;
	344	}
[df40a56]	345
[78da4ab]	346	// Allocate an submit queue entry.
	347	// The kernel cannot see these entries until they are submitted, but other threads must be
	348	// able to see which entries can be used and which are already un used by an other thread
	349	// for convenience, return both the index and the pointer to the sqe
	350	// sqe == &sqes[idx]
[108345a]	351	struct $io_context * cfa_io_allocate(struct io_uring_sqe * sqes[], __u32 idxs[], __u32 want) libcfa_public {
[1e6ffb44]	352	// __cfadbg_print_safe(io, "Kernel I/O : attempting to allocate %u\n", want);
[df40a56]	353
[78da4ab]	354	disable_interrupts();
	355	processor * proc = __cfaabi_tls.this_processor;
[dddb3dd0]	356	$io_context * ctx = proc->io.ctx;
[78da4ab]	357	/* paranoid */ verify( __cfaabi_tls.this_processor );
[dddb3dd0]	358	/* paranoid */ verify( ctx );
[78da4ab]	359
[1e6ffb44]	360	// __cfadbg_print_safe(io, "Kernel I/O : attempting to fast allocation\n");
[78da4ab]	361
[dddb3dd0]	362	// We can proceed to the fast path
	363	if( __alloc(ctx, idxs, want) ) {
	364	// Allocation was successful
	365	__STATS__( true, io.alloc.fast += 1; )
[a3821fa]	366	enable_interrupts();
[df40a56]	367
[1e6ffb44]	368	// __cfadbg_print_safe(io, "Kernel I/O : fast allocation successful from ring %d\n", ctx->fd);
[2fafe7e]	369
[dddb3dd0]	370	__fill( sqes, want, idxs, ctx );
	371	return ctx;
[df40a56]	372	}
[dddb3dd0]	373	// The fast path failed, fallback
	374	__STATS__( true, io.alloc.fail += 1; )
[df40a56]	375
[78da4ab]	376	// Fast path failed, fallback on arbitration
[d60d30e]	377	__STATS__( true, io.alloc.slow += 1; )
[a3821fa]	378	enable_interrupts();
[78da4ab]	379
[dddb3dd0]	380	$io_arbiter * ioarb = proc->cltr->io.arbiter;
	381	/* paranoid */ verify( ioarb );
	382
[1e6ffb44]	383	// __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for allocation\n");
[78da4ab]	384
[11054eb]	385	struct $io_context * ret = __ioarbiter_allocate(*ioarb, idxs, want);
[78da4ab]	386
[1e6ffb44]	387	// __cfadbg_print_safe(io, "Kernel I/O : slow allocation completed from ring %d\n", ret->fd);
[df40a56]	388
[78da4ab]	389	__fill( sqes, want, idxs,ret );
	390	return ret;
[df40a56]	391	}
	392
[78da4ab]	393	//=============================================================================================
	394	// submission
[2432e8e]	395	static inline void __submit_only( struct $io_context * ctx, __u32 idxs[], __u32 have) {
[78da4ab]	396	// We can proceed to the fast path
	397	// Get the right objects
	398	__sub_ring_t & sq = ctx->sq;
	399	const __u32 mask = *sq.mask;
[dddb3dd0]	400	__u32 tail = *sq.kring.tail;
[78da4ab]	401
	402	// Add the sqes to the array
	403	for( i; have ) {
[1e6ffb44]	404	// __cfadbg_print_safe(io, "Kernel I/O : __submit loop\n");
[78da4ab]	405	sq.kring.array[ (tail + i) & mask ] = idxs[i];
[426f60c]	406	}
	407
[78da4ab]	408	// Make the sqes visible to the submitter
[dddb3dd0]	409	__atomic_store_n(sq.kring.tail, tail + have, __ATOMIC_RELEASE);
[e8ac228]	410	sq.to_submit += have;
[426f60c]	411
[d529ad0]	412	__atomic_store_n(&ctx->proc->io.pending, true, __ATOMIC_RELAXED);
	413	__atomic_store_n(&ctx->proc->io.dirty , true, __ATOMIC_RELAXED);
[2432e8e]	414	}
	415
	416	static inline void __submit( struct $io_context * ctx, __u32 idxs[], __u32 have, bool lazy) {
	417	__sub_ring_t & sq = ctx->sq;
	418	__submit_only(ctx, idxs, have);
	419
[70b4aeb9]	420	if(sq.to_submit > 30) {
	421	__tls_stats()->io.flush.full++;
[18f7858]	422	__cfa_io_flush( ctx->proc );
[70b4aeb9]	423	}
	424	if(!lazy) {
	425	__tls_stats()->io.flush.eager++;
[18f7858]	426	__cfa_io_flush( ctx->proc );
[dddb3dd0]	427	}
[78da4ab]	428	}
[2489d31]	429
[108345a]	430	void cfa_io_submit( struct $io_context * inctx, __u32 idxs[], __u32 have, bool lazy ) __attribute__((nonnull (1))) libcfa_public {
[1e6ffb44]	431	// __cfadbg_print_safe(io, "Kernel I/O : attempting to submit %u (%s)\n", have, lazy ? "lazy" : "eager");
[5dadc9b7]	432
[78da4ab]	433	disable_interrupts();
	434	processor * proc = __cfaabi_tls.this_processor;
	435	$io_context * ctx = proc->io.ctx;
[dddb3dd0]	436	/* paranoid */ verify( __cfaabi_tls.this_processor );
	437	/* paranoid */ verify( ctx );
[e46c753]	438
[78da4ab]	439	// Can we proceed to the fast path
[dddb3dd0]	440	if( ctx == inctx ) // We have the right instance?
[78da4ab]	441	{
[dddb3dd0]	442	__submit(ctx, idxs, have, lazy);
[e46c753]	443
[78da4ab]	444	// Mark the instance as no longer in-use, re-enable interrupts and return
[d60d30e]	445	__STATS__( true, io.submit.fast += 1; )
[a3821fa]	446	enable_interrupts();
[ece0e80]	447
[1e6ffb44]	448	// __cfadbg_print_safe(io, "Kernel I/O : submitted on fast path\n");
[78da4ab]	449	return;
[e46c753]	450	}
[d384787]	451
[78da4ab]	452	// Fast path failed, fallback on arbitration
[d60d30e]	453	__STATS__( true, io.submit.slow += 1; )
[a3821fa]	454	enable_interrupts();
[5dadc9b7]	455
[1e6ffb44]	456	// __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for submission\n");
[426f60c]	457
[11054eb]	458	__ioarbiter_submit(inctx, idxs, have, lazy);
[78da4ab]	459	}
[2fab24e3]	460
[78da4ab]	461	//=============================================================================================
	462	// Flushing
[426f60c]	463	// Go through the ring's submit queue and release everything that has already been consumed
	464	// by io_uring
[78da4ab]	465	// This cannot be done by multiple threads
	466	static __u32 __release_sqes( struct $io_context & ctx ) {
	467	const __u32 mask = *ctx.sq.mask;
[732b406]	468
[426f60c]	469	__attribute__((unused))
[78da4ab]	470	__u32 ctail = *ctx.sq.kring.tail; // get the current tail of the queue
	471	__u32 chead = *ctx.sq.kring.head; // get the current head of the queue
	472	__u32 phead = ctx.sq.kring.released; // get the head the last time we were here
	473
	474	__u32 ftail = ctx.sq.free_ring.tail; // get the current tail of the queue
[732b406]	475
[426f60c]	476	// the 3 fields are organized like this diagram
	477	// except it's are ring
	478	// ---+--------+--------+----
	479	// ---+--------+--------+----
	480	// ^ ^ ^
	481	// phead chead ctail
	482
	483	// make sure ctail doesn't wrap around and reach phead
	484	/* paranoid */ verify(
	485	(ctail >= chead && chead >= phead)
	486	\|\| (chead >= phead && phead >= ctail)
	487	\|\| (phead >= ctail && ctail >= chead)
	488	);
	489
	490	// find the range we need to clear
[4998155]	491	__u32 count = chead - phead;
[426f60c]	492
[78da4ab]	493	if(count == 0) {
	494	return 0;
	495	}
	496
[426f60c]	497	// We acquired an previous-head/current-head range
	498	// go through the range and release the sqes
[34b61882]	499	for( i; count ) {
[1e6ffb44]	500	// __cfadbg_print_safe(io, "Kernel I/O : release loop\n");
[78da4ab]	501	__u32 idx = ctx.sq.kring.array[ (phead + i) & mask ];
	502	ctx.sq.free_ring.array[ (ftail + i) & mask ] = idx;
[34b61882]	503	}
[78da4ab]	504
	505	ctx.sq.kring.released = chead; // note up to were we processed
	506	__atomic_store_n(&ctx.sq.free_ring.tail, ftail + count, __ATOMIC_SEQ_CST);
	507
	508	__ioarbiter_notify(ctx);
	509
[34b61882]	510	return count;
	511	}
[35285fd]	512
[78da4ab]	513	//=============================================================================================
	514	// I/O Arbiter
	515	//=============================================================================================
[9f5a71eb]	516	static inline bool enqueue(__outstanding_io_queue & queue, __outstanding_io & item) {
	517	bool was_empty;
	518
[11054eb]	519	// Lock the list, it's not thread safe
	520	lock( queue.lock __cfaabi_dbg_ctx2 );
	521	{
[9f5a71eb]	522	was_empty = empty(queue.queue);
	523
[11054eb]	524	// Add our request to the list
	525	add( queue.queue, item );
	526
	527	// Mark as pending
	528	__atomic_store_n( &queue.empty, false, __ATOMIC_SEQ_CST );
	529	}
	530	unlock( queue.lock );
	531
[9f5a71eb]	532	return was_empty;
[11054eb]	533	}
	534
	535	static inline bool empty(__outstanding_io_queue & queue ) {
	536	return __atomic_load_n( &queue.empty, __ATOMIC_SEQ_CST);
	537	}
	538
	539	static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want ) {
[1e6ffb44]	540	// __cfadbg_print_safe(io, "Kernel I/O : arbiter allocating\n");
[78da4ab]	541
[d60d30e]	542	__STATS__( false, io.alloc.block += 1; )
	543
[78da4ab]	544	// No one has any resources left, wait for something to finish
[11054eb]	545	// We need to add ourself to a list of pending allocs and wait for an answer
	546	__pending_alloc pa;
	547	pa.idxs = idxs;
	548	pa.want = want;
[78da4ab]	549
[9f5a71eb]	550	enqueue(this.pending, (__outstanding_io&)pa);
	551
	552	wait( pa.sem );
[78da4ab]	553
[11054eb]	554	return pa.ctx;
[dddb3dd0]	555
[78da4ab]	556	}
	557
[11054eb]	558	static void __ioarbiter_notify( $io_arbiter & this, $io_context * ctx ) {
	559	/* paranoid */ verify( !empty(this.pending.queue) );
[78da4ab]	560
[11054eb]	561	lock( this.pending.lock __cfaabi_dbg_ctx2 );
	562	{
	563	while( !empty(this.pending.queue) ) {
	564	__cfadbg_print_safe(io, "Kernel I/O : notifying\n");
	565	__u32 have = ctx->sq.free_ring.tail - ctx->sq.free_ring.head;
	566	__pending_alloc & pa = (__pending_alloc&)head( this.pending.queue );
[78da4ab]	567
[11054eb]	568	if( have > pa.want ) goto DONE;
	569	drop( this.pending.queue );
[78da4ab]	570
[11054eb]	571	/* paranoid */__attribute__((unused)) bool ret =
[78da4ab]	572
[11054eb]	573	__alloc(ctx, pa.idxs, pa.want);
	574
	575	/* paranoid */ verify( ret );
	576
	577	pa.ctx = ctx;
	578
	579	post( pa.sem );
	580	}
	581
	582	this.pending.empty = true;
	583	DONE:;
	584	}
	585	unlock( this.pending.lock );
[78da4ab]	586	}
	587
	588	static void __ioarbiter_notify( $io_context & ctx ) {
[11054eb]	589	if(!empty( ctx.arbiter->pending )) {
[78da4ab]	590	__ioarbiter_notify( *ctx.arbiter, &ctx );
	591	}
	592	}
	593
	594	// Simply append to the pending
[11054eb]	595	static void __ioarbiter_submit( $io_context * ctx, __u32 idxs[], __u32 have, bool lazy ) {
[78da4ab]	596	__cfadbg_print_safe(io, "Kernel I/O : submitting %u from the arbiter to context %u\n", have, ctx->fd);
	597
	598	__cfadbg_print_safe(io, "Kernel I/O : waiting to submit %u\n", have);
	599
[11054eb]	600	__external_io ei;
	601	ei.idxs = idxs;
	602	ei.have = have;
	603	ei.lazy = lazy;
[78da4ab]	604
[9f5a71eb]	605	bool we = enqueue(ctx->ext_sq, (__outstanding_io&)ei);
	606
[d529ad0]	607	__atomic_store_n(&ctx->proc->io.pending, true, __ATOMIC_SEQ_CST);
[9f5a71eb]	608
	609	if( we ) {
	610	sigval_t value = { PREEMPT_IO };
	611	pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
	612	}
	613
	614	wait( ei.sem );
[78da4ab]	615
	616	__cfadbg_print_safe(io, "Kernel I/O : %u submitted from arbiter\n", have);
	617	}
	618
[11054eb]	619	static void __ioarbiter_flush( $io_context & ctx ) {
	620	if(!empty( ctx.ext_sq )) {
	621	__STATS__( false, io.flush.external += 1; )
[78da4ab]	622
[11054eb]	623	__cfadbg_print_safe(io, "Kernel I/O : arbiter flushing\n");
[d60d30e]	624
[11054eb]	625	lock( ctx.ext_sq.lock __cfaabi_dbg_ctx2 );
	626	{
	627	while( !empty(ctx.ext_sq.queue) ) {
	628	__external_io & ei = (__external_io&)drop( ctx.ext_sq.queue );
[78da4ab]	629
[2432e8e]	630	__submit_only(&ctx, ei.idxs, ei.have);
[78da4ab]	631
[11054eb]	632	post( ei.sem );
	633	}
	634
	635	ctx.ext_sq.empty = true;
	636	}
	637	unlock(ctx.ext_sq.lock );
	638	}
[78da4ab]	639	}
[7ef162b2]	640
[d3605f8]	641	#if defined(CFA_WITH_IO_URING_IDLE)
	642	bool __kernel_read(processor * proc, io_future_t & future, iovec & iov, int fd) {
[6ddef36]	643	$io_context * ctx = proc->io.ctx;
	644	/* paranoid */ verify( ! __preemption_enabled() );
	645	/* paranoid */ verify( proc == __cfaabi_tls.this_processor );
	646	/* paranoid */ verify( ctx );
[7ef162b2]	647
[6ddef36]	648	__u32 idx;
	649	struct io_uring_sqe * sqe;
[7ef162b2]	650
[6ddef36]	651	// We can proceed to the fast path
[010636f]	652	if( !__alloc(ctx, &idx, 1) ) {
	653	/* paranoid */ verify( false ); // for now check if this happens, next time just abort the sleep.
	654	return false;
	655	}
[6ddef36]	656
	657	// Allocation was successful
	658	__fill( &sqe, 1, &idx, ctx );
	659
	660	sqe->user_data = (uintptr_t)&future;
	661	sqe->flags = 0;
[a1f3d93]	662	sqe->fd = fd;
[6ddef36]	663	sqe->off = 0;
[d3605f8]	664	sqe->ioprio = 0;
[6ddef36]	665	sqe->fsync_flags = 0;
	666	sqe->__pad2[0] = 0;
	667	sqe->__pad2[1] = 0;
	668	sqe->__pad2[2] = 0;
[d3605f8]	669
	670	#if defined(CFA_HAVE_IORING_OP_READ)
	671	sqe->opcode = IORING_OP_READ;
	672	sqe->addr = (uint64_t)iov.iov_base;
	673	sqe->len = iov.iov_len;
	674	#elif defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV)
	675	sqe->opcode = IORING_OP_READV;
	676	sqe->addr = (uintptr_t)&iov;
	677	sqe->len = 1;
	678	#else
	679	#error CFA_WITH_IO_URING_IDLE but none of CFA_HAVE_READV, CFA_HAVE_IORING_OP_READV or CFA_HAVE_IORING_OP_READ defined
	680	#endif
[6ddef36]	681
	682	asm volatile("": : :"memory");
	683
	684	/* paranoid */ verify( sqe->user_data == (uintptr_t)&future );
[010636f]	685	__submit_only( ctx, &idx, 1 );
[6ddef36]	686
	687	/* paranoid */ verify( proc == __cfaabi_tls.this_processor );
	688	/* paranoid */ verify( ! __preemption_enabled() );
[078fb05]	689
	690	return true;
[6ddef36]	691	}
[18f7858]	692
	693	void __cfa_io_idle( processor * proc ) {
	694	iovec iov;
	695	__atomic_acquire( &proc->io.ctx->cq.lock );
	696
[262fafd9]	697	__attribute__((used)) volatile bool was_reset = false;
	698
[d5cdbed]	699	with( proc->idle_wctx) {
[18f7858]	700
[37a3aa23]	701	// Do we already have a pending read
	702	if(available(*ftr)) {
	703	// There is no pending read, we need to add one
	704	reset(*ftr);
	705
	706	iov.iov_base = rdbuf;
	707	iov.iov_len = sizeof(eventfd_t);
	708	__kernel_read(proc, *ftr, iov, evfd );
[262fafd9]	709	ftr->result = 0xDEADDEAD;
	710	((eventfd_t )rdbuf) = 0xDEADDEADDEADDEAD;
	711	was_reset = true;
[37a3aa23]	712	}
[18f7858]	713	}
	714
[010636f]	715	if( !__atomic_load_n( &proc->do_terminate, __ATOMIC_SEQ_CST ) ) {
	716	__ioarbiter_flush( *proc->io.ctx );
[262fafd9]	717	proc->idle_wctx.sleep_time = rdtscl();
[010636f]	718	ioring_syscsll( *proc->io.ctx, 1, IORING_ENTER_GETEVENTS);
	719	}
[18f7858]	720
[d5cdbed]	721	ready_schedule_lock();
[18f7858]	722	__cfa_do_drain( proc->io.ctx, proc->cltr );
[d5cdbed]	723	ready_schedule_unlock();
[262fafd9]	724
	725	asm volatile ("" :: "m" (was_reset));
[18f7858]	726	}
[6ddef36]	727	#endif
[47746a2]	728	#endif

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

Download in other formats:

Original Format