Context Navigation

ready_queue.cfa@ af67ee1

Visit:

ADT ast-experimental enum forall-pointer-decay jacob/cs343-translation pthread-emulation qualifiedEnum

Last change on this file since af67ee1 was c86ee4c, checked in by Thierry Delisle <tdelisle@…>, 4 years ago
Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc
Property mode set to `100644`
File size: 31.1 KB

Rev	Line
[7768b8d]	1	//
	2	// Cforall Version 1.0.0 Copyright (C) 2019 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	// ready_queue.cfa --
	8	//
	9	// Author : Thierry Delisle
	10	// Created On : Mon Nov dd 16:29:18 2019
	11	// Last Modified By :
	12	// Last Modified On :
	13	// Update Count :
	14	//
	15
	16	#define __cforall_thread__
[43784ac]	17	#define _GNU_SOURCE
	18
[1b143de]	19	// #define __CFA_DEBUG_PRINT_READY_QUEUE__
[7768b8d]	20
[1eb239e4]	21
[6ba6846]	22	#define USE_RELAXED_FIFO
[9cc3a18]	23	// #define USE_WORK_STEALING
[6ba6846]	24	// #define USE_CPU_WORK_STEALING
[9cc3a18]	25
[7768b8d]	26	#include "bits/defs.hfa"
[12daa43]	27	#include "device/cpu.hfa"
[7768b8d]	28	#include "kernel_private.hfa"
	29
	30	#include "stdlib.hfa"
[61d7bec]	31	#include "math.hfa"
[7768b8d]	32
[0ee224b]	33	#include <errno.h>
[04b5cef]	34	#include <unistd.h>
	35
[0ee224b]	36	extern "C" {
	37	#include <sys/syscall.h> // __NR_xxx
	38	}
	39
[13c5e19]	40	#include "ready_subqueue.hfa"
	41
[7768b8d]	42	static const size_t cache_line_size = 64;
	43
[d2fadeb]	44	#if !defined(__CFA_NO_STATISTICS__)
	45	#define __STATS(...) __VA_ARGS__
	46	#else
	47	#define __STATS(...)
	48	#endif
	49
[dca5802]	50	// No overriden function, no environment variable, no define
	51	// fall back to a magic number
	52	#ifndef __CFA_MAX_PROCESSORS__
[b388ee81]	53	#define __CFA_MAX_PROCESSORS__ 1024
[dca5802]	54	#endif
[7768b8d]	55
[12daa43]	56	#if defined(USE_CPU_WORK_STEALING)
	57	#define READYQ_SHARD_FACTOR 2
	58	#elif defined(USE_RELAXED_FIFO)
[9cc3a18]	59	#define BIAS 4
	60	#define READYQ_SHARD_FACTOR 4
[5f6a172]	61	#define SEQUENTIAL_SHARD 1
[9cc3a18]	62	#elif defined(USE_WORK_STEALING)
	63	#define READYQ_SHARD_FACTOR 2
[5f6a172]	64	#define SEQUENTIAL_SHARD 2
[9cc3a18]	65	#else
	66	#error no scheduling strategy selected
	67	#endif
	68
[e84ab3d]	69	static inline struct thread$ * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats));
	70	static inline struct thread$ * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats));
	71	static inline struct thread$ * search(struct cluster * cltr);
[d2fadeb]	72	static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred);
[9cc3a18]	73
[04b5cef]	74
[dca5802]	75	// returns the maximum number of processors the RWLock support
[7768b8d]	76	__attribute__((weak)) unsigned __max_processors() {
	77	const char * max_cores_s = getenv("CFA_MAX_PROCESSORS");
	78	if(!max_cores_s) {
[504a7dc]	79	__cfadbg_print_nolock(ready_queue, "No CFA_MAX_PROCESSORS in ENV\n");
[dca5802]	80	return __CFA_MAX_PROCESSORS__;
[7768b8d]	81	}
	82
	83	char * endptr = 0p;
	84	long int max_cores_l = strtol(max_cores_s, &endptr, 10);
	85	if(max_cores_l < 1 \|\| max_cores_l > 65535) {
[504a7dc]	86	__cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS out of range : %ld\n", max_cores_l);
[dca5802]	87	return __CFA_MAX_PROCESSORS__;
[7768b8d]	88	}
	89	if('\0' != *endptr) {
[504a7dc]	90	__cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS not a decimal number : %s\n", max_cores_s);
[dca5802]	91	return __CFA_MAX_PROCESSORS__;
[7768b8d]	92	}
	93
	94	return max_cores_l;
	95	}
	96
[0ee224b]	97	#if defined(CFA_HAVE_LINUX_LIBRSEQ)
	98	// No forward declaration needed
	99	#define __kernel_rseq_register rseq_register_current_thread
	100	#define __kernel_rseq_unregister rseq_unregister_current_thread
	101	#elif defined(CFA_HAVE_LINUX_RSEQ_H)
	102	void __kernel_raw_rseq_register (void);
	103	void __kernel_raw_rseq_unregister(void);
	104
	105	#define __kernel_rseq_register __kernel_raw_rseq_register
	106	#define __kernel_rseq_unregister __kernel_raw_rseq_unregister
	107	#else
	108	// No forward declaration needed
	109	// No initialization needed
	110	static inline void noop(void) {}
	111
	112	#define __kernel_rseq_register noop
	113	#define __kernel_rseq_unregister noop
	114	#endif
	115
[7768b8d]	116	//=======================================================================
	117	// Cluster wide reader-writer lock
	118	//=======================================================================
[b388ee81]	119	void ?{}(__scheduler_RWLock_t & this) {
[7768b8d]	120	this.max = __max_processors();
	121	this.alloc = 0;
	122	this.ready = 0;
	123	this.data = alloc(this.max);
[c993b15]	124	this.write_lock = false;
[7768b8d]	125
	126	/paranoid/ verify(__atomic_is_lock_free(sizeof(this.alloc), &this.alloc));
	127	/paranoid/ verify(__atomic_is_lock_free(sizeof(this.ready), &this.ready));
	128
	129	}
[b388ee81]	130	void ^?{}(__scheduler_RWLock_t & this) {
[7768b8d]	131	free(this.data);
	132	}
	133
	134
	135	//=======================================================================
	136	// Lock-Free registering/unregistering of threads
[c993b15]	137	unsigned register_proc_id( void ) with(*__scheduler_lock) {
[0ee224b]	138	__kernel_rseq_register();
	139
[b388ee81]	140	__cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
[c993b15]	141	bool * handle = (bool *)&kernelTLS().sched_lock;
[504a7dc]	142
[7768b8d]	143	// Step - 1 : check if there is already space in the data
	144	uint_fast32_t s = ready;
	145
	146	// Check among all the ready
	147	for(uint_fast32_t i = 0; i < s; i++) {
[c993b15]	148	bool * volatile * cell = (bool * volatile *)&data[i]; // Cforall is bugged and the double volatiles causes problems
	149	/* paranoid / verify( handle != cell );
	150
	151	bool * null = 0p; // Re-write every loop since compare thrashes it
	152	if( __atomic_load_n(cell, (int)__ATOMIC_RELAXED) == null
	153	&& __atomic_compare_exchange_n( cell, &null, handle, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
	154	/* paranoid */ verify(i < ready);
	155	/* paranoid */ verify( (kernelTLS().sched_id = i, true) );
	156	return i;
[7768b8d]	157	}
	158	}
	159
[b388ee81]	160	if(max <= alloc) abort("Trying to create more than %ud processors", __scheduler_lock->max);
[7768b8d]	161
	162	// Step - 2 : F&A to get a new spot in the array.
	163	uint_fast32_t n = __atomic_fetch_add(&alloc, 1, __ATOMIC_SEQ_CST);
[b388ee81]	164	if(max <= n) abort("Trying to create more than %ud processors", __scheduler_lock->max);
[7768b8d]	165
	166	// Step - 3 : Mark space as used and then publish it.
[c993b15]	167	data[n] = handle;
[fd9b524]	168	while() {
[7768b8d]	169	unsigned copy = n;
	170	if( __atomic_load_n(&ready, __ATOMIC_RELAXED) == n
	171	&& __atomic_compare_exchange_n(&ready, &copy, n + 1, true, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST))
	172	break;
[fd9b524]	173	Pause();
[7768b8d]	174	}
	175
[1b143de]	176	__cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n);
[504a7dc]	177
[7768b8d]	178	// Return new spot.
[c993b15]	179	/* paranoid */ verify(n < ready);
	180	/* paranoid */ verify( (kernelTLS().sched_id = n, true) );
	181	return n;
[7768b8d]	182	}
	183
[c993b15]	184	void unregister_proc_id( unsigned id ) with(*__scheduler_lock) {
	185	/* paranoid */ verify(id < ready);
	186	/* paranoid */ verify(id == kernelTLS().sched_id);
	187	/* paranoid */ verify(data[id] == &kernelTLS().sched_lock);
	188
	189	bool * volatile * cell = (bool * volatile *)&data[id]; // Cforall is bugged and the double volatiles causes problems
	190
	191	__atomic_store_n(cell, 0p, __ATOMIC_RELEASE);
[504a7dc]	192
	193	__cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc);
[0ee224b]	194
	195	__kernel_rseq_unregister();
[7768b8d]	196	}
	197
	198	//-----------------------------------------------------------------------
	199	// Writer side : acquire when changing the ready queue, e.g. adding more
	200	// queues or removing them.
[b388ee81]	201	uint_fast32_t ready_mutate_lock( void ) with(*__scheduler_lock) {
[8fc652e0]	202	/* paranoid */ verify( ! __preemption_enabled() );
[c993b15]	203	/* paranoid */ verify( ! kernelTLS().sched_lock );
[62502cc4]	204
[7768b8d]	205	// Step 1 : lock global lock
	206	// It is needed to avoid processors that register mid Critical-Section
	207	// to simply lock their own lock and enter.
[c993b15]	208	__atomic_acquire( &write_lock );
[7768b8d]	209
	210	// Step 2 : lock per-proc lock
	211	// Processors that are currently being registered aren't counted
	212	// but can't be in read_lock or in the critical section.
	213	// All other processors are counted
	214	uint_fast32_t s = ready;
	215	for(uint_fast32_t i = 0; i < s; i++) {
[c993b15]	216	volatile bool * llock = data[i];
	217	if(llock) __atomic_acquire( llock );
[7768b8d]	218	}
	219
[8fc652e0]	220	/* paranoid */ verify( ! __preemption_enabled() );
[7768b8d]	221	return s;
	222	}
	223
[b388ee81]	224	void ready_mutate_unlock( uint_fast32_t last_s ) with(*__scheduler_lock) {
[8fc652e0]	225	/* paranoid */ verify( ! __preemption_enabled() );
[62502cc4]	226
[7768b8d]	227	// Step 1 : release local locks
	228	// This must be done while the global lock is held to avoid
	229	// threads that where created mid critical section
	230	// to race to lock their local locks and have the writer
	231	// immidiately unlock them
	232	// Alternative solution : return s in write_lock and pass it to write_unlock
	233	for(uint_fast32_t i = 0; i < last_s; i++) {
[c993b15]	234	volatile bool * llock = data[i];
	235	if(llock) __atomic_store_n(llock, (bool)false, __ATOMIC_RELEASE);
[7768b8d]	236	}
	237
	238	// Step 2 : release global lock
[c993b15]	239	/paranoid/ assert(true == write_lock);
	240	__atomic_store_n(&write_lock, (bool)false, __ATOMIC_RELEASE);
[62502cc4]	241
[8fc652e0]	242	/* paranoid */ verify( ! __preemption_enabled() );
[7768b8d]	243	}
	244
	245	//=======================================================================
[9cc3a18]	246	// Cforall Ready Queue used for scheduling
[b798713]	247	//=======================================================================
	248	void ?{}(__ready_queue_t & this) with (this) {
[12daa43]	249	#if defined(USE_CPU_WORK_STEALING)
	250	lanes.count = cpu_info.hthrd_count * READYQ_SHARD_FACTOR;
	251	lanes.data = alloc( lanes.count );
	252	lanes.tscs = alloc( lanes.count );
	253
	254	for( idx; (size_t)lanes.count ) {
	255	(lanes.data[idx]){};
	256	lanes.tscs[idx].tv = rdtscl();
	257	}
	258	#else
	259	lanes.data = 0p;
	260	lanes.tscs = 0p;
	261	lanes.count = 0;
	262	#endif
[b798713]	263	}
	264
	265	void ^?{}(__ready_queue_t & this) with (this) {
[12daa43]	266	#if !defined(USE_CPU_WORK_STEALING)
	267	verify( SEQUENTIAL_SHARD == lanes.count );
	268	#endif
	269
[dca5802]	270	free(lanes.data);
[9cc3a18]	271	free(lanes.tscs);
[dca5802]	272	}
	273
[64a7146]	274	//-----------------------------------------------------------------------
[12daa43]	275	#if defined(USE_CPU_WORK_STEALING)
[e84ab3d]	276	__attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, bool push_local) with (cltr->ready_queue) {
[12daa43]	277	__cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
	278
	279	processor * const proc = kernelTLS().this_processor;
	280	const bool external = !push_local \|\| (!proc) \|\| (cltr != proc->cltr);
	281
	282	const int cpu = __kernel_getcpu();
	283	/* paranoid */ verify(cpu >= 0);
	284	/* paranoid */ verify(cpu < cpu_info.hthrd_count);
	285	/* paranoid / verify(cpu READYQ_SHARD_FACTOR < lanes.count);
	286
[df7597e0]	287	const cpu_map_entry_t & map = cpu_info.llc_map[cpu];
	288	/* paranoid / verify(map.start READYQ_SHARD_FACTOR < lanes.count);
	289	/* paranoid / verify(map.self READYQ_SHARD_FACTOR < lanes.count);
[5614552a]	290	/* paranoid / verifyf((map.start + map.count) READYQ_SHARD_FACTOR <= lanes.count, "have %zu lanes but map can go up to %u", lanes.count, (map.start + map.count) * READYQ_SHARD_FACTOR);
[df7597e0]	291
	292	const int start = map.self * READYQ_SHARD_FACTOR;
[12daa43]	293	unsigned i;
	294	do {
	295	unsigned r;
	296	if(unlikely(external)) { r = __tls_rand(); }
	297	else { r = proc->rdq.its++; }
	298	i = start + (r % READYQ_SHARD_FACTOR);
	299	// If we can't lock it retry
	300	} while( !__atomic_try_acquire( &lanes.data[i].lock ) );
	301
	302	// Actually push it
	303	push(lanes.data[i], thrd);
	304
	305	// Unlock and return
	306	__atomic_unlock( &lanes.data[i].lock );
	307
	308	#if !defined(__CFA_NO_STATISTICS__)
	309	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
	310	else __tls_stats()->ready.push.local.success++;
	311	#endif
	312
	313	__cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
	314
	315	}
	316
	317	// Pop from the ready queue from a given cluster
[e84ab3d]	318	__attribute__((hot)) thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
[12daa43]	319	/* paranoid */ verify( lanes.count > 0 );
	320	/* paranoid */ verify( kernelTLS().this_processor );
	321
	322	const int cpu = __kernel_getcpu();
	323	/* paranoid */ verify(cpu >= 0);
	324	/* paranoid */ verify(cpu < cpu_info.hthrd_count);
[df7597e0]	325	/* paranoid / verify(cpu READYQ_SHARD_FACTOR < lanes.count);
	326
	327	const cpu_map_entry_t & map = cpu_info.llc_map[cpu];
	328	/* paranoid / verify(map.start READYQ_SHARD_FACTOR < lanes.count);
	329	/* paranoid / verify(map.self READYQ_SHARD_FACTOR < lanes.count);
[5614552a]	330	/* paranoid / verifyf((map.start + map.count) READYQ_SHARD_FACTOR <= lanes.count, "have %zu lanes but map can go up to %u", lanes.count, (map.start + map.count) * READYQ_SHARD_FACTOR);
[12daa43]	331
	332	processor * const proc = kernelTLS().this_processor;
[df7597e0]	333	const int start = map.self * READYQ_SHARD_FACTOR;
[12daa43]	334
	335	// Did we already have a help target
	336	if(proc->rdq.target == -1u) {
	337	// if We don't have a
	338	unsigned long long min = ts(lanes.data[start]);
	339	for(i; READYQ_SHARD_FACTOR) {
	340	unsigned long long tsc = ts(lanes.data[start + i]);
	341	if(tsc < min) min = tsc;
	342	}
	343	proc->rdq.cutoff = min;
[953827a]	344
	345	/* paranoid */ verify(lanes.count < 65536); // The following code assumes max 65536 cores.
	346	/* paranoid */ verify(map.count < 65536); // The following code assumes max 65536 cores.
	347
[1f45c7d]	348	if(0 == (__tls_rand() % 10_000)) {
	349	proc->rdq.target = __tls_rand() % lanes.count;
[953827a]	350	} else {
[1f45c7d]	351	unsigned cpu_chaos = map.start + (__tls_rand() % map.count);
	352	proc->rdq.target = (cpu_chaos * READYQ_SHARD_FACTOR) + (__tls_rand() % READYQ_SHARD_FACTOR);
[953827a]	353	/* paranoid / verify(proc->rdq.target >= (map.start READYQ_SHARD_FACTOR));
	354	/* paranoid / verify(proc->rdq.target < ((map.start + map.count) READYQ_SHARD_FACTOR));
	355	}
	356
	357	/* paranoid */ verify(proc->rdq.target != -1u);
[12daa43]	358	}
	359	else {
	360	const unsigned long long bias = 0; //2_500_000_000;
	361	const unsigned long long cutoff = proc->rdq.cutoff > bias ? proc->rdq.cutoff - bias : proc->rdq.cutoff;
	362	{
	363	unsigned target = proc->rdq.target;
	364	proc->rdq.target = -1u;
	365	if(lanes.tscs[target].tv < cutoff && ts(lanes.data[target]) < cutoff) {
[e84ab3d]	366	thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
[12daa43]	367	proc->rdq.last = target;
	368	if(t) return t;
	369	}
	370	}
	371
	372	unsigned last = proc->rdq.last;
	373	if(last != -1u && lanes.tscs[last].tv < cutoff && ts(lanes.data[last]) < cutoff) {
[e84ab3d]	374	thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.help));
[12daa43]	375	if(t) return t;
	376	}
	377	else {
	378	proc->rdq.last = -1u;
	379	}
	380	}
	381
	382	for(READYQ_SHARD_FACTOR) {
	383	unsigned i = start + (proc->rdq.itr++ % READYQ_SHARD_FACTOR);
[e84ab3d]	384	if(thread$ * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
[12daa43]	385	}
	386
	387	// All lanes where empty return 0p
	388	return 0p;
	389	}
	390
[e84ab3d]	391	__attribute__((hot)) struct thread$ * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
[12daa43]	392	processor * const proc = kernelTLS().this_processor;
	393	unsigned last = proc->rdq.last;
[953827a]	394	if(last != -1u) {
[e84ab3d]	395	struct thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.steal));
[953827a]	396	if(t) return t;
	397	proc->rdq.last = -1u;
	398	}
[12daa43]	399
	400	unsigned i = __tls_rand() % lanes.count;
	401	return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));
	402	}
[e84ab3d]	403	__attribute__((hot)) struct thread$ * pop_search(struct cluster * cltr) {
[12daa43]	404	return search(cltr);
	405	}
	406	#endif
[431cd4f]	407	#if defined(USE_RELAXED_FIFO)
	408	//-----------------------------------------------------------------------
	409	// get index from random number with or without bias towards queues
	410	static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred) {
	411	unsigned i;
	412	bool local;
	413	unsigned rlow = r % BIAS;
	414	unsigned rhigh = r / BIAS;
	415	if((0 != rlow) && preferred >= 0) {
	416	// (BIAS - 1) out of BIAS chances
	417	// Use perferred queues
	418	i = preferred + (rhigh % READYQ_SHARD_FACTOR);
	419	local = true;
	420	}
	421	else {
	422	// 1 out of BIAS chances
	423	// Use all queues
	424	i = rhigh;
	425	local = false;
	426	}
	427	return [i, local];
	428	}
	429
[e84ab3d]	430	__attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, bool push_local) with (cltr->ready_queue) {
[431cd4f]	431	__cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
[1b143de]	432
[b808625]	433	const bool external = !push_local \|\| (!kernelTLS().this_processor) \|\| (cltr != kernelTLS().this_processor->cltr);
[431cd4f]	434	/* paranoid */ verify(external \|\| kernelTLS().this_processor->rdq.id < lanes.count );
[fd1f65e]	435
[431cd4f]	436	bool local;
	437	int preferred = external ? -1 : kernelTLS().this_processor->rdq.id;
[52769ba]	438
[431cd4f]	439	// Try to pick a lane and lock it
	440	unsigned i;
	441	do {
	442	// Pick the index of a lane
	443	unsigned r = __tls_rand_fwd();
	444	[i, local] = idx_from_r(r, preferred);
[772411a]	445
[431cd4f]	446	i %= __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
	447
	448	#if !defined(__CFA_NO_STATISTICS__)
[d2fadeb]	449	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
	450	else if(local) __tls_stats()->ready.push.local.attempt++;
	451	else __tls_stats()->ready.push.share.attempt++;
[431cd4f]	452	#endif
[b798713]	453
[431cd4f]	454	// If we can't lock it retry
	455	} while( !__atomic_try_acquire( &lanes.data[i].lock ) );
	456
	457	// Actually push it
	458	push(lanes.data[i], thrd);
	459
[b808625]	460	// Unlock and return
	461	__atomic_unlock( &lanes.data[i].lock );
[431cd4f]	462
	463	// Mark the current index in the tls rng instance as having an item
	464	__tls_rand_advance_bck();
	465
	466	__cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
	467
	468	// Update statistics
[b798713]	469	#if !defined(__CFA_NO_STATISTICS__)
[d2fadeb]	470	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
	471	else if(local) __tls_stats()->ready.push.local.success++;
	472	else __tls_stats()->ready.push.share.success++;
[b798713]	473	#endif
[431cd4f]	474	}
[b798713]	475
[431cd4f]	476	// Pop from the ready queue from a given cluster
[e84ab3d]	477	__attribute__((hot)) thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
[431cd4f]	478	/* paranoid */ verify( lanes.count > 0 );
	479	/* paranoid */ verify( kernelTLS().this_processor );
	480	/* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
[b798713]	481
[431cd4f]	482	unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
	483	int preferred = kernelTLS().this_processor->rdq.id;
[dca5802]	484
	485
[431cd4f]	486	// As long as the list is not empty, try finding a lane that isn't empty and pop from it
	487	for(25) {
	488	// Pick two lists at random
	489	unsigned ri = __tls_rand_bck();
	490	unsigned rj = __tls_rand_bck();
[c426b03]	491
[431cd4f]	492	unsigned i, j;
	493	__attribute__((unused)) bool locali, localj;
	494	[i, locali] = idx_from_r(ri, preferred);
	495	[j, localj] = idx_from_r(rj, preferred);
[1b143de]	496
[431cd4f]	497	i %= count;
	498	j %= count;
[9cc3a18]	499
[431cd4f]	500	// try popping from the 2 picked lists
[e84ab3d]	501	struct thread$ * thrd = try_pop(cltr, i, j __STATS(, *(locali \|\| localj ? &__tls_stats()->ready.pop.local : &__tls_stats()->ready.pop.help)));
[431cd4f]	502	if(thrd) {
	503	return thrd;
	504	}
	505	}
[13c5e19]	506
[431cd4f]	507	// All lanes where empty return 0p
	508	return 0p;
	509	}
[772411a]	510
[e84ab3d]	511	__attribute__((hot)) struct thread$ * pop_slow(struct cluster * cltr) { return pop_fast(cltr); }
	512	__attribute__((hot)) struct thread$ * pop_search(struct cluster * cltr) {
[431cd4f]	513	return search(cltr);
	514	}
	515	#endif
	516	#if defined(USE_WORK_STEALING)
[e84ab3d]	517	__attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, bool push_local) with (cltr->ready_queue) {
[431cd4f]	518	__cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
[772411a]	519
[d3ba775]	520	// #define USE_PREFERRED
	521	#if !defined(USE_PREFERRED)
[b808625]	522	const bool external = !push_local \|\| (!kernelTLS().this_processor) \|\| (cltr != kernelTLS().this_processor->cltr);
[431cd4f]	523	/* paranoid */ verify(external \|\| kernelTLS().this_processor->rdq.id < lanes.count );
[d3ba775]	524	#else
	525	unsigned preferred = thrd->preferred;
[b808625]	526	const bool external = push_local \|\| (!kernelTLS().this_processor) \|\| preferred == -1u \|\| thrd->curr_cluster != cltr;
[d3ba775]	527	/* paranoid */ verifyf(external \|\| preferred < lanes.count, "Invalid preferred queue %u for %u lanes", preferred, lanes.count );
[772411a]	528
[d3ba775]	529	unsigned r = preferred % READYQ_SHARD_FACTOR;
	530	const unsigned start = preferred - r;
[2b96031]	531	#endif
[431cd4f]	532
	533	// Try to pick a lane and lock it
	534	unsigned i;
	535	do {
[d2fadeb]	536	#if !defined(__CFA_NO_STATISTICS__)
	537	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
	538	else __tls_stats()->ready.push.local.attempt++;
	539	#endif
	540
[431cd4f]	541	if(unlikely(external)) {
	542	i = __tls_rand() % lanes.count;
	543	}
	544	else {
[d3ba775]	545	#if !defined(USE_PREFERRED)
[b808625]	546	processor * proc = kernelTLS().this_processor;
	547	unsigned r = proc->rdq.its++;
	548	i = proc->rdq.id + (r % READYQ_SHARD_FACTOR);
	549	#else
[d3ba775]	550	i = start + (r++ % READYQ_SHARD_FACTOR);
	551	#endif
	552	}
[431cd4f]	553	// If we can't lock it retry
	554	} while( !__atomic_try_acquire( &lanes.data[i].lock ) );
[13c5e19]	555
[431cd4f]	556	// Actually push it
	557	push(lanes.data[i], thrd);
[13c5e19]	558
[b808625]	559	// Unlock and return
	560	__atomic_unlock( &lanes.data[i].lock );
[431cd4f]	561
[d2fadeb]	562	#if !defined(__CFA_NO_STATISTICS__)
	563	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
	564	else __tls_stats()->ready.push.local.success++;
	565	#endif
	566
[431cd4f]	567	__cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
[13c5e19]	568	}
	569
[431cd4f]	570	// Pop from the ready queue from a given cluster
[e84ab3d]	571	__attribute__((hot)) thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
[431cd4f]	572	/* paranoid */ verify( lanes.count > 0 );
	573	/* paranoid */ verify( kernelTLS().this_processor );
	574	/* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
	575
	576	processor * proc = kernelTLS().this_processor;
	577
	578	if(proc->rdq.target == -1u) {
[1680072]	579	unsigned long long min = ts(lanes.data[proc->rdq.id]);
	580	for(int i = 0; i < READYQ_SHARD_FACTOR; i++) {
	581	unsigned long long tsc = ts(lanes.data[proc->rdq.id + i]);
	582	if(tsc < min) min = tsc;
	583	}
	584	proc->rdq.cutoff = min;
[f55d54d]	585	proc->rdq.target = __tls_rand() % lanes.count;
[431cd4f]	586	}
[341aa39]	587	else {
	588	unsigned target = proc->rdq.target;
[431cd4f]	589	proc->rdq.target = -1u;
[9cac0da]	590	const unsigned long long bias = 0; //2_500_000_000;
	591	const unsigned long long cutoff = proc->rdq.cutoff > bias ? proc->rdq.cutoff - bias : proc->rdq.cutoff;
	592	if(lanes.tscs[target].tv < cutoff && ts(lanes.data[target]) < cutoff) {
[e84ab3d]	593	thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
[341aa39]	594	if(t) return t;
	595	}
[431cd4f]	596	}
[13c5e19]	597
[431cd4f]	598	for(READYQ_SHARD_FACTOR) {
[f55d54d]	599	unsigned i = proc->rdq.id + (proc->rdq.itr++ % READYQ_SHARD_FACTOR);
[e84ab3d]	600	if(thread$ * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
[431cd4f]	601	}
	602	return 0p;
[1eb239e4]	603	}
	604
[e84ab3d]	605	__attribute__((hot)) struct thread$ * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
[fc59df78]	606	unsigned i = __tls_rand() % lanes.count;
	607	return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));
	608	}
[431cd4f]	609
[e84ab3d]	610	__attribute__((hot)) struct thread$ * pop_search(struct cluster * cltr) with (cltr->ready_queue) {
[431cd4f]	611	return search(cltr);
	612	}
	613	#endif
[1eb239e4]	614
[9cc3a18]	615	//=======================================================================
	616	// Various Ready Queue utilities
	617	//=======================================================================
	618	// these function work the same or almost the same
	619	// whether they are using work-stealing or relaxed fifo scheduling
[1eb239e4]	620
[9cc3a18]	621	//-----------------------------------------------------------------------
	622	// try to pop from a lane given by index w
[e84ab3d]	623	static inline struct thread$ * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
[d2fadeb]	624	__STATS( stats.attempt++; )
	625
[dca5802]	626	// Get relevant elements locally
	627	__intrusive_lane_t & lane = lanes.data[w];
	628
[b798713]	629	// If list looks empty retry
[d2fadeb]	630	if( is_empty(lane) ) {
	631	return 0p;
	632	}
[b798713]	633
	634	// If we can't get the lock retry
[d2fadeb]	635	if( !__atomic_try_acquire(&lane.lock) ) {
	636	return 0p;
	637	}
[b798713]	638
	639	// If list is empty, unlock and retry
[dca5802]	640	if( is_empty(lane) ) {
	641	__atomic_unlock(&lane.lock);
[b798713]	642	return 0p;
	643	}
	644
	645	// Actually pop the list
[e84ab3d]	646	struct thread$ * thrd;
[f302d80]	647	unsigned long long tsv;
	648	[thrd, tsv] = pop(lane);
[b798713]	649
[dca5802]	650	/* paranoid */ verify(thrd);
[78ea291]	651	/* paranoid */ verify(tsv);
[dca5802]	652	/* paranoid */ verify(lane.lock);
[b798713]	653
	654	// Unlock and return
[dca5802]	655	__atomic_unlock(&lane.lock);
[b798713]	656
[dca5802]	657	// Update statistics
[d2fadeb]	658	__STATS( stats.success++; )
[b798713]	659
[431cd4f]	660	#if defined(USE_WORK_STEALING)
[f302d80]	661	lanes.tscs[w].tv = tsv;
[9cc3a18]	662	#endif
[d72c074]	663
[d3ba775]	664	thrd->preferred = w;
	665
[dca5802]	666	// return the popped thread
[b798713]	667	return thrd;
	668	}
[04b5cef]	669
[9cc3a18]	670	//-----------------------------------------------------------------------
	671	// try to pop from any lanes making sure you don't miss any threads push
	672	// before the start of the function
[e84ab3d]	673	static inline struct thread$ * search(struct cluster * cltr) with (cltr->ready_queue) {
[9cc3a18]	674	/* paranoid */ verify( lanes.count > 0 );
	675	unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
	676	unsigned offset = __tls_rand();
	677	for(i; count) {
	678	unsigned idx = (offset + i) % count;
[e84ab3d]	679	struct thread$ * thrd = try_pop(cltr, idx __STATS(, __tls_stats()->ready.pop.search));
[9cc3a18]	680	if(thrd) {
	681	return thrd;
	682	}
[13c5e19]	683	}
[9cc3a18]	684
	685	// All lanes where empty return 0p
	686	return 0p;
[b798713]	687	}
	688
	689	//-----------------------------------------------------------------------
[9cc3a18]	690	// Check that all the intrusive queues in the data structure are still consistent
[b798713]	691	static void check( __ready_queue_t & q ) with (q) {
[d3ba775]	692	#if defined(__CFA_WITH_VERIFY__)
[b798713]	693	{
[dca5802]	694	for( idx ; lanes.count ) {
	695	__intrusive_lane_t & sl = lanes.data[idx];
	696	assert(!lanes.data[idx].lock);
[b798713]	697
[2b96031]	698	if(is_empty(sl)) {
	699	assert( sl.anchor.next == 0p );
[ef94ae7]	700	assert( sl.anchor.ts == -1llu );
[2b96031]	701	assert( mock_head(sl) == sl.prev );
	702	} else {
	703	assert( sl.anchor.next != 0p );
[ef94ae7]	704	assert( sl.anchor.ts != -1llu );
[2b96031]	705	assert( mock_head(sl) != sl.prev );
	706	}
[b798713]	707	}
	708	}
	709	#endif
	710	}
	711
[9cc3a18]	712	//-----------------------------------------------------------------------
	713	// Given 2 indexes, pick the list with the oldest push an try to pop from it
[e84ab3d]	714	static inline struct thread$ * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
[9cc3a18]	715	// Pick the bet list
	716	int w = i;
	717	if( __builtin_expect(!is_empty(lanes.data[j]), true) ) {
	718	w = (ts(lanes.data[i]) < ts(lanes.data[j])) ? i : j;
	719	}
	720
[d2fadeb]	721	return try_pop(cltr, w __STATS(, stats));
[9cc3a18]	722	}
	723
[b798713]	724	// Call this function of the intrusive list was moved using memcpy
[dca5802]	725	// fixes the list so that the pointers back to anchors aren't left dangling
	726	static inline void fix(__intrusive_lane_t & ll) {
[2b96031]	727	if(is_empty(ll)) {
	728	verify(ll.anchor.next == 0p);
	729	ll.prev = mock_head(ll);
	730	}
[b798713]	731	}
	732
[69914cbc]	733	static void assign_list(unsigned & value, dlist(processor) & list, unsigned count) {
[a017ee7]	734	processor * it = &list`first;
	735	for(unsigned i = 0; i < count; i++) {
	736	/* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
[431cd4f]	737	it->rdq.id = value;
	738	it->rdq.target = -1u;
[9cc3a18]	739	value += READYQ_SHARD_FACTOR;
[a017ee7]	740	it = &(*it)`next;
	741	}
	742	}
	743
[9cc3a18]	744	static void reassign_cltr_id(struct cluster * cltr) {
[a017ee7]	745	unsigned preferred = 0;
[9cc3a18]	746	assign_list(preferred, cltr->procs.actives, cltr->procs.total - cltr->procs.idle);
	747	assign_list(preferred, cltr->procs.idles , cltr->procs.idle );
[a017ee7]	748	}
	749
[431cd4f]	750	static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
	751	#if defined(USE_WORK_STEALING)
	752	lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
	753	for(i; lanes.count) {
[ef94ae7]	754	unsigned long long tsc1 = ts(lanes.data[i]);
[6ba6846]	755	unsigned long long tsc2 = rdtscl();
[ef94ae7]	756	lanes.tscs[i].tv = min(tsc1, tsc2);
[431cd4f]	757	}
	758	#endif
	759	}
	760
[12daa43]	761	#if defined(USE_CPU_WORK_STEALING)
	762	// ready_queue size is fixed in this case
	763	void ready_queue_grow(struct cluster * cltr) {}
	764	void ready_queue_shrink(struct cluster * cltr) {}
	765	#else
	766	// Grow the ready queue
	767	void ready_queue_grow(struct cluster * cltr) {
	768	size_t ncount;
	769	int target = cltr->procs.total;
	770
	771	/* paranoid */ verify( ready_mutate_islocked() );
	772	__cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue\n");
	773
	774	// Make sure that everything is consistent
	775	/* paranoid */ check( cltr->ready_queue );
	776
	777	// grow the ready queue
	778	with( cltr->ready_queue ) {
	779	// Find new count
	780	// Make sure we always have atleast 1 list
	781	if(target >= 2) {
	782	ncount = target * READYQ_SHARD_FACTOR;
	783	} else {
	784	ncount = SEQUENTIAL_SHARD;
	785	}
[b798713]	786
[12daa43]	787	// Allocate new array (uses realloc and memcpies the data)
	788	lanes.data = alloc( ncount, lanes.data`realloc );
[b798713]	789
[12daa43]	790	// Fix the moved data
	791	for( idx; (size_t)lanes.count ) {
	792	fix(lanes.data[idx]);
	793	}
[b798713]	794
[12daa43]	795	// Construct new data
	796	for( idx; (size_t)lanes.count ~ ncount) {
	797	(lanes.data[idx]){};
	798	}
[b798713]	799
[12daa43]	800	// Update original
	801	lanes.count = ncount;
	802	}
[b798713]	803
[12daa43]	804	fix_times(cltr);
[9cc3a18]	805
[12daa43]	806	reassign_cltr_id(cltr);
[a017ee7]	807
[12daa43]	808	// Make sure that everything is consistent
	809	/* paranoid */ check( cltr->ready_queue );
[dca5802]	810
[12daa43]	811	__cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue done\n");
[dca5802]	812
[12daa43]	813	/* paranoid */ verify( ready_mutate_islocked() );
	814	}
[b798713]	815
[12daa43]	816	// Shrink the ready queue
	817	void ready_queue_shrink(struct cluster * cltr) {
	818	/* paranoid */ verify( ready_mutate_islocked() );
	819	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue\n");
[dca5802]	820
[12daa43]	821	// Make sure that everything is consistent
	822	/* paranoid */ check( cltr->ready_queue );
[dca5802]	823
[12daa43]	824	int target = cltr->procs.total;
[a017ee7]	825
[12daa43]	826	with( cltr->ready_queue ) {
	827	// Remember old count
	828	size_t ocount = lanes.count;
[b798713]	829
[12daa43]	830	// Find new count
	831	// Make sure we always have atleast 1 list
	832	lanes.count = target >= 2 ? target * READYQ_SHARD_FACTOR: SEQUENTIAL_SHARD;
	833	/* paranoid */ verify( ocount >= lanes.count );
	834	/* paranoid / verify( lanes.count == target READYQ_SHARD_FACTOR \|\| target < 2 );
[dca5802]	835
[12daa43]	836	// for printing count the number of displaced threads
	837	#if defined(__CFA_DEBUG_PRINT__) \|\| defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
	838	__attribute__((unused)) size_t displaced = 0;
	839	#endif
[b798713]	840
[12daa43]	841	// redistribute old data
	842	for( idx; (size_t)lanes.count ~ ocount) {
	843	// Lock is not strictly needed but makes checking invariants much easier
	844	__attribute__((unused)) bool locked = __atomic_try_acquire(&lanes.data[idx].lock);
	845	verify(locked);
[dca5802]	846
[12daa43]	847	// As long as we can pop from this lane to push the threads somewhere else in the queue
	848	while(!is_empty(lanes.data[idx])) {
[e84ab3d]	849	struct thread$ * thrd;
[12daa43]	850	unsigned long long _;
	851	[thrd, _] = pop(lanes.data[idx]);
[dca5802]	852
[12daa43]	853	push(cltr, thrd, true);
[dca5802]	854
[12daa43]	855	// for printing count the number of displaced threads
	856	#if defined(__CFA_DEBUG_PRINT__) \|\| defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
	857	displaced++;
	858	#endif
	859	}
[b798713]	860
[12daa43]	861	// Unlock the lane
	862	__atomic_unlock(&lanes.data[idx].lock);
[b798713]	863
[12daa43]	864	// TODO print the queue statistics here
[b798713]	865
[12daa43]	866	^(lanes.data[idx]){};
	867	}
[b798713]	868
[12daa43]	869	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue displaced %zu threads\n", displaced);
[c84b4be]	870
[12daa43]	871	// Allocate new array (uses realloc and memcpies the data)
	872	lanes.data = alloc( lanes.count, lanes.data`realloc );
[b798713]	873
[12daa43]	874	// Fix the moved data
	875	for( idx; (size_t)lanes.count ) {
	876	fix(lanes.data[idx]);
	877	}
[b798713]	878	}
	879
[12daa43]	880	fix_times(cltr);
[9cc3a18]	881
[12daa43]	882	reassign_cltr_id(cltr);
[a017ee7]	883
[12daa43]	884	// Make sure that everything is consistent
	885	/* paranoid */ check( cltr->ready_queue );
[dca5802]	886
[12daa43]	887	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue done\n");
	888	/* paranoid */ verify( ready_mutate_islocked() );
	889	}
	890	#endif
[8cd5434]	891
	892	#if !defined(__CFA_NO_STATISTICS__)
	893	unsigned cnt(const __ready_queue_t & this, unsigned idx) {
	894	/* paranoid */ verify(this.lanes.count > idx);
	895	return this.lanes.data[idx].cnt;
	896	}
	897	#endif
[0ee224b]	898
	899
	900	#if defined(CFA_HAVE_LINUX_LIBRSEQ)
	901	// No definition needed
	902	#elif defined(CFA_HAVE_LINUX_RSEQ_H)
	903
	904	#if defined( __x86_64 ) \|\| defined( __i386 )
	905	#define RSEQ_SIG 0x53053053
	906	#elif defined( __ARM_ARCH )
	907	#ifdef __ARMEB__
	908	#define RSEQ_SIG 0xf3def5e7 /* udf #24035 ; 0x5de3 (ARMv6+) */
	909	#else
	910	#define RSEQ_SIG 0xe7f5def3 /* udf #24035 ; 0x5de3 */
	911	#endif
	912	#endif
	913
	914	extern void __disable_interrupts_hard();
	915	extern void __enable_interrupts_hard();
	916
	917	void __kernel_raw_rseq_register (void) {
	918	/* paranoid */ verify( __cfaabi_rseq.cpu_id == RSEQ_CPU_ID_UNINITIALIZED );
	919
	920	// int ret = syscall(__NR_rseq, &__cfaabi_rseq, sizeof(struct rseq), 0, (sigset_t *)0p, _NSIG / 8);
	921	int ret = syscall(__NR_rseq, &__cfaabi_rseq, sizeof(struct rseq), 0, RSEQ_SIG);
	922	if(ret != 0) {
	923	int e = errno;
	924	switch(e) {
	925	case EINVAL: abort("KERNEL ERROR: rseq register invalid argument");
	926	case ENOSYS: abort("KERNEL ERROR: rseq register no supported");
	927	case EFAULT: abort("KERNEL ERROR: rseq register with invalid argument");
	928	case EBUSY : abort("KERNEL ERROR: rseq register already registered");
	929	case EPERM : abort("KERNEL ERROR: rseq register sig argument on unregistration does not match the signature received on registration");
	930	default: abort("KERNEL ERROR: rseq register unexpected return %d", e);
	931	}
	932	}
	933	}
	934
	935	void __kernel_raw_rseq_unregister(void) {
	936	/* paranoid */ verify( __cfaabi_rseq.cpu_id >= 0 );
	937
	938	// int ret = syscall(__NR_rseq, &__cfaabi_rseq, sizeof(struct rseq), RSEQ_FLAG_UNREGISTER, (sigset_t *)0p, _NSIG / 8);
	939	int ret = syscall(__NR_rseq, &__cfaabi_rseq, sizeof(struct rseq), RSEQ_FLAG_UNREGISTER, RSEQ_SIG);
	940	if(ret != 0) {
	941	int e = errno;
	942	switch(e) {
	943	case EINVAL: abort("KERNEL ERROR: rseq unregister invalid argument");
	944	case ENOSYS: abort("KERNEL ERROR: rseq unregister no supported");
	945	case EFAULT: abort("KERNEL ERROR: rseq unregister with invalid argument");
	946	case EBUSY : abort("KERNEL ERROR: rseq unregister already registered");
	947	case EPERM : abort("KERNEL ERROR: rseq unregister sig argument on unregistration does not match the signature received on registration");
	948	default: abort("KERNEL ERROR: rseq unregisteunexpected return %d", e);
	949	}
	950	}
	951	}
	952	#else
	953	// No definition needed
	954	#endif

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

Download in other formats:

Original Format