Context Navigation

io.cfa@ fa2e183

Visit:

ADT ast-experimental

Last change on this file since fa2e183 was df6cc9d, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
Merge branch 'master' into pthread-emulation
Property mode set to `100644`
File size: 21.8 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
[8bee858]	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$ & );
	88	static inline void __ioarbiter_notify( io_context$ & ctx );
[92976d9]	89	//=============================================================================================
	90	// I/O Polling
	91	//=============================================================================================
[8bee858]	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
[8bee858]	96	static void ioring_syscsll( struct io_context$ & ctx, unsigned int min_comp, unsigned int flags ) {
[18f7858]	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
[8bee858]	134	static bool try_acquire( io_context$ * ctx ) __attribute__((nonnull(1))) {
[18f7858]	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
[8bee858]	155	static bool __cfa_do_drain( io_context$ * ctx, cluster * cltr ) __attribute__((nonnull(1, 2))) {
[18f7858]	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
[5f9c42b]	203	touch_tsc( cltr->sched.io.tscs, ctx->cq.id, ts_prev, ts_next, false );
[78a580d]	204
[c1c95b1]	205	return true;
[92976d9]	206	}
	207
[1756e08]	208	bool __cfa_io_drain( struct processor * proc ) {
[4479890]	209	bool local = false;
	210	bool remote = false;
	211
[18f7858]	212	ready_schedule_lock();
	213
[4479890]	214	cluster * const cltr = proc->cltr;
[8bee858]	215	io_context$ * const ctx = proc->io.ctx;
[4479890]	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;
[2af1943]	243	/* paranoid */ verify( io.tscs[target].t.tv != ULLONG_MAX );
[18f7858]	244	HELP: if(target < ctxs_count) {
[31c967b]	245	const __readyQ_avg_t cutoff = calc_cutoff(ctsc, ctx->cq.id, ctxs_count, io.data, io.tscs, __shard_factor.io, false);
	246	const __readyQ_avg_t age = moving_average(ctsc, io.tscs[target].t.tv, io.tscs[target].t.ma, false);
[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
[1756e08]	275	bool __cfa_io_flush( struct processor * proc ) {
[dddb3dd0]	276	/* paranoid */ verify( ! __preemption_enabled() );
	277	/* paranoid */ verify( proc );
	278	/* paranoid */ verify( proc->io.ctx );
[1539bbd]	279
[8bee858]	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
[8bee858]	314	static inline void __fill(struct io_uring_sqe * out_sqes[], __u32 want, __u32 idxs[], struct io_context$ * ctx) {
[78da4ab]	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
[8bee858]	324	static inline bool __alloc(struct io_context$ * ctx, __u32 idxs[], __u32 want) {
[78da4ab]	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]
[8bee858]	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();
[1756e08]	355	struct processor * proc = __cfaabi_tls.this_processor;
[8bee858]	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
[8bee858]	380	io_arbiter$ * ioarb = proc->cltr->io.arbiter;
[dddb3dd0]	381	/* paranoid */ verify( ioarb );
	382
[1e6ffb44]	383	// __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for allocation\n");
[78da4ab]	384
[8bee858]	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
[8bee858]	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
[8bee858]	416	static inline void __submit( struct io_context$ * ctx, __u32 idxs[], __u32 have, bool lazy) {
[2432e8e]	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
[8bee858]	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();
[7ce8873]	434	__STATS__( true, if(!lazy) io.submit.eagr += 1; )
[1756e08]	435	struct processor * proc = __cfaabi_tls.this_processor;
[8bee858]	436	io_context$ * ctx = proc->io.ctx;
[dddb3dd0]	437	/* paranoid */ verify( __cfaabi_tls.this_processor );
	438	/* paranoid */ verify( ctx );
[e46c753]	439
[78da4ab]	440	// Can we proceed to the fast path
[dddb3dd0]	441	if( ctx == inctx ) // We have the right instance?
[78da4ab]	442	{
[dddb3dd0]	443	__submit(ctx, idxs, have, lazy);
[e46c753]	444
[78da4ab]	445	// Mark the instance as no longer in-use, re-enable interrupts and return
[d60d30e]	446	__STATS__( true, io.submit.fast += 1; )
[a3821fa]	447	enable_interrupts();
[ece0e80]	448
[1e6ffb44]	449	// __cfadbg_print_safe(io, "Kernel I/O : submitted on fast path\n");
[78da4ab]	450	return;
[e46c753]	451	}
[d384787]	452
[78da4ab]	453	// Fast path failed, fallback on arbitration
[d60d30e]	454	__STATS__( true, io.submit.slow += 1; )
[a3821fa]	455	enable_interrupts();
[5dadc9b7]	456
[1e6ffb44]	457	// __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for submission\n");
[426f60c]	458
[11054eb]	459	__ioarbiter_submit(inctx, idxs, have, lazy);
[78da4ab]	460	}
[2fab24e3]	461
[78da4ab]	462	//=============================================================================================
	463	// Flushing
[426f60c]	464	// Go through the ring's submit queue and release everything that has already been consumed
	465	// by io_uring
[78da4ab]	466	// This cannot be done by multiple threads
[8bee858]	467	static __u32 __release_sqes( struct io_context$ & ctx ) {
[78da4ab]	468	const __u32 mask = *ctx.sq.mask;
[732b406]	469
[426f60c]	470	__attribute__((unused))
[78da4ab]	471	__u32 ctail = *ctx.sq.kring.tail; // get the current tail of the queue
	472	__u32 chead = *ctx.sq.kring.head; // get the current head of the queue
	473	__u32 phead = ctx.sq.kring.released; // get the head the last time we were here
	474
	475	__u32 ftail = ctx.sq.free_ring.tail; // get the current tail of the queue
[732b406]	476
[426f60c]	477	// the 3 fields are organized like this diagram
	478	// except it's are ring
	479	// ---+--------+--------+----
	480	// ---+--------+--------+----
	481	// ^ ^ ^
	482	// phead chead ctail
	483
	484	// make sure ctail doesn't wrap around and reach phead
	485	/* paranoid */ verify(
	486	(ctail >= chead && chead >= phead)
	487	\|\| (chead >= phead && phead >= ctail)
	488	\|\| (phead >= ctail && ctail >= chead)
	489	);
	490
	491	// find the range we need to clear
[4998155]	492	__u32 count = chead - phead;
[426f60c]	493
[78da4ab]	494	if(count == 0) {
	495	return 0;
	496	}
	497
[426f60c]	498	// We acquired an previous-head/current-head range
	499	// go through the range and release the sqes
[34b61882]	500	for( i; count ) {
[1e6ffb44]	501	// __cfadbg_print_safe(io, "Kernel I/O : release loop\n");
[78da4ab]	502	__u32 idx = ctx.sq.kring.array[ (phead + i) & mask ];
	503	ctx.sq.free_ring.array[ (ftail + i) & mask ] = idx;
[34b61882]	504	}
[78da4ab]	505
	506	ctx.sq.kring.released = chead; // note up to were we processed
	507	__atomic_store_n(&ctx.sq.free_ring.tail, ftail + count, __ATOMIC_SEQ_CST);
	508
	509	__ioarbiter_notify(ctx);
	510
[34b61882]	511	return count;
	512	}
[35285fd]	513
[78da4ab]	514	//=============================================================================================
	515	// I/O Arbiter
	516	//=============================================================================================
[9f5a71eb]	517	static inline bool enqueue(__outstanding_io_queue & queue, __outstanding_io & item) {
	518	bool was_empty;
	519
[11054eb]	520	// Lock the list, it's not thread safe
	521	lock( queue.lock __cfaabi_dbg_ctx2 );
	522	{
[9f5a71eb]	523	was_empty = empty(queue.queue);
	524
[11054eb]	525	// Add our request to the list
	526	add( queue.queue, item );
	527
	528	// Mark as pending
	529	__atomic_store_n( &queue.empty, false, __ATOMIC_SEQ_CST );
	530	}
	531	unlock( queue.lock );
	532
[9f5a71eb]	533	return was_empty;
[11054eb]	534	}
	535
	536	static inline bool empty(__outstanding_io_queue & queue ) {
	537	return __atomic_load_n( &queue.empty, __ATOMIC_SEQ_CST);
	538	}
	539
[8bee858]	540	static io_context$ * __ioarbiter_allocate( io_arbiter$ & this, __u32 idxs[], __u32 want ) {
[1e6ffb44]	541	// __cfadbg_print_safe(io, "Kernel I/O : arbiter allocating\n");
[78da4ab]	542
[d60d30e]	543	__STATS__( false, io.alloc.block += 1; )
	544
[78da4ab]	545	// No one has any resources left, wait for something to finish
[11054eb]	546	// We need to add ourself to a list of pending allocs and wait for an answer
	547	__pending_alloc pa;
	548	pa.idxs = idxs;
	549	pa.want = want;
[78da4ab]	550
[9f5a71eb]	551	enqueue(this.pending, (__outstanding_io&)pa);
	552
[a55472cc]	553	wait( pa.waitctx );
[78da4ab]	554
[11054eb]	555	return pa.ctx;
[dddb3dd0]	556
[78da4ab]	557	}
	558
[8bee858]	559	static void __ioarbiter_notify( io_arbiter$ & this, io_context$ * ctx ) {
[11054eb]	560	/* paranoid */ verify( !empty(this.pending.queue) );
[78da4ab]	561
[11054eb]	562	lock( this.pending.lock __cfaabi_dbg_ctx2 );
	563	{
	564	while( !empty(this.pending.queue) ) {
	565	__cfadbg_print_safe(io, "Kernel I/O : notifying\n");
	566	__u32 have = ctx->sq.free_ring.tail - ctx->sq.free_ring.head;
	567	__pending_alloc & pa = (__pending_alloc&)head( this.pending.queue );
[78da4ab]	568
[11054eb]	569	if( have > pa.want ) goto DONE;
	570	drop( this.pending.queue );
[78da4ab]	571
[11054eb]	572	/* paranoid */__attribute__((unused)) bool ret =
[78da4ab]	573
[11054eb]	574	__alloc(ctx, pa.idxs, pa.want);
	575
	576	/* paranoid */ verify( ret );
	577
	578	pa.ctx = ctx;
	579
[a55472cc]	580	post( pa.waitctx );
[11054eb]	581	}
	582
	583	this.pending.empty = true;
	584	DONE:;
	585	}
	586	unlock( this.pending.lock );
[78da4ab]	587	}
	588
[8bee858]	589	static void __ioarbiter_notify( io_context$ & ctx ) {
[11054eb]	590	if(!empty( ctx.arbiter->pending )) {
[78da4ab]	591	__ioarbiter_notify( *ctx.arbiter, &ctx );
	592	}
	593	}
	594
	595	// Simply append to the pending
[8bee858]	596	static void __ioarbiter_submit( io_context$ * ctx, __u32 idxs[], __u32 have, bool lazy ) {
[78da4ab]	597	__cfadbg_print_safe(io, "Kernel I/O : submitting %u from the arbiter to context %u\n", have, ctx->fd);
	598
	599	__cfadbg_print_safe(io, "Kernel I/O : waiting to submit %u\n", have);
	600
[11054eb]	601	__external_io ei;
	602	ei.idxs = idxs;
	603	ei.have = have;
	604	ei.lazy = lazy;
[78da4ab]	605
[9f5a71eb]	606	bool we = enqueue(ctx->ext_sq, (__outstanding_io&)ei);
	607
[d529ad0]	608	__atomic_store_n(&ctx->proc->io.pending, true, __ATOMIC_SEQ_CST);
[9f5a71eb]	609
	610	if( we ) {
	611	sigval_t value = { PREEMPT_IO };
[95dab9e]	612	__cfaabi_pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
[9f5a71eb]	613	}
	614
[a55472cc]	615	wait( ei.waitctx );
[78da4ab]	616
	617	__cfadbg_print_safe(io, "Kernel I/O : %u submitted from arbiter\n", have);
	618	}
	619
[8bee858]	620	static void __ioarbiter_flush( io_context$ & ctx ) {
[11054eb]	621	if(!empty( ctx.ext_sq )) {
	622	__STATS__( false, io.flush.external += 1; )
[78da4ab]	623
[11054eb]	624	__cfadbg_print_safe(io, "Kernel I/O : arbiter flushing\n");
[d60d30e]	625
[11054eb]	626	lock( ctx.ext_sq.lock __cfaabi_dbg_ctx2 );
	627	{
	628	while( !empty(ctx.ext_sq.queue) ) {
	629	__external_io & ei = (__external_io&)drop( ctx.ext_sq.queue );
[78da4ab]	630
[2432e8e]	631	__submit_only(&ctx, ei.idxs, ei.have);
[78da4ab]	632
[a55472cc]	633	post( ei.waitctx );
[11054eb]	634	}
	635
	636	ctx.ext_sq.empty = true;
	637	}
	638	unlock(ctx.ext_sq.lock );
	639	}
[78da4ab]	640	}
[7ef162b2]	641
[d3605f8]	642	#if defined(CFA_WITH_IO_URING_IDLE)
[1756e08]	643	bool __kernel_read(struct processor * proc, io_future_t & future, iovec & iov, int fd) {
[8bee858]	644	io_context$ * ctx = proc->io.ctx;
[6ddef36]	645	/* paranoid */ verify( ! __preemption_enabled() );
	646	/* paranoid */ verify( proc == __cfaabi_tls.this_processor );
	647	/* paranoid */ verify( ctx );
[7ef162b2]	648
[6ddef36]	649	__u32 idx;
	650	struct io_uring_sqe * sqe;
[7ef162b2]	651
[6ddef36]	652	// We can proceed to the fast path
[010636f]	653	if( !__alloc(ctx, &idx, 1) ) {
	654	/* paranoid */ verify( false ); // for now check if this happens, next time just abort the sleep.
	655	return false;
	656	}
[6ddef36]	657
	658	// Allocation was successful
	659	__fill( &sqe, 1, &idx, ctx );
	660
	661	sqe->user_data = (uintptr_t)&future;
	662	sqe->flags = 0;
[a1f3d93]	663	sqe->fd = fd;
[6ddef36]	664	sqe->off = 0;
[d3605f8]	665	sqe->ioprio = 0;
[6ddef36]	666	sqe->fsync_flags = 0;
	667	sqe->__pad2[0] = 0;
	668	sqe->__pad2[1] = 0;
	669	sqe->__pad2[2] = 0;
[d3605f8]	670
	671	#if defined(CFA_HAVE_IORING_OP_READ)
	672	sqe->opcode = IORING_OP_READ;
	673	sqe->addr = (uint64_t)iov.iov_base;
	674	sqe->len = iov.iov_len;
	675	#elif defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV)
	676	sqe->opcode = IORING_OP_READV;
	677	sqe->addr = (uintptr_t)&iov;
	678	sqe->len = 1;
	679	#else
	680	#error CFA_WITH_IO_URING_IDLE but none of CFA_HAVE_READV, CFA_HAVE_IORING_OP_READV or CFA_HAVE_IORING_OP_READ defined
	681	#endif
[6ddef36]	682
	683	asm volatile("": : :"memory");
	684
	685	/* paranoid */ verify( sqe->user_data == (uintptr_t)&future );
[010636f]	686	__submit_only( ctx, &idx, 1 );
[6ddef36]	687
	688	/* paranoid */ verify( proc == __cfaabi_tls.this_processor );
	689	/* paranoid */ verify( ! __preemption_enabled() );
[078fb05]	690
	691	return true;
[6ddef36]	692	}
[18f7858]	693
[1756e08]	694	void __cfa_io_idle( struct processor * proc ) {
[18f7858]	695	iovec iov;
	696	__atomic_acquire( &proc->io.ctx->cq.lock );
	697
[262fafd9]	698	__attribute__((used)) volatile bool was_reset = false;
	699
[d5cdbed]	700	with( proc->idle_wctx) {
[18f7858]	701
[37a3aa23]	702	// Do we already have a pending read
	703	if(available(*ftr)) {
	704	// There is no pending read, we need to add one
	705	reset(*ftr);
	706
	707	iov.iov_base = rdbuf;
	708	iov.iov_len = sizeof(eventfd_t);
	709	__kernel_read(proc, *ftr, iov, evfd );
[262fafd9]	710	ftr->result = 0xDEADDEAD;
	711	((eventfd_t )rdbuf) = 0xDEADDEADDEADDEAD;
	712	was_reset = true;
[37a3aa23]	713	}
[18f7858]	714	}
	715
[010636f]	716	if( !__atomic_load_n( &proc->do_terminate, __ATOMIC_SEQ_CST ) ) {
	717	__ioarbiter_flush( *proc->io.ctx );
[262fafd9]	718	proc->idle_wctx.sleep_time = rdtscl();
[010636f]	719	ioring_syscsll( *proc->io.ctx, 1, IORING_ENTER_GETEVENTS);
	720	}
[18f7858]	721
[d5cdbed]	722	ready_schedule_lock();
[18f7858]	723	__cfa_do_drain( proc->io.ctx, proc->cltr );
[d5cdbed]	724	ready_schedule_unlock();
[262fafd9]	725
	726	asm volatile ("" :: "m" (was_reset));
[18f7858]	727	}
[6ddef36]	728	#endif
[47746a2]	729	#endif

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

Download in other formats:

Original Format