Context Navigation

ready_queue.cfa @ fcd65ca

ADTast-experimentalenumforall-pointer-decaypthread-emulationqualifiedEnum

Last change on this file since fcd65ca was fcd65ca, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
Fixed incorrect access into lanes.help
Property mode set to `100644`
File size: 32.1 KB

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

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

Download in other formats:

Original Format