Context Navigation

io.cfa@ b230091

Visit:

ADT ast-experimental enum pthread-emulation qualifiedEnum

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

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

Download in other formats:

Original Format