source: src/libcfa/concurrency/monitor.c@ 561f730

ADT aaron-thesis arm-eh ast-experimental cleanup-dtors deferred_resn demangler enum forall-pointer-decay jacob/cs343-translation jenkins-sandbox new-ast new-ast-unique-expr new-env no_list persistent-indexer pthread-emulation qualifiedEnum resolv-new with_gc
Last change on this file since 561f730 was a933dcf4, checked in by Thierry Delisle <tdelisle@…>, 9 years ago
  • updated internal scheduler test for multi monitors
  • fixed branding for monitors
  • Property mode set to 100644
File size: 11.2 KB
Line 
1// -*- Mode: CFA -*-
2//
3// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo
4//
5// The contents of this file are covered under the licence agreement in the
6// file "LICENCE" distributed with Cforall.
7//
8// monitor_desc.c --
9//
10// Author : Thierry Delisle
11// Created On : Thd Feb 23 12:27:26 2017
12// Last Modified By : Thierry Delisle
13// Last Modified On : --
14// Update Count : 0
15//
16
17#include "monitor"
18
19#include <stdlib>
20
21#include "kernel_private.h"
22#include "libhdr.h"
23
24//-----------------------------------------------------------------------------
25// Forward declarations
26static inline void set_owner( monitor_desc * this, thread_desc * owner );
27static inline thread_desc * next_thread( monitor_desc * this );
28
29static inline void lock_all( spinlock ** locks, unsigned short count );
30static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count );
31static inline void unlock_all( spinlock ** locks, unsigned short count );
32static inline void unlock_all( monitor_desc ** locks, unsigned short count );
33
34static inline void save_recursion ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count );
35static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count );
36
37static inline thread_desc * check_condition( __condition_criterion_t * );
38static inline void brand_condition( condition * );
39static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val );
40
41//-----------------------------------------------------------------------------
42// Enter/Leave routines
43
44
45extern "C" {
46 void __enter_monitor_desc(monitor_desc * this) {
47 lock( &this->lock );
48 thread_desc * thrd = this_thread();
49
50 LIB_DEBUG_PRINT_SAFE("%p Entering %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
51
52 if( !this->owner ) {
53 //No one has the monitor, just take it
54 set_owner( this, thrd );
55 }
56 else if( this->owner == thrd) {
57 //We already have the monitor, just not how many times we took it
58 assert( this->recursion > 0 );
59 this->recursion += 1;
60 }
61 else {
62 //Some one else has the monitor, wait in line for it
63 append( &this->entry_queue, thrd );
64 ScheduleInternal( &this->lock );
65
66 //ScheduleInternal will unlock spinlock, no need to unlock ourselves
67 return;
68 }
69
70 unlock( &this->lock );
71 return;
72 }
73
74 // leave pseudo code :
75 // TODO
76 void __leave_monitor_desc(monitor_desc * this) {
77 lock( &this->lock );
78
79 thread_desc * thrd = this_thread();
80
81 LIB_DEBUG_PRINT_SAFE("%p Leaving %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
82 assertf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i)", thrd, this->owner, this->recursion );
83
84 //Leaving a recursion level, decrement the counter
85 this->recursion -= 1;
86
87 //If we haven't left the last level of recursion
88 //it means we don't need to do anything
89 if( this->recursion != 0) {
90 unlock( &this->lock );
91 return;
92 }
93
94 thread_desc * new_owner = next_thread( this );
95
96 //We can now let other threads in safely
97 unlock( &this->lock );
98
99 //We need to wake-up the thread
100 ScheduleThread( new_owner );
101 }
102}
103
104static inline void enter(monitor_desc ** monitors, int count) {
105 for(int i = 0; i < count; i++) {
106 __enter_monitor_desc( monitors[i] );
107 }
108}
109
110static inline void leave(monitor_desc ** monitors, int count) {
111 for(int i = count - 1; i >= 0; i--) {
112 __leave_monitor_desc( monitors[i] );
113 }
114}
115
116void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ) {
117 this->m = m;
118 this->count = count;
119 qsort(this->m, count);
120 enter( this->m, this->count );
121
122 this->prev_mntrs = this_thread()->current_monitors;
123 this->prev_count = this_thread()->current_monitor_count;
124
125 this_thread()->current_monitors = m;
126 this_thread()->current_monitor_count = count;
127}
128
129void ^?{}( monitor_guard_t * this ) {
130 leave( this->m, this->count );
131
132 this_thread()->current_monitors = this->prev_mntrs;
133 this_thread()->current_monitor_count = this->prev_count;
134}
135
136void debug_break() __attribute__(( noinline ))
137{
138
139}
140
141//-----------------------------------------------------------------------------
142// Internal scheduling
143void wait( condition * this ) {
144 LIB_DEBUG_PRINT_SAFE("Waiting\n");
145
146 brand_condition( this );
147
148 //Check that everything is as expected
149 assertf( this->monitors != NULL, "Waiting with no monitors (%p)", this->monitors );
150 assertf( this->monitor_count != 0, "Waiting with 0 monitors (%i)", this->monitor_count );
151
152 unsigned short count = this->monitor_count;
153 unsigned int recursions[ count ]; //Save the current recursion levels to restore them later
154 spinlock * locks [ count ]; //We need to pass-in an array of locks to ScheduleInternal
155
156 LIB_DEBUG_PRINT_SAFE("count %i\n", count);
157
158 __condition_node_t waiter;
159 waiter.waiting_thread = this_thread();
160 waiter.count = count;
161 waiter.next = NULL;
162
163 __condition_criterion_t criteria[count];
164 for(int i = 0; i < count; i++) {
165 criteria[i].ready = false;
166 criteria[i].target = this->monitors[i];
167 criteria[i].owner = &waiter;
168 criteria[i].next = NULL;
169 LIB_DEBUG_PRINT_SAFE( "Criterion %p\n", &criteria[i] );
170 }
171
172 waiter.criteria = criteria;
173 append( &this->blocked, &waiter );
174
175 lock_all( this->monitors, locks, count );
176 save_recursion( this->monitors, recursions, count );
177 //DON'T unlock, ask the kernel to do it
178
179 //Find the next thread(s) to run
180 unsigned short thread_count = 0;
181 thread_desc * threads[ count ];
182 for(int i = 0; i < count; i++) {
183 threads[i] = 0;
184 }
185
186 debug_break();
187
188 for( int i = 0; i < count; i++) {
189 thread_desc * new_owner = next_thread( this->monitors[i] );
190 thread_count = insert_unique( threads, thread_count, new_owner );
191 }
192
193 debug_break();
194
195 LIB_DEBUG_PRINT_SAFE("Will unblock: ");
196 for(int i = 0; i < thread_count; i++) {
197 LIB_DEBUG_PRINT_SAFE("%p ", threads[i]);
198 }
199 LIB_DEBUG_PRINT_SAFE("\n");
200
201 // Everything is ready to go to sleep
202 ScheduleInternal( locks, count, threads, thread_count );
203
204
205 //WE WOKE UP
206
207
208 //We are back, restore the owners and recursions
209 lock_all( locks, count );
210 restore_recursion( this->monitors, recursions, count );
211 unlock_all( locks, count );
212}
213
214void signal( condition * this ) {
215 if( !this->blocked.head ) {
216 LIB_DEBUG_PRINT_SAFE("Nothing to signal\n");
217 return;
218 }
219
220 //Check that everything is as expected
221 assert( this->monitors );
222 assert( this->monitor_count != 0 );
223
224 unsigned short count = this->monitor_count;
225
226 LIB_DEBUG_DO(
227 thread_desc * this_thrd = this_thread();
228 if ( this->monitor_count != this_thrd->current_monitor_count ) {
229 abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", this, this->monitor_count, this_thrd->current_monitor_count );
230 } // if
231
232 for(int i = 0; i < this->monitor_count; i++) {
233 if ( this->monitors[i] != this_thrd->current_monitors[i] ) {
234 abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->current_monitors[i] );
235 } // if
236 }
237 );
238
239 lock_all( this->monitors, NULL, count );
240 LIB_DEBUG_PRINT_SAFE("Signalling");
241
242 __condition_node_t * node = pop_head( &this->blocked );
243 for(int i = 0; i < count; i++) {
244 __condition_criterion_t * crit = &node->criteria[i];
245 LIB_DEBUG_PRINT_SAFE(" %p", crit->target);
246 assert( !crit->ready );
247 push( &crit->target->signal_stack, crit );
248 }
249
250 LIB_DEBUG_PRINT_SAFE("\n");
251
252 unlock_all( this->monitors, count );
253}
254
255//-----------------------------------------------------------------------------
256// Utilities
257
258static inline void set_owner( monitor_desc * this, thread_desc * owner ) {
259 //Pass the monitor appropriately
260 this->owner = owner;
261
262 //We are passing the monitor to someone else, which means recursion level is not 0
263 this->recursion = owner ? 1 : 0;
264}
265
266static inline thread_desc * next_thread( monitor_desc * this ) {
267 //Check the signaller stack
268 __condition_criterion_t * urgent = pop( &this->signal_stack );
269 if( urgent ) {
270 //The signaller stack is not empty,
271 //regardless of if we are ready to baton pass,
272 //we need to set the monitor as in use
273 set_owner( this, urgent->owner->waiting_thread );
274
275 return check_condition( urgent );
276 }
277
278 // No signaller thread
279 // Get the next thread in the entry_queue
280 thread_desc * new_owner = pop_head( &this->entry_queue );
281 set_owner( this, new_owner );
282
283 return new_owner;
284}
285
286static inline void lock_all( spinlock ** locks, unsigned short count ) {
287 for( int i = 0; i < count; i++ ) {
288 lock( locks[i] );
289 }
290}
291
292static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count ) {
293 for( int i = 0; i < count; i++ ) {
294 spinlock * l = &source[i]->lock;
295 lock( l );
296 if(locks) locks[i] = l;
297 }
298}
299
300static inline void unlock_all( spinlock ** locks, unsigned short count ) {
301 for( int i = 0; i < count; i++ ) {
302 unlock( locks[i] );
303 }
304}
305
306static inline void unlock_all( monitor_desc ** locks, unsigned short count ) {
307 for( int i = 0; i < count; i++ ) {
308 unlock( &locks[i]->lock );
309 }
310}
311
312
313static inline void save_recursion ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count ) {
314 for( int i = 0; i < count; i++ ) {
315 recursions[i] = ctx[i]->recursion;
316 }
317}
318
319static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count ) {
320 for( int i = 0; i < count; i++ ) {
321 ctx[i]->recursion = recursions[i];
322 }
323}
324
325// Function has 2 different behavior
326// 1 - Marks a monitors as being ready to run
327// 2 - Checks if all the monitors are ready to run
328// if so return the thread to run
329static inline thread_desc * check_condition( __condition_criterion_t * target ) {
330 __condition_node_t * node = target->owner;
331 unsigned short count = node->count;
332 __condition_criterion_t * criteria = node->criteria;
333
334 bool ready2run = true;
335
336 for( int i = 0; i < count; i++ ) {
337 LIB_DEBUG_PRINT_SAFE( "Checking %p for %p\n", &criteria[i], target );
338 if( &criteria[i] == target ) {
339 criteria[i].ready = true;
340 LIB_DEBUG_PRINT_SAFE( "True\n" );
341 }
342
343 ready2run = criteria[i].ready && ready2run;
344 }
345
346 LIB_DEBUG_PRINT_SAFE( "Runing %i\n", ready2run );
347 return ready2run ? node->waiting_thread : NULL;
348}
349
350static inline void brand_condition( condition * this ) {
351 thread_desc * thrd = this_thread();
352 if( !this->monitors ) {
353 LIB_DEBUG_PRINT_SAFE("Branding\n");
354 assertf( thrd->current_monitors != NULL, "No current monitor to brand condition", thrd->current_monitors );
355 this->monitor_count = thrd->current_monitor_count;
356
357 this->monitors = malloc( this->monitor_count * sizeof( *this->monitors ) );
358 for( int i = 0; i < this->monitor_count; i++ ) {
359 this->monitors[i] = thrd->current_monitors[i];
360 }
361 }
362}
363
364static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val ) {
365 if( !val ) return end;
366
367 for(int i = 0; i <= end; i++) {
368 if( thrds[i] == val ) return end;
369 }
370
371 thrds[end] = val;
372 return end + 1;
373}
374
375void ?{}( __condition_blocked_queue_t * this ) {
376 this->head = NULL;
377 this->tail = &this->head;
378}
379
380void append( __condition_blocked_queue_t * this, __condition_node_t * c ) {
381 assert(this->tail != NULL);
382 *this->tail = c;
383 this->tail = &c->next;
384}
385
386__condition_node_t * pop_head( __condition_blocked_queue_t * this ) {
387 __condition_node_t * head = this->head;
388 if( head ) {
389 this->head = head->next;
390 if( !head->next ) {
391 this->tail = &this->head;
392 }
393 head->next = NULL;
394 }
395 return head;
396}
Note: See TracBrowser for help on using the repository browser.