source: libcfa/src/concurrency/monitor.cfa@ 10a9479d

Last change on this file since 10a9479d was 1cd2839, checked in by Peter A. Buhr <pabuhr@…>, 11 months ago

test patch for respecting lexical priority in waitfor statement
  • Property mode set to 100644
File size: 37.0 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2016 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// monitor.cfa --
8//
9// Author : Thierry Delisle
10// Created On : Thd Feb 23 12:27:26 2017
11// Last Modified By : Peter A. Buhr
12// Last Modified On : Thu Nov 21 08:31:55 2024
13// Update Count : 18
14//
15
16#define __cforall_thread__
17
18#include "monitor.hfa"
19
20#include <stdlib.hfa>
21#include <inttypes.h>
22
23#include "kernel/private.hfa"
24
25#include "bits/algorithm.hfa"
26
27//-----------------------------------------------------------------------------
28// Forward declarations
29static inline void __set_owner ( monitor$ * this, thread$ * owner );
30static inline void __set_owner ( monitor$ * storage [], __lock_size_t count, thread$ * owner );
31static inline void set_mask ( monitor$ * storage [], __lock_size_t count, const __waitfor_mask_t & mask );
32static inline void reset_mask( monitor$ * this );
33
34static inline thread$ * next_thread( monitor$ * this );
35static inline bool is_accepted( monitor$ * this, const __monitor_group_t & monitors );
36
37static inline void lock_all ( __spinlock_t * locks [], __lock_size_t count );
38static inline void lock_all ( monitor$ * source [], __spinlock_t * /*out*/ locks [], __lock_size_t count );
39static inline void unlock_all( __spinlock_t * locks [], __lock_size_t count );
40static inline void unlock_all( monitor$ * locks [], __lock_size_t count );
41
42static inline void save ( monitor$ * ctx [], __lock_size_t count, __spinlock_t * locks [], unsigned int /*out*/ recursions [], __waitfor_mask_t /*out*/ masks [] );
43static inline void restore( monitor$ * ctx [], __lock_size_t count, __spinlock_t * locks [], unsigned int /*in */ recursions [], __waitfor_mask_t /*in */ masks [] );
44
45static inline void ?{}(__condition_node_t & this, thread$ * waiting_thread, __lock_size_t count, uintptr_t user_info );
46static inline void ?{}(__condition_criterion_t & this );
47static inline void ?{}(__condition_criterion_t & this, monitor$ * target, __condition_node_t * owner );
48
49static inline void init ( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] );
50static inline void init_push( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] );
51
52static inline thread$ * check_condition ( __condition_criterion_t * );
53static inline void brand_condition ( condition & );
54static inline [thread$ *, int] search_entry_queue( const __waitfor_mask_t &, monitor$ * monitors [], __lock_size_t count );
55
56forall(T & | sized( T ))
57static inline __lock_size_t insert_unique( T * array [], __lock_size_t & size, T * val );
58static inline __lock_size_t count_max ( const __waitfor_mask_t & mask );
59static inline __lock_size_t aggregate ( monitor$ * storage [], const __waitfor_mask_t & mask );
60
61//-----------------------------------------------------------------------------
62// Useful defines
63#define wait_ctx(thrd, user_info) /* Create the necessary information to use the signaller stack */ \
64 __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \
65 __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \
66 init( count, monitors, waiter, criteria ); /* Link everything together */ \
67
68#define wait_ctx_primed(thrd, user_info) /* Create the necessary information to use the signaller stack */ \
69 __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \
70 __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \
71 init_push( count, monitors, waiter, criteria ); /* Link everything together and push it to the AS-Stack */ \
72
73#define monitor_ctx( mons, cnt ) /* Define that create the necessary struct for internal/external scheduling operations */ \
74 monitor$ ** monitors = mons; /* Save the targeted monitors */ \
75 __lock_size_t count = cnt; /* Save the count to a local variable */ \
76 unsigned int recursions[ count ]; /* Save the current recursion levels to restore them later */ \
77 __waitfor_mask_t masks [ count ]; /* Save the current waitfor masks to restore them later */ \
78 __spinlock_t * locks [ count ]; /* We need to pass-in an array of locks to BlockInternal */ \
79
80#define monitor_save save ( monitors, count, locks, recursions, masks )
81#define monitor_restore restore( monitors, count, locks, recursions, masks )
82
83
84//-----------------------------------------------------------------------------
85// Enter/Leave routines
86// Enter single monitor
87static void __enter( monitor$ * this, const __monitor_group_t & group ) {
88 thread$ * thrd = active_thread();
89
90 // Lock the monitor spinlock
91 lock( this->lock __cfaabi_dbg_ctx2 );
92
93 __cfaabi_dbg_print_safe( "Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner);
94
95 if( unlikely(0 != (0x1 & (uintptr_t)this->owner)) ) {
96 abort( "Attempt by thread \"%.256s\" (%p) to access joined monitor %p.", thrd->self_cor.name, thrd, this );
97 }
98 else if( !this->owner ) {
99 // No one has the monitor, just take it
100 __set_owner( this, thrd );
101
102 __cfaabi_dbg_print_safe( "Kernel : mon is free \n" );
103 }
104 else if( this->owner == thrd) {
105 // We already have the monitor, just note how many times we took it
106 this->recursion += 1;
107
108 __cfaabi_dbg_print_safe( "Kernel : mon already owned \n" );
109 }
110 else if( is_accepted( this, group) ) {
111 // Some one was waiting for us, enter
112 __set_owner( this, thrd );
113
114 // Reset mask
115 reset_mask( this );
116
117 __cfaabi_dbg_print_safe( "Kernel : mon accepts \n" );
118 }
119 else {
120 __cfaabi_dbg_print_safe( "Kernel : blocking \n" );
121
122 // Some one else has the monitor, wait in line for it
123 /* paranoid */ verify( thrd->user_link.next == 0p );
124 append( this->entry_queue, thrd );
125 /* paranoid */ verify( thrd->user_link.next == 1p );
126
127 unlock( this->lock );
128 park();
129
130 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
131
132 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
133 return;
134 }
135
136 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
137
138 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
139 /* paranoid */ verify( this->lock.lock );
140
141 // Release the lock and leave
142 unlock( this->lock );
143 return;
144}
145
146static void __dtor_enter( monitor$ * this, fptr_t func, bool join ) {
147 thread$ * thrd = active_thread();
148 #if defined( __CFA_WITH_VERIFY__ )
149 bool is_thrd = this == &thrd->self_mon;
150 #endif
151
152 // Lock the monitor spinlock
153 lock( this->lock __cfaabi_dbg_ctx2 );
154
155 __cfaabi_dbg_print_safe( "Kernel : %10p Entering dtor for mon %p (%p)\n", thrd, this, this->owner);
156
157
158 if( !this->owner ) {
159 __cfaabi_dbg_print_safe( "Kernel : Destroying free mon %p\n", this);
160
161 // No one has the monitor, just take it
162 __set_owner( this, thrd );
163
164 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
165 /* paranoid */ verify( !is_thrd || thrd->state == Halted || thrd->state == Cancelled );
166
167 unlock( this->lock );
168 return;
169 }
170 else if( this->owner == thrd && !join) {
171 // We already have the monitor... but where about to destroy it so the nesting will fail
172 // Abort!
173 abort( "Attempt to destroy monitor %p by thread \"%.256s\" (%p) in nested mutex.", this, thrd->self_cor.name, thrd );
174 }
175 // SKULLDUGGERY: join will act as a dtor so it would normally trigger to above check
176 // because join will not release the monitor after it executed.
177 // to avoid that it sets the owner to the special value thrd | 1p before exiting
178 else if( this->owner == (thread$*)(1 | (uintptr_t)thrd) ) {
179 // restore the owner and just return
180 __cfaabi_dbg_print_safe( "Kernel : Destroying free mon %p\n", this);
181
182 // No one has the monitor, just take it
183 __set_owner( this, thrd );
184
185 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
186 /* paranoid */ verify( !is_thrd || thrd->state == Halted || thrd->state == Cancelled );
187
188 unlock( this->lock );
189 return;
190 }
191
192 // The monitor is busy, if this is a thread and the thread owns itself, it better be active
193 /* paranoid */ verify( !is_thrd || this->owner != thrd || (thrd->state != Halted && thrd->state != Cancelled) );
194
195 __lock_size_t count = 1;
196 monitor$ ** monitors = &this;
197 __monitor_group_t group = { &this, 1, func };
198 if( is_accepted( this, group) ) {
199 __cfaabi_dbg_print_safe( "Kernel : mon accepts dtor, block and signal it \n" );
200
201 // Wake the thread that is waiting for this
202 __condition_criterion_t * urgent = pop( this->signal_stack );
203 /* paranoid */ verify( urgent );
204
205 // Reset mask
206 reset_mask( this );
207
208 // Create the node specific to this wait operation
209 wait_ctx_primed( thrd, 0 )
210
211 // Some one else has the monitor, wait for him to finish and then run
212 unlock( this->lock );
213
214 // Release the next thread
215 /* paranoid */ verifyf( urgent->owner->waiting_thread == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
216 unpark( urgent->owner->waiting_thread );
217
218 // Park current thread waiting
219 park();
220
221 // Some one was waiting for us, enter
222 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
223
224 __cfaabi_dbg_print_safe( "Kernel : Destroying %p\n", this);
225 return;
226 }
227 else {
228 __cfaabi_dbg_print_safe( "Kernel : blocking \n" );
229
230 wait_ctx( thrd, 0 )
231 this->dtor_node = &waiter;
232
233 // Some one else has the monitor, wait in line for it
234 /* paranoid */ verify( thrd->user_link.next == 0p );
235 append( this->entry_queue, thrd );
236 /* paranoid */ verify( thrd->user_link.next == 1p );
237 unlock( this->lock );
238
239 // Park current thread waiting
240 park();
241
242 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
243 return;
244 }
245}
246
247// Leave single monitor
248static void __leave( monitor$ * this ) {
249 // Lock the monitor spinlock
250 lock( this->lock __cfaabi_dbg_ctx2 );
251
252 __cfaabi_dbg_print_safe( "Kernel : %10p Leaving mon %p (%p)\n", active_thread(), this, this->owner);
253
254 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
255
256 // Leaving a recursion level, decrement the counter
257 this->recursion -= 1;
258
259 // If we haven't left the last level of recursion
260 // it means we don't need to do anything
261 if( this->recursion != 0) {
262 __cfaabi_dbg_print_safe( "Kernel : recursion still %d\n", this->recursion);
263 unlock( this->lock );
264 return;
265 }
266
267 // Get the next thread, will be null on low contention monitor
268 thread$ * new_owner = next_thread( this );
269
270 // Check the new owner is consistent with who we wake-up
271 // new_owner might be null even if someone owns the monitor when the owner is still waiting for another monitor
272 /* paranoid */ verifyf( !new_owner || new_owner == this->owner, "Expected owner to be %p, got %p (m: %p)", new_owner, this->owner, this );
273
274 // We can now let other threads in safely
275 unlock( this->lock );
276
277 //We need to wake-up the thread
278 /* paranoid */ verifyf( !new_owner || new_owner == this->owner, "Expected owner to be %p, got %p (m: %p)", new_owner, this->owner, this );
279 unpark( new_owner );
280}
281
282// Leave single monitor for the last time
283static void __dtor_leave( monitor$ * this, bool join ) {
284 __cfaabi_dbg_debug_do(
285 if( active_thread() != this->owner ) {
286 abort( "Destroyed monitor %p has inconsistent owner, expected %p got %p.\n", this, active_thread(), this->owner);
287 }
288 if( this->recursion != 1 && !join ) {
289 abort( "Destroyed monitor %p has %d outstanding nested calls.\n", this, this->recursion - 1);
290 }
291 )
292
293 this->owner = (thread$*)(1 | (uintptr_t)this->owner);
294}
295
296void __thread_finish( thread$ * thrd ) {
297 monitor$ * this = &thrd->self_mon;
298
299 // Lock the monitor now
300 /* paranoid */ verify( 0x0D15EA5E0D15EA5Ep == thrd->canary );
301 /* paranoid */ verify( this->lock.lock );
302 /* paranoid */ verify( thrd->context.SP );
303 /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) > ((uintptr_t)__get_stack(thrd->curr_cor)->limit), "ERROR : thread$ %p has been corrupted.\n StackPointer too large.\n", thrd );
304 /* paranoid */ verifyf( ((uintptr_t)thrd->context.SP) < ((uintptr_t)__get_stack(thrd->curr_cor)->base ), "ERROR : thread$ %p has been corrupted.\n StackPointer too small.\n", thrd );
305 /* paranoid */ verify( ! __preemption_enabled() );
306
307 /* paranoid */ verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", thrd, this->owner, this->recursion, this );
308 /* paranoid */ verify( thrd->state == Halting );
309 /* paranoid */ verify( this->recursion == 1 );
310
311 // Leaving a recursion level, decrement the counter
312 this->recursion -= 1;
313 this->owner = 0p;
314
315 // Fetch the next thread, can be null
316 thread$ * new_owner = next_thread( this );
317
318 // Mark the state as fully halted
319 thrd->state = Halted;
320
321 // Release the monitor lock
322 unlock( this->lock );
323
324 // Unpark the next owner if needed
325 /* paranoid */ verifyf( !new_owner || new_owner == this->owner, "Expected owner to be %p, got %p (m: %p)", new_owner, this->owner, this );
326 /* paranoid */ verify( ! __preemption_enabled() );
327 /* paranoid */ verify( thrd->state == Halted );
328 unpark( new_owner );
329}
330
331// Enter multiple monitor
332// relies on the monitor array being sorted
333static inline void enter( __monitor_group_t monitors ) {
334 for( __lock_size_t i = 0; i < monitors.size; i++) {
335 __enter( monitors[i], monitors );
336 }
337}
338
339// Leave multiple monitor
340// relies on the monitor array being sorted
341static inline void leave(monitor$ * monitors [], __lock_size_t count) {
342 for( __lock_size_t i = count - 1; i >= 0; i--) {
343 __leave( monitors[i] );
344 }
345}
346
347// Ctor for monitor guard
348// Sorts monitors before entering
349void ?{}( monitor_guard_t & this, monitor$ * m [], __lock_size_t count, fptr_t func ) libcfa_public {
350 thread$ * thrd = active_thread();
351
352 // Store current array
353 this.m = m;
354 this.count = count;
355
356 // Sort monitors based on address
357 __libcfa_small_sort(this.m, count);
358
359 // Save previous thread context
360 this.prev = thrd->monitors;
361
362 // Update thread context (needed for conditions)
363 (thrd->monitors){m, count, func};
364
365 // __cfaabi_dbg_print_safe( "MGUARD : enter %d\n", count);
366
367 // Enter the monitors in order
368 __monitor_group_t group = {this.m, this.count, func};
369 enter( group );
370
371 // __cfaabi_dbg_print_safe( "MGUARD : entered\n" );
372}
373
374void ?{}( monitor_guard_t & this, monitor$ * m [], __lock_size_t count ) libcfa_public {
375 this{ m, count, 0p };
376}
377
378
379// Dtor for monitor guard
380void ^?{}( monitor_guard_t & this ) libcfa_public {
381 // __cfaabi_dbg_print_safe( "MGUARD : leaving %d\n", this.count);
382
383 // Leave the monitors in order
384 leave( this.m, this.count );
385
386 // __cfaabi_dbg_print_safe( "MGUARD : left\n" );
387
388 // Restore thread context
389 active_thread()->monitors = this.prev;
390}
391
392// Ctor for monitor guard
393// Sorts monitors before entering
394void ?{}( monitor_dtor_guard_t & this, monitor$ * m [], fptr_t func, bool join ) libcfa_public {
395 // optimization
396 thread$ * thrd = active_thread();
397
398 // Store current array
399 this.m = *m;
400
401 // Save previous thread context
402 this.prev = thrd->monitors;
403
404 // Save whether we are in a join or not
405 this.join = join;
406
407 // Update thread context (needed for conditions)
408 (thrd->monitors){m, 1, func};
409
410 __dtor_enter( this.m, func, join );
411}
412
413// Dtor for monitor guard
414void ^?{}( monitor_dtor_guard_t & this ) libcfa_public {
415 // Leave the monitors in order
416 __dtor_leave( this.m, this.join );
417
418 // Restore thread context
419 active_thread()->monitors = this.prev;
420}
421
422//-----------------------------------------------------------------------------
423// Internal scheduling types
424static void ?{}(__condition_node_t & this, thread$ * waiting_thread, __lock_size_t count, uintptr_t user_info ) {
425 this.waiting_thread = waiting_thread;
426 this.count = count;
427 this.next = 0p;
428 this.user_info = user_info;
429}
430
431static void ?{}(__condition_criterion_t & this ) with( this ) {
432 ready = false;
433 target = 0p;
434 owner = 0p;
435 next = 0p;
436}
437
438static void ?{}(__condition_criterion_t & this, monitor$ * target, __condition_node_t & owner ) {
439 this.ready = false;
440 this.target = target;
441 this.owner = &owner;
442 this.next = 0p;
443}
444
445//-----------------------------------------------------------------------------
446// Internal scheduling
447void wait( condition & this, uintptr_t user_info = 0 ) libcfa_public {
448 brand_condition( this );
449
450 // Check that everything is as expected
451 assertf( this.monitors != 0p, "Waiting with no monitors (%p)", this.monitors );
452 verifyf( this.monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", this.monitor_count );
453 verifyf( this.monitor_count < 32u, "Excessive monitor count (%"PRIiFAST16")", this.monitor_count );
454
455 // Create storage for monitor context
456 monitor_ctx( this.monitors, this.monitor_count );
457
458 // Create the node specific to this wait operation
459 wait_ctx( active_thread(), user_info );
460
461 // Append the current wait operation to the ones already queued on the condition
462 // We don't need locks for that since conditions must always be waited on inside monitor mutual exclusion
463 /* paranoid */ verify( waiter.next == 0p );
464 append( this.blocked, &waiter );
465 /* paranoid */ verify( waiter.next == 1p );
466
467 // Lock all monitors (aggregates the locks as well)
468 lock_all( monitors, locks, count );
469
470 // Find the next thread(s) to run
471 __lock_size_t thread_count = 0;
472 thread$ * threads[ count ];
473 __builtin_memset( threads, 0, sizeof( threads ) );
474
475 // Save monitor states
476 monitor_save;
477
478 // Remove any duplicate threads
479 for( __lock_size_t i = 0; i < count; i++) {
480 thread$ * new_owner = next_thread( monitors[i] );
481 insert_unique( threads, thread_count, new_owner );
482 }
483
484 // Unlock the locks, we don't need them anymore
485 for(int i = 0; i < count; i++) {
486 unlock( *locks[i] );
487 }
488
489 // Wake the threads
490 for(int i = 0; i < thread_count; i++) {
491 unpark( threads[i] );
492 }
493
494 // Everything is ready to go to sleep
495 park();
496
497 // We are back, restore the owners and recursions
498 monitor_restore;
499}
500
501bool signal( condition & this ) libcfa_public {
502 if( is_empty( this ) ) { return false; }
503
504 //Check that everything is as expected
505 verify( this.monitors );
506 verify( this.monitor_count != 0 );
507
508 //Some more checking in debug
509 __cfaabi_dbg_debug_do(
510 thread$ * this_thrd = active_thread();
511 if ( this.monitor_count != this_thrd->monitors.size ) {
512 abort( "Signal on condition %p made with different number of monitor(s), expected %zi got %zi", &this, this.monitor_count, this_thrd->monitors.size );
513 }
514
515 for(int i = 0; i < this.monitor_count; i++) {
516 if ( this.monitors[i] != this_thrd->monitors[i] ) {
517 abort( "Signal on condition %p made with different monitor, expected %p got %p", &this, this.monitors[i], this_thrd->monitors[i] );
518 }
519 }
520 );
521
522 __lock_size_t count = this.monitor_count;
523
524 // Lock all monitors
525 lock_all( this.monitors, 0p, count );
526
527 //Pop the head of the waiting queue
528 __condition_node_t * node = pop_head( this.blocked );
529
530 //Add the thread to the proper AS stack
531 for(int i = 0; i < count; i++) {
532 __condition_criterion_t * crit = &node->criteria[i];
533 assert( !crit->ready );
534 push( crit->target->signal_stack, crit );
535 }
536
537 //Release
538 unlock_all( this.monitors, count );
539
540 return true;
541}
542
543bool signal_block( condition & this ) libcfa_public {
544 if( !this.blocked.head ) { return false; }
545
546 //Check that everything is as expected
547 verifyf( this.monitors != 0p, "Waiting with no monitors (%p)", this.monitors );
548 verifyf( this.monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", this.monitor_count );
549
550 // Create storage for monitor context
551 monitor_ctx( this.monitors, this.monitor_count );
552
553 // Lock all monitors (aggregates the locks them as well)
554 lock_all( monitors, locks, count );
555
556
557 // Create the node specific to this wait operation
558 wait_ctx_primed( active_thread(), 0 )
559
560 //save contexts
561 monitor_save;
562
563 //Find the thread to run
564 thread$ * signallee = pop_head( this.blocked )->waiting_thread;
565 __set_owner( monitors, count, signallee );
566
567 __cfaabi_dbg_print_buffer_decl( "Kernel : signal_block condition %p (s: %p)\n", &this, signallee );
568
569 // unlock all the monitors
570 unlock_all( locks, count );
571
572 // unpark the thread we signalled
573 unpark( signallee );
574
575 //Everything is ready to go to sleep
576 park();
577
578
579 // WE WOKE UP
580
581
582 __cfaabi_dbg_print_buffer_local( "Kernel : signal_block returned\n" );
583
584 //We are back, restore the masks and recursions
585 monitor_restore;
586
587 return true;
588}
589
590// Access the user_info of the thread waiting at the front of the queue
591uintptr_t front( condition & this ) libcfa_public {
592 verifyf( !is_empty(this),
593 "Attempt to access user data on an empty condition.\n"
594 "Possible cause is not checking if the condition is empty before reading stored data."
595 );
596 return ((typeof(this.blocked.head))this.blocked.head)->user_info;
597}
598
599//-----------------------------------------------------------------------------
600// External scheduling
601// cases to handle :
602// - target already there :
603// block and wake
604// - dtor already there
605// put thread on signaller stack
606// - non-blocking
607// return else
608// - timeout
609// return timeout
610// - block
611// setup mask
612// block
613void __waitfor_internal( const __waitfor_mask_t & mask, int duration ) libcfa_public {
614 // This statment doesn't have a contiguous list of monitors...
615 // Create one!
616 __lock_size_t max = count_max( mask );
617 monitor$ * mon_storage[max];
618 __builtin_memset( mon_storage, 0, sizeof( mon_storage ) );
619 __lock_size_t actual_count = aggregate( mon_storage, mask );
620
621 __cfaabi_dbg_print_buffer_decl( "Kernel : waitfor %"PRIdFAST16" (s: %"PRIdFAST16", m: %"PRIdFAST16")\n", actual_count, mask.size, (__lock_size_t)max);
622
623 if(actual_count == 0) return;
624
625 __cfaabi_dbg_print_buffer_local( "Kernel : waitfor internal proceeding\n" );
626
627 // Create storage for monitor context
628 monitor_ctx( mon_storage, actual_count );
629
630 // Lock all monitors (aggregates the locks as well)
631 lock_all( monitors, locks, count );
632
633 {
634 // Check if the entry queue
635 thread$ * next; int index;
636 [next, index] = search_entry_queue( mask, monitors, count );
637
638 if( next ) {
639 *mask.accepted = index;
640 __acceptable_t& accepted = mask[index];
641 if( accepted.is_dtor ) {
642 __cfaabi_dbg_print_buffer_local( "Kernel : dtor already there\n" );
643 verifyf( accepted.size == 1, "ERROR: Accepted dtor has more than 1 mutex parameter." );
644
645 monitor$ * mon2dtor = accepted[0];
646 verifyf( mon2dtor->dtor_node, "ERROR: Accepted monitor has no dtor_node." );
647
648 __condition_criterion_t * dtor_crit = mon2dtor->dtor_node->criteria;
649 push( mon2dtor->signal_stack, dtor_crit );
650
651 unlock_all( locks, count );
652 }
653 else {
654 __cfaabi_dbg_print_buffer_local( "Kernel : thread present, baton-passing\n" );
655
656 // Create the node specific to this wait operation
657 wait_ctx_primed( active_thread(), 0 );
658
659 // Save monitor states
660 monitor_save;
661
662 __cfaabi_dbg_print_buffer_local( "Kernel : baton of %"PRIdFAST16" monitors : ", count );
663 #ifdef __CFA_DEBUG_PRINT__
664 for( int i = 0; i < count; i++) {
665 __cfaabi_dbg_print_buffer_local( "%p %p ", monitors[i], monitors[i]->signal_stack.top );
666 }
667 #endif
668 __cfaabi_dbg_print_buffer_local( "\n" );
669
670 // Set the owners to be the next thread
671 __set_owner( monitors, count, next );
672
673 // unlock all the monitors
674 unlock_all( locks, count );
675
676 // unpark the thread we signalled
677 unpark( next );
678
679 //Everything is ready to go to sleep
680 park();
681
682 // We are back, restore the owners and recursions
683 monitor_restore;
684
685 __cfaabi_dbg_print_buffer_local( "Kernel : thread present, returned\n" );
686 }
687
688 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
689 return;
690 }
691 }
692
693
694 if( duration == 0 ) {
695 __cfaabi_dbg_print_buffer_local( "Kernel : non-blocking, exiting\n" );
696
697 unlock_all( locks, count );
698
699 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
700 return;
701 }
702
703
704 verifyf( duration < 0, "Timeout on waitfor statments not supported yet." );
705
706 __cfaabi_dbg_print_buffer_local( "Kernel : blocking waitfor\n" );
707
708 // Create the node specific to this wait operation
709 wait_ctx_primed( active_thread(), 0 );
710
711 monitor_save;
712 set_mask( monitors, count, mask );
713
714 for( __lock_size_t i = 0; i < count; i++) {
715 verify( monitors[i]->owner == active_thread() );
716 }
717
718 // unlock all the monitors
719 unlock_all( locks, count );
720
721 //Everything is ready to go to sleep
722 park();
723
724
725 // WE WOKE UP
726
727
728 //We are back, restore the masks and recursions
729 monitor_restore;
730
731 __cfaabi_dbg_print_buffer_local( "Kernel : exiting\n" );
732
733 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
734}
735
736//-----------------------------------------------------------------------------
737// Utilities
738
739static inline void __set_owner( monitor$ * this, thread$ * owner ) {
740 /* paranoid */ verify( this->lock.lock );
741
742 //Pass the monitor appropriately
743 this->owner = owner;
744
745 //We are passing the monitor to someone else, which means recursion level is not 0
746 this->recursion = owner ? 1 : 0;
747}
748
749static inline void __set_owner( monitor$ * monitors [], __lock_size_t count, thread$ * owner ) {
750 /* paranoid */ verify ( monitors[0]->lock.lock );
751 /* paranoid */ verifyf( monitors[0]->owner == active_thread(), "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), monitors[0]->owner, monitors[0]->recursion, monitors[0] );
752 monitors[0]->owner = owner;
753 monitors[0]->recursion = 1;
754 for( __lock_size_t i = 1; i < count; i++ ) {
755 /* paranoid */ verify ( monitors[i]->lock.lock );
756 /* paranoid */ verifyf( monitors[i]->owner == active_thread(), "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), monitors[i]->owner, monitors[i]->recursion, monitors[i] );
757 monitors[i]->owner = owner;
758 monitors[i]->recursion = 0;
759 }
760}
761
762static inline void set_mask( monitor$ * storage [], __lock_size_t count, const __waitfor_mask_t & mask ) {
763 for( __lock_size_t i = 0; i < count; i++) {
764 storage[i]->mask = mask;
765 }
766}
767
768static inline void reset_mask( monitor$ * this ) {
769 this->mask.accepted = 0p;
770 this->mask.data = 0p;
771 this->mask.size = 0;
772}
773
774static inline thread$ * next_thread( monitor$ * this ) {
775 //Check the signaller stack
776 __cfaabi_dbg_print_safe( "Kernel : mon %p AS-stack top %p\n", this, this->signal_stack.top);
777 __condition_criterion_t * urgent = pop( this->signal_stack );
778 if( urgent ) {
779 //The signaller stack is not empty,
780 //regardless of if we are ready to baton pass,
781 //we need to set the monitor as in use
782 /* paranoid */ verifyf( !this->owner || active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
783 __set_owner( this, urgent->owner->waiting_thread );
784
785 return check_condition( urgent );
786 }
787
788 // No signaller thread
789 // Get the next thread in the entry_queue
790 thread$ * new_owner = pop_head( this->entry_queue );
791 /* paranoid */ verifyf( !this->owner || active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
792 /* paranoid */ verify( !new_owner || new_owner->user_link.next == 0p );
793 __set_owner( this, new_owner );
794
795 return new_owner;
796}
797
798static inline bool is_accepted( monitor$ * this, const __monitor_group_t & group ) {
799 __acceptable_t * it = this->mask.data; // Optim
800 __lock_size_t count = this->mask.size;
801
802 // Check if there are any acceptable functions
803 if( !it ) return false;
804
805 // If this isn't the first monitor to test this, there is no reason to repeat the test.
806 if( this != group[0] ) return group[0]->mask.accepted >= 0;
807
808 // For all acceptable functions check if this is the current function.
809 for( __lock_size_t i = 0; i < count; i++, it++ ) {
810 if( *it == group ) {
811 *this->mask.accepted = i;
812 return true;
813 }
814 }
815
816 // No function matched
817 return false;
818}
819
820static inline void init( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
821 for( __lock_size_t i = 0; i < count; i++) {
822 (criteria[i]){ monitors[i], waiter };
823 }
824
825 waiter.criteria = criteria;
826}
827
828static inline void init_push( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
829 for( __lock_size_t i = 0; i < count; i++) {
830 (criteria[i]){ monitors[i], waiter };
831 __cfaabi_dbg_print_safe( "Kernel : target %p = %p\n", criteria[i].target, &criteria[i] );
832 push( criteria[i].target->signal_stack, &criteria[i] );
833 }
834
835 waiter.criteria = criteria;
836}
837
838static inline void lock_all( __spinlock_t * locks [], __lock_size_t count ) {
839 for( __lock_size_t i = 0; i < count; i++ ) {
840 lock( *locks[i] __cfaabi_dbg_ctx2 );
841 }
842}
843
844static inline void lock_all( monitor$ * source [], __spinlock_t * /*out*/ locks [], __lock_size_t count ) {
845 for( __lock_size_t i = 0; i < count; i++ ) {
846 __spinlock_t * l = &source[i]->lock;
847 lock( *l __cfaabi_dbg_ctx2 );
848 if(locks) locks[i] = l;
849 }
850}
851
852static inline void unlock_all( __spinlock_t * locks [], __lock_size_t count ) {
853 for( __lock_size_t i = 0; i < count; i++ ) {
854 unlock( *locks[i] );
855 }
856}
857
858static inline void unlock_all( monitor$ * locks [], __lock_size_t count ) {
859 for( __lock_size_t i = 0; i < count; i++ ) {
860 unlock( locks[i]->lock );
861 }
862}
863
864static inline void save(
865 monitor$ * ctx [],
866 __lock_size_t count,
867 __attribute((unused)) __spinlock_t * locks [],
868 unsigned int /*out*/ recursions [],
869 __waitfor_mask_t /*out*/ masks []
870) {
871 for( __lock_size_t i = 0; i < count; i++ ) {
872 recursions[i] = ctx[i]->recursion;
873 masks[i] = ctx[i]->mask;
874 }
875}
876
877static inline void restore(
878 monitor$ * ctx [],
879 __lock_size_t count,
880 __spinlock_t * locks [],
881 unsigned int /*out*/ recursions [],
882 __waitfor_mask_t /*out*/ masks []
883) {
884 lock_all( locks, count );
885 for( __lock_size_t i = 0; i < count; i++ ) {
886 ctx[i]->recursion = recursions[i];
887 ctx[i]->mask = masks[i];
888 }
889 unlock_all( locks, count );
890}
891
892// Function has 2 different behavior
893// 1 - Marks a monitors as being ready to run
894// 2 - Checks if all the monitors are ready to run
895// if so return the thread to run
896static inline thread$ * check_condition( __condition_criterion_t * target ) {
897 __condition_node_t * node = target->owner;
898 unsigned short count = node->count;
899 __condition_criterion_t * criteria = node->criteria;
900
901 bool ready2run = true;
902
903 for( int i = 0; i < count; i++ ) {
904
905 // __cfaabi_dbg_print_safe( "Checking %p for %p\n", &criteria[i], target );
906 if( &criteria[i] == target ) {
907 criteria[i].ready = true;
908 // __cfaabi_dbg_print_safe( "True\n" );
909 }
910
911 ready2run = criteria[i].ready && ready2run;
912 }
913
914 __cfaabi_dbg_print_safe( "Kernel : Runing %i (%p)\n", ready2run, ready2run ? (thread*)node->waiting_thread : (thread*)0p );
915 return ready2run ? node->waiting_thread : 0p;
916}
917
918static inline void brand_condition( condition & this ) {
919 thread$ * thrd = active_thread();
920 if( !this.monitors ) {
921 // __cfaabi_dbg_print_safe( "Branding\n" );
922 assertf( thrd->monitors.data != 0p, "No current monitor to brand condition %p", thrd->monitors.data );
923 this.monitor_count = thrd->monitors.size;
924
925 this.monitors = (monitor$ **)malloc( this.monitor_count * sizeof( *this.monitors ) );
926 for( int i = 0; i < this.monitor_count; i++ ) {
927 this.monitors[i] = thrd->monitors[i];
928 }
929 }
930}
931
932static inline [thread$ *, int] search_entry_queue( const __waitfor_mask_t & mask, monitor$ * monitors [], __lock_size_t count ) {
933 __queue_t(thread$) & entry_queue = monitors[0]->entry_queue;
934
935#if 0
936 #if defined( __CFA_WITH_VERIFY__ )
937 thread$ * last = 0p;
938 #endif
939 // For each thread in the entry-queue
940 for ( thread$ ** thrd_it = &entry_queue.head; (*thrd_it) != 1p; thrd_it = &get_next(**thrd_it) ) {
941 thread$ * curr = *thrd_it;
942
943 /* paranoid */ verifyf( !last || last->user_link.next == curr, "search not making progress, from %p (%p) to %p",
944 last, last->user_link.next, curr );
945 /* paranoid */ verifyf( curr != last, "search not making progress, from %p to %p", last, curr );
946
947 // For each acceptable check if it matches
948 int i = 0;
949 __acceptable_t * end = end (mask);
950 __acceptable_t * begin = begin(mask);
951 for ( __acceptable_t * it = begin; it != end; it++, i++ ) {
952 // Check for match
953 if ( *it == curr->monitors ) {
954 // If match, return it after removeing it from the entry queue
955 return [remove( entry_queue, thrd_it ), i];
956 }
957 }
958
959 #if defined( __CFA_WITH_VERIFY__ )
960 last = curr;
961 #endif
962 }
963#endif
964 int i = 0;
965 __acceptable_t * end = end (mask);
966 __acceptable_t * begin = begin(mask);
967 // For each acceptable (respect lexical priority in waitfor statement)
968 for ( __acceptable_t * it = begin; it != end; it++, i++ ) {
969 #if defined( __CFA_WITH_VERIFY__ )
970 thread$ * last = 0p;
971 #endif // __CFA_WITH_VERIFY__
972
973 for ( thread$ ** thrd_it = &entry_queue.head; (*thrd_it) != 1p; thrd_it = &get_next(**thrd_it) ) {
974 thread$ * curr = *thrd_it;
975
976 /* paranoid */ verifyf( !last || last->user_link.next == curr, "search not making progress, from %p (%p) to %p",
977 last, last->user_link.next, curr );
978 /* paranoid */ verifyf( curr != last, "search not making progress, from %p to %p", last, curr );
979
980 // For each thread in the entry-queue check for a match
981 if ( *it == curr->monitors ) {
982 // If match, return it after removeing from the entry queue
983 return [remove( entry_queue, thrd_it ), i];
984 } // if
985
986 #if defined( __CFA_WITH_VERIFY__ )
987 last = curr;
988 #endif
989 } // for
990 } // for
991 return [0, -1];
992}
993
994forall(T & | sized( T ))
995static inline __lock_size_t insert_unique( T * array [], __lock_size_t & size, T * val ) {
996 if( !val ) return size;
997
998 for( __lock_size_t i = 0; i <= size; i++) {
999 if( array[i] == val ) return size;
1000 }
1001
1002 array[size] = val;
1003 size = size + 1;
1004 return size;
1005}
1006
1007static inline __lock_size_t count_max( const __waitfor_mask_t & mask ) {
1008 __lock_size_t max = 0;
1009 for( __lock_size_t i = 0; i < mask.size; i++ ) {
1010 __acceptable_t & accepted = mask[i];
1011 max += accepted.size;
1012 }
1013 return max;
1014}
1015
1016static inline __lock_size_t aggregate( monitor$ * storage [], const __waitfor_mask_t & mask ) {
1017 __lock_size_t size = 0;
1018 for( __lock_size_t i = 0; i < mask.size; i++ ) {
1019 __acceptable_t & accepted = mask[i];
1020 __libcfa_small_sort( accepted.data, accepted.size );
1021 for( __lock_size_t j = 0; j < accepted.size; j++) {
1022 insert_unique( storage, size, accepted[j] );
1023 }
1024 }
1025 // TODO insertion sort instead of this
1026 __libcfa_small_sort( storage, size );
1027 return size;
1028}
1029
1030//-----------------------------------------------------------------------------
1031// Enter routine for mutex stmt
1032// Can't be accepted since a mutex stmt is effectively an anonymous routine
1033// Thus we do not need a monitor group
1034void lock( monitor$ * this ) libcfa_public {
1035 thread$ * thrd = active_thread();
1036
1037 // Lock the monitor spinlock
1038 lock( this->lock __cfaabi_dbg_ctx2 );
1039
1040 __cfaabi_dbg_print_safe( "Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner);
1041
1042 if( unlikely(0 != (0x1 & (uintptr_t)this->owner)) ) {
1043 abort( "Attempt by thread \"%.256s\" (%p) to access joined monitor %p.", thrd->self_cor.name, thrd, this );
1044 }
1045 else if( !this->owner ) {
1046 // No one has the monitor, just take it
1047 __set_owner( this, thrd );
1048
1049 __cfaabi_dbg_print_safe( "Kernel : mon is free \n" );
1050 }
1051 else if( this->owner == thrd) {
1052 // We already have the monitor, just note how many times we took it
1053 this->recursion += 1;
1054
1055 __cfaabi_dbg_print_safe( "Kernel : mon already owned \n" );
1056 }
1057 else {
1058 __cfaabi_dbg_print_safe( "Kernel : blocking \n" );
1059
1060 // Some one else has the monitor, wait in line for it
1061 /* paranoid */ verify( thrd->user_link.next == 0p );
1062 append( this->entry_queue, thrd );
1063 /* paranoid */ verify( thrd->user_link.next == 1p );
1064
1065 unlock( this->lock );
1066 park();
1067
1068 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
1069
1070 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
1071 return;
1072 }
1073
1074 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
1075
1076 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
1077 /* paranoid */ verify( this->lock.lock );
1078
1079 // Release the lock and leave
1080 unlock( this->lock );
1081 return;
1082}
1083
1084// Leave routine for mutex stmt
1085// Is just a wrapper around __leave for the is_lock trait to see
1086void unlock( monitor$ * this ) libcfa_public { __leave( this ); }
1087
1088// Local Variables: //
1089// mode: c //
1090// tab-width: 4 //
1091// End: //
Note: See TracBrowser for help on using the repository browser.