source: libcfa/src/concurrency/monitor.cfa@ b28ce93

Last change on this file since b28ce93 was 6b33e89, checked in by Peter A. Buhr <pabuhr@…>, 5 months ago

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