Context Navigation

ready_queue.cfa @ bac0ba8

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum

Last change on this file since bac0ba8 was c993b15, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
Changed RW lock to avoid hitting the global array on schedule.
Property mode set to `100644`
File size: 22.3 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 __CFA_DEBUG_PRINT_READY_QUEUE__
18
19	// #define USE_MPSC
20
21	#define USE_RELAXED_FIFO
22	// #define USE_WORK_STEALING
23
24	#include "bits/defs.hfa"
25	#include "kernel_private.hfa"
26
27	#define _GNU_SOURCE
28	#include "stdlib.hfa"
29	#include "math.hfa"
30
31	#include <unistd.h>
32
33	#include "ready_subqueue.hfa"
34
35	static const size_t cache_line_size = 64;
36
37	#if !defined(__CFA_NO_STATISTICS__)
38	#define __STATS(...) __VA_ARGS__
39	#else
40	#define __STATS(...)
41	#endif
42
43	// No overriden function, no environment variable, no define
44	// fall back to a magic number
45	#ifndef __CFA_MAX_PROCESSORS__
46	#define __CFA_MAX_PROCESSORS__ 1024
47	#endif
48
49	#if defined(USE_RELAXED_FIFO)
50	#define BIAS 4
51	#define READYQ_SHARD_FACTOR 4
52	#define SEQUENTIAL_SHARD 1
53	#elif defined(USE_WORK_STEALING)
54	#define READYQ_SHARD_FACTOR 2
55	#define SEQUENTIAL_SHARD 2
56	#else
57	#error no scheduling strategy selected
58	#endif
59
60	static inline struct $thread * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats));
61	static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats));
62	static inline struct $thread * search(struct cluster * cltr);
63	static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred);
64
65
66	// returns the maximum number of processors the RWLock support
67	__attribute__((weak)) unsigned __max_processors() {
68	const char * max_cores_s = getenv("CFA_MAX_PROCESSORS");
69	if(!max_cores_s) {
70	__cfadbg_print_nolock(ready_queue, "No CFA_MAX_PROCESSORS in ENV\n");
71	return __CFA_MAX_PROCESSORS__;
72	}
73
74	char * endptr = 0p;
75	long int max_cores_l = strtol(max_cores_s, &endptr, 10);
76	if(max_cores_l < 1 \|\| max_cores_l > 65535) {
77	__cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS out of range : %ld\n", max_cores_l);
78	return __CFA_MAX_PROCESSORS__;
79	}
80	if('\0' != *endptr) {
81	__cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS not a decimal number : %s\n", max_cores_s);
82	return __CFA_MAX_PROCESSORS__;
83	}
84
85	return max_cores_l;
86	}
87
88	//=======================================================================
89	// Cluster wide reader-writer lock
90	//=======================================================================
91	void ?{}(__scheduler_RWLock_t & this) {
92	this.max = __max_processors();
93	this.alloc = 0;
94	this.ready = 0;
95	this.data = alloc(this.max);
96	this.write_lock = false;
97
98	/paranoid/ verify(__atomic_is_lock_free(sizeof(this.alloc), &this.alloc));
99	/paranoid/ verify(__atomic_is_lock_free(sizeof(this.ready), &this.ready));
100
101	}
102	void ^?{}(__scheduler_RWLock_t & this) {
103	free(this.data);
104	}
105
106
107	//=======================================================================
108	// Lock-Free registering/unregistering of threads
109	unsigned register_proc_id( void ) with(*__scheduler_lock) {
110	__cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
111	bool * handle = (bool *)&kernelTLS().sched_lock;
112
113	// Step - 1 : check if there is already space in the data
114	uint_fast32_t s = ready;
115
116	// Check among all the ready
117	for(uint_fast32_t i = 0; i < s; i++) {
118	bool * volatile * cell = (bool * volatile *)&data[i]; // Cforall is bugged and the double volatiles causes problems
119	/* paranoid / verify( handle != cell );
120
121	bool * null = 0p; // Re-write every loop since compare thrashes it
122	if( __atomic_load_n(cell, (int)__ATOMIC_RELAXED) == null
123	&& __atomic_compare_exchange_n( cell, &null, handle, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
124	/* paranoid */ verify(i < ready);
125	/* paranoid */ verify( (kernelTLS().sched_id = i, true) );
126	return i;
127	}
128	}
129
130	if(max <= alloc) abort("Trying to create more than %ud processors", __scheduler_lock->max);
131
132	// Step - 2 : F&A to get a new spot in the array.
133	uint_fast32_t n = __atomic_fetch_add(&alloc, 1, __ATOMIC_SEQ_CST);
134	if(max <= n) abort("Trying to create more than %ud processors", __scheduler_lock->max);
135
136	// Step - 3 : Mark space as used and then publish it.
137	data[n] = handle;
138	while() {
139	unsigned copy = n;
140	if( __atomic_load_n(&ready, __ATOMIC_RELAXED) == n
141	&& __atomic_compare_exchange_n(&ready, &copy, n + 1, true, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST))
142	break;
143	Pause();
144	}
145
146	__cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n);
147
148	// Return new spot.
149	/* paranoid */ verify(n < ready);
150	/* paranoid */ verify( (kernelTLS().sched_id = n, true) );
151	return n;
152	}
153
154	void unregister_proc_id( unsigned id ) with(*__scheduler_lock) {
155	/* paranoid */ verify(id < ready);
156	/* paranoid */ verify(id == kernelTLS().sched_id);
157	/* paranoid */ verify(data[id] == &kernelTLS().sched_lock);
158
159	bool * volatile * cell = (bool * volatile *)&data[id]; // Cforall is bugged and the double volatiles causes problems
160
161	__atomic_store_n(cell, 0p, __ATOMIC_RELEASE);
162
163	__cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc);
164	}
165
166	//-----------------------------------------------------------------------
167	// Writer side : acquire when changing the ready queue, e.g. adding more
168	// queues or removing them.
169	uint_fast32_t ready_mutate_lock( void ) with(*__scheduler_lock) {
170	/* paranoid */ verify( ! __preemption_enabled() );
171	/* paranoid */ verify( ! kernelTLS().sched_lock );
172
173	// Step 1 : lock global lock
174	// It is needed to avoid processors that register mid Critical-Section
175	// to simply lock their own lock and enter.
176	__atomic_acquire( &write_lock );
177
178	// Step 2 : lock per-proc lock
179	// Processors that are currently being registered aren't counted
180	// but can't be in read_lock or in the critical section.
181	// All other processors are counted
182	uint_fast32_t s = ready;
183	for(uint_fast32_t i = 0; i < s; i++) {
184	volatile bool * llock = data[i];
185	if(llock) __atomic_acquire( llock );
186	}
187
188	/* paranoid */ verify( ! __preemption_enabled() );
189	return s;
190	}
191
192	void ready_mutate_unlock( uint_fast32_t last_s ) with(*__scheduler_lock) {
193	/* paranoid */ verify( ! __preemption_enabled() );
194
195	// Step 1 : release local locks
196	// This must be done while the global lock is held to avoid
197	// threads that where created mid critical section
198	// to race to lock their local locks and have the writer
199	// immidiately unlock them
200	// Alternative solution : return s in write_lock and pass it to write_unlock
201	for(uint_fast32_t i = 0; i < last_s; i++) {
202	volatile bool * llock = data[i];
203	if(llock) __atomic_store_n(llock, (bool)false, __ATOMIC_RELEASE);
204	}
205
206	// Step 2 : release global lock
207	/paranoid/ assert(true == write_lock);
208	__atomic_store_n(&write_lock, (bool)false, __ATOMIC_RELEASE);
209
210	/* paranoid */ verify( ! __preemption_enabled() );
211	}
212
213	//=======================================================================
214	// Cforall Ready Queue used for scheduling
215	//=======================================================================
216	void ?{}(__ready_queue_t & this) with (this) {
217	lanes.data = 0p;
218	lanes.tscs = 0p;
219	lanes.count = 0;
220	}
221
222	void ^?{}(__ready_queue_t & this) with (this) {
223	verify( SEQUENTIAL_SHARD == lanes.count );
224	free(lanes.data);
225	free(lanes.tscs);
226	}
227
228	//-----------------------------------------------------------------------
229	#if defined(USE_RELAXED_FIFO)
230	//-----------------------------------------------------------------------
231	// get index from random number with or without bias towards queues
232	static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred) {
233	unsigned i;
234	bool local;
235	unsigned rlow = r % BIAS;
236	unsigned rhigh = r / BIAS;
237	if((0 != rlow) && preferred >= 0) {
238	// (BIAS - 1) out of BIAS chances
239	// Use perferred queues
240	i = preferred + (rhigh % READYQ_SHARD_FACTOR);
241	local = true;
242	}
243	else {
244	// 1 out of BIAS chances
245	// Use all queues
246	i = rhigh;
247	local = false;
248	}
249	return [i, local];
250	}
251
252	__attribute__((hot)) void push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
253	__cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
254
255	const bool external = (!kernelTLS().this_processor) \|\| (cltr != kernelTLS().this_processor->cltr);
256	/* paranoid */ verify(external \|\| kernelTLS().this_processor->rdq.id < lanes.count );
257
258	// write timestamp
259	thrd->link.ts = rdtscl();
260
261	bool local;
262	int preferred = external ? -1 : kernelTLS().this_processor->rdq.id;
263
264	// Try to pick a lane and lock it
265	unsigned i;
266	do {
267	// Pick the index of a lane
268	unsigned r = __tls_rand_fwd();
269	[i, local] = idx_from_r(r, preferred);
270
271	i %= __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
272
273	#if !defined(__CFA_NO_STATISTICS__)
274	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
275	else if(local) __tls_stats()->ready.push.local.attempt++;
276	else __tls_stats()->ready.push.share.attempt++;
277	#endif
278
279	#if defined(USE_MPSC)
280	// mpsc always succeeds
281	} while( false );
282	#else
283	// If we can't lock it retry
284	} while( !__atomic_try_acquire( &lanes.data[i].lock ) );
285	#endif
286
287	// Actually push it
288	push(lanes.data[i], thrd);
289
290	#if !defined(USE_MPSC)
291	// Unlock and return
292	__atomic_unlock( &lanes.data[i].lock );
293	#endif
294
295	// Mark the current index in the tls rng instance as having an item
296	__tls_rand_advance_bck();
297
298	__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);
299
300	// Update statistics
301	#if !defined(__CFA_NO_STATISTICS__)
302	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
303	else if(local) __tls_stats()->ready.push.local.success++;
304	else __tls_stats()->ready.push.share.success++;
305	#endif
306	}
307
308	// Pop from the ready queue from a given cluster
309	__attribute__((hot)) $thread * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
310	/* paranoid */ verify( lanes.count > 0 );
311	/* paranoid */ verify( kernelTLS().this_processor );
312	/* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
313
314	unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
315	int preferred = kernelTLS().this_processor->rdq.id;
316
317
318	// As long as the list is not empty, try finding a lane that isn't empty and pop from it
319	for(25) {
320	// Pick two lists at random
321	unsigned ri = __tls_rand_bck();
322	unsigned rj = __tls_rand_bck();
323
324	unsigned i, j;
325	__attribute__((unused)) bool locali, localj;
326	[i, locali] = idx_from_r(ri, preferred);
327	[j, localj] = idx_from_r(rj, preferred);
328
329	i %= count;
330	j %= count;
331
332	// try popping from the 2 picked lists
333	struct $thread * thrd = try_pop(cltr, i, j __STATS(, *(locali \|\| localj ? &__tls_stats()->ready.pop.local : &__tls_stats()->ready.pop.help)));
334	if(thrd) {
335	return thrd;
336	}
337	}
338
339	// All lanes where empty return 0p
340	return 0p;
341	}
342
343	__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) { return pop_fast(cltr); }
344	__attribute__((hot)) struct $thread * pop_search(struct cluster * cltr) {
345	return search(cltr);
346	}
347	#endif
348	#if defined(USE_WORK_STEALING)
349	__attribute__((hot)) void push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
350	__cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
351
352	const bool external = (!kernelTLS().this_processor) \|\| (cltr != kernelTLS().this_processor->cltr);
353	/* paranoid */ verify(external \|\| kernelTLS().this_processor->rdq.id < lanes.count );
354
355	// write timestamp
356	thrd->link.ts = rdtscl();
357
358	// Try to pick a lane and lock it
359	unsigned i;
360	do {
361	#if !defined(__CFA_NO_STATISTICS__)
362	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
363	else __tls_stats()->ready.push.local.attempt++;
364	#endif
365
366	if(unlikely(external)) {
367	i = __tls_rand() % lanes.count;
368	}
369	else {
370	processor * proc = kernelTLS().this_processor;
371	unsigned r = proc->rdq.its++;
372	i = proc->rdq.id + (r % READYQ_SHARD_FACTOR);
373	}
374
375
376	#if defined(USE_MPSC)
377	// mpsc always succeeds
378	} while( false );
379	#else
380	// If we can't lock it retry
381	} while( !__atomic_try_acquire( &lanes.data[i].lock ) );
382	#endif
383
384	// Actually push it
385	push(lanes.data[i], thrd);
386
387	#if !defined(USE_MPSC)
388	// Unlock and return
389	__atomic_unlock( &lanes.data[i].lock );
390	#endif
391
392	#if !defined(__CFA_NO_STATISTICS__)
393	if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
394	else __tls_stats()->ready.push.local.success++;
395	#endif
396
397	__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);
398	}
399
400	// Pop from the ready queue from a given cluster
401	__attribute__((hot)) $thread * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
402	/* paranoid */ verify( lanes.count > 0 );
403	/* paranoid */ verify( kernelTLS().this_processor );
404	/* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
405
406	processor * proc = kernelTLS().this_processor;
407
408	if(proc->rdq.target == -1u) {
409	proc->rdq.target = __tls_rand() % lanes.count;
410	unsigned it1 = proc->rdq.itr;
411	unsigned it2 = proc->rdq.itr + 1;
412	unsigned idx1 = proc->rdq.id + (it1 % READYQ_SHARD_FACTOR);
413	unsigned idx2 = proc->rdq.id + (it2 % READYQ_SHARD_FACTOR);
414	unsigned long long tsc1 = ts(lanes.data[idx1]);
415	unsigned long long tsc2 = ts(lanes.data[idx2]);
416	proc->rdq.cutoff = min(tsc1, tsc2);
417	if(proc->rdq.cutoff == 0) proc->rdq.cutoff = -1ull;
418	}
419	else {
420	unsigned target = proc->rdq.target;
421	proc->rdq.target = -1u;
422	if(lanes.tscs[target].tv < proc->rdq.cutoff) {
423	$thread * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
424	if(t) return t;
425	}
426	}
427
428	for(READYQ_SHARD_FACTOR) {
429	unsigned i = proc->rdq.id + (--proc->rdq.itr % READYQ_SHARD_FACTOR);
430	if($thread * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
431	}
432	return 0p;
433	}
434
435	__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
436	unsigned i = __tls_rand() % lanes.count;
437	return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));
438	}
439
440	__attribute__((hot)) struct $thread * pop_search(struct cluster * cltr) with (cltr->ready_queue) {
441	return search(cltr);
442	}
443	#endif
444
445	//=======================================================================
446	// Various Ready Queue utilities
447	//=======================================================================
448	// these function work the same or almost the same
449	// whether they are using work-stealing or relaxed fifo scheduling
450
451	//-----------------------------------------------------------------------
452	// try to pop from a lane given by index w
453	static inline struct $thread * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
454	__STATS( stats.attempt++; )
455
456	// Get relevant elements locally
457	__intrusive_lane_t & lane = lanes.data[w];
458
459	// If list looks empty retry
460	if( is_empty(lane) ) {
461	__STATS( stats.espec++; )
462	return 0p;
463	}
464
465	// If we can't get the lock retry
466	if( !__atomic_try_acquire(&lane.lock) ) {
467	__STATS( stats.elock++; )
468	return 0p;
469	}
470
471	// If list is empty, unlock and retry
472	if( is_empty(lane) ) {
473	__atomic_unlock(&lane.lock);
474	__STATS( stats.eempty++; )
475	return 0p;
476	}
477
478	// Actually pop the list
479	struct $thread * thrd;
480	thrd = pop(lane);
481
482	/* paranoid */ verify(thrd);
483	/* paranoid */ verify(lane.lock);
484
485	// Unlock and return
486	__atomic_unlock(&lane.lock);
487
488	// Update statistics
489	__STATS( stats.success++; )
490
491	#if defined(USE_WORK_STEALING)
492	lanes.tscs[w].tv = thrd->link.ts;
493	#endif
494
495	// return the popped thread
496	return thrd;
497	}
498
499	//-----------------------------------------------------------------------
500	// try to pop from any lanes making sure you don't miss any threads push
501	// before the start of the function
502	static inline struct $thread * search(struct cluster * cltr) with (cltr->ready_queue) {
503	/* paranoid */ verify( lanes.count > 0 );
504	unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
505	unsigned offset = __tls_rand();
506	for(i; count) {
507	unsigned idx = (offset + i) % count;
508	struct $thread * thrd = try_pop(cltr, idx __STATS(, __tls_stats()->ready.pop.search));
509	if(thrd) {
510	return thrd;
511	}
512	}
513
514	// All lanes where empty return 0p
515	return 0p;
516	}
517
518	//-----------------------------------------------------------------------
519	// Check that all the intrusive queues in the data structure are still consistent
520	static void check( __ready_queue_t & q ) with (q) {
521	#if defined(__CFA_WITH_VERIFY__) && !defined(USE_MPSC)
522	{
523	for( idx ; lanes.count ) {
524	__intrusive_lane_t & sl = lanes.data[idx];
525	assert(!lanes.data[idx].lock);
526
527	assert(head(sl)->link.prev == 0p );
528	assert(head(sl)->link.next->link.prev == head(sl) );
529	assert(tail(sl)->link.next == 0p );
530	assert(tail(sl)->link.prev->link.next == tail(sl) );
531
532	if(is_empty(sl)) {
533	assert(tail(sl)->link.prev == head(sl));
534	assert(head(sl)->link.next == tail(sl));
535	} else {
536	assert(tail(sl)->link.prev != head(sl));
537	assert(head(sl)->link.next != tail(sl));
538	}
539	}
540	}
541	#endif
542	}
543
544	//-----------------------------------------------------------------------
545	// Given 2 indexes, pick the list with the oldest push an try to pop from it
546	static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
547	// Pick the bet list
548	int w = i;
549	if( __builtin_expect(!is_empty(lanes.data[j]), true) ) {
550	w = (ts(lanes.data[i]) < ts(lanes.data[j])) ? i : j;
551	}
552
553	return try_pop(cltr, w __STATS(, stats));
554	}
555
556	// Call this function of the intrusive list was moved using memcpy
557	// fixes the list so that the pointers back to anchors aren't left dangling
558	static inline void fix(__intrusive_lane_t & ll) {
559	#if !defined(USE_MPSC)
560	// if the list is not empty then follow he pointer and fix its reverse
561	if(!is_empty(ll)) {
562	head(ll)->link.next->link.prev = head(ll);
563	tail(ll)->link.prev->link.next = tail(ll);
564	}
565	// Otherwise just reset the list
566	else {
567	verify(tail(ll)->link.next == 0p);
568	tail(ll)->link.prev = head(ll);
569	head(ll)->link.next = tail(ll);
570	verify(head(ll)->link.prev == 0p);
571	}
572	#endif
573	}
574
575	static void assign_list(unsigned & value, dlist(processor, processor) & list, unsigned count) {
576	processor * it = &list`first;
577	for(unsigned i = 0; i < count; i++) {
578	/* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
579	it->rdq.id = value;
580	it->rdq.target = -1u;
581	value += READYQ_SHARD_FACTOR;
582	it = &(*it)`next;
583	}
584	}
585
586	static void reassign_cltr_id(struct cluster * cltr) {
587	unsigned preferred = 0;
588	assign_list(preferred, cltr->procs.actives, cltr->procs.total - cltr->procs.idle);
589	assign_list(preferred, cltr->procs.idles , cltr->procs.idle );
590	}
591
592	static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
593	#if defined(USE_WORK_STEALING)
594	lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
595	for(i; lanes.count) {
596	lanes.tscs[i].tv = ts(lanes.data[i]);
597	}
598	#endif
599	}
600
601	// Grow the ready queue
602	void ready_queue_grow(struct cluster * cltr) {
603	size_t ncount;
604	int target = cltr->procs.total;
605
606	/* paranoid */ verify( ready_mutate_islocked() );
607	__cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue\n");
608
609	// Make sure that everything is consistent
610	/* paranoid */ check( cltr->ready_queue );
611
612	// grow the ready queue
613	with( cltr->ready_queue ) {
614	// Find new count
615	// Make sure we always have atleast 1 list
616	if(target >= 2) {
617	ncount = target * READYQ_SHARD_FACTOR;
618	} else {
619	ncount = SEQUENTIAL_SHARD;
620	}
621
622	// Allocate new array (uses realloc and memcpies the data)
623	lanes.data = alloc( ncount, lanes.data`realloc );
624
625	// Fix the moved data
626	for( idx; (size_t)lanes.count ) {
627	fix(lanes.data[idx]);
628	}
629
630	// Construct new data
631	for( idx; (size_t)lanes.count ~ ncount) {
632	(lanes.data[idx]){};
633	}
634
635	// Update original
636	lanes.count = ncount;
637	}
638
639	fix_times(cltr);
640
641	reassign_cltr_id(cltr);
642
643	// Make sure that everything is consistent
644	/* paranoid */ check( cltr->ready_queue );
645
646	__cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue done\n");
647
648	/* paranoid */ verify( ready_mutate_islocked() );
649	}
650
651	// Shrink the ready queue
652	void ready_queue_shrink(struct cluster * cltr) {
653	/* paranoid */ verify( ready_mutate_islocked() );
654	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue\n");
655
656	// Make sure that everything is consistent
657	/* paranoid */ check( cltr->ready_queue );
658
659	int target = cltr->procs.total;
660
661	with( cltr->ready_queue ) {
662	// Remember old count
663	size_t ocount = lanes.count;
664
665	// Find new count
666	// Make sure we always have atleast 1 list
667	lanes.count = target >= 2 ? target * READYQ_SHARD_FACTOR: SEQUENTIAL_SHARD;
668	/* paranoid */ verify( ocount >= lanes.count );
669	/* paranoid / verify( lanes.count == target READYQ_SHARD_FACTOR \|\| target < 2 );
670
671	// for printing count the number of displaced threads
672	#if defined(__CFA_DEBUG_PRINT__) \|\| defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
673	__attribute__((unused)) size_t displaced = 0;
674	#endif
675
676	// redistribute old data
677	for( idx; (size_t)lanes.count ~ ocount) {
678	// Lock is not strictly needed but makes checking invariants much easier
679	__attribute__((unused)) bool locked = __atomic_try_acquire(&lanes.data[idx].lock);
680	verify(locked);
681
682	// As long as we can pop from this lane to push the threads somewhere else in the queue
683	while(!is_empty(lanes.data[idx])) {
684	struct $thread * thrd;
685	thrd = pop(lanes.data[idx]);
686
687	push(cltr, thrd);
688
689	// for printing count the number of displaced threads
690	#if defined(__CFA_DEBUG_PRINT__) \|\| defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
691	displaced++;
692	#endif
693	}
694
695	// Unlock the lane
696	__atomic_unlock(&lanes.data[idx].lock);
697
698	// TODO print the queue statistics here
699
700	^(lanes.data[idx]){};
701	}
702
703	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue displaced %zu threads\n", displaced);
704
705	// Allocate new array (uses realloc and memcpies the data)
706	lanes.data = alloc( lanes.count, lanes.data`realloc );
707
708	// Fix the moved data
709	for( idx; (size_t)lanes.count ) {
710	fix(lanes.data[idx]);
711	}
712	}
713
714	fix_times(cltr);
715
716	reassign_cltr_id(cltr);
717
718	// Make sure that everything is consistent
719	/* paranoid */ check( cltr->ready_queue );
720
721	__cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue done\n");
722	/* paranoid */ verify( ready_mutate_islocked() );
723	}

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

Download in other formats:

Original Format