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

Last change on this file since e426c6f was e426c6f, checked in by Peter A. Buhr <pabuhr@…>, 6 days ago

fix signal_block on empty condition queue
  • Property mode set to 100644
File size: 35.1 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 Mar 26 22:37:42 2026
13// Update Count : 91
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 with( this ) {
496 if ( empty( this ) ) { return false; }
497
498 verifyf( monitors != 0p, "Waiting with no monitors (%p)", monitors );
499 verifyf( monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", monitor_count );
500
501 __cfaabi_dbg_debug_do( // more checking in debug
502 thread$ * this_thrd = active_thread();
503 if ( monitor_count != this_thrd->monitors.size ) {
504 abort( "Signal on condition %p made with different number of monitor(s), expected %zi got %zi",
505 &this, monitor_count, this_thrd->monitors.size );
506 } // if
507
508 for ( i; monitor_count ) {
509 if ( monitors[i] != this_thrd->monitors[i] ) {
510 abort( "Signal on condition %p made with different monitor, expected %p got %p",
511 &this, monitors[i], this_thrd->monitors[i] );
512 } // if
513 } // for
514 );
515
516 __lock_size_t count = monitor_count;
517 lock_all( monitors, 0p, count ); // lock all monitors
518 __condition_node_t * node = pop_head( blocked ); // pop head of waiting queue
519
520 for ( i; count ) { // add threads to the proper AS stack
521 __condition_criterion_t * crit = &node->criteria[i];
522 assert( ! crit->ready );
523 push( crit->target->signal_stack, crit );
524 } // for
525
526 unlock_all( monitors, count ); // release
527 return true;
528} // signal
529
530bool signal_block( condition & this ) libcfa_public {
531 if ( empty( this ) ) { return false; }
532
533 verifyf( this.monitors != 0p, "Waiting with no monitors (%p)", this.monitors );
534 verifyf( this.monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", this.monitor_count );
535
536 // Create storage for monitor context
537 monitor_ctx( this.monitors, this.monitor_count ); // creates monitors, count, recursions, masks, locks
538
539 // Lock all monitors (aggregates the locks them as well)
540 lock_all( monitors, locks, count );
541
542 // Create the node specific to this wait operation
543 wait_ctx_primed( active_thread(), 0 )
544
545 monitor_save; // save contexts
546
547 //Find the thread to run
548 thread$ * signallee = pop_head( this.blocked )->waiting_thread;
549 __set_owner( monitors, count, signallee );
550
551 __cfaabi_dbg_print_buffer_decl( "Kernel : signal_block condition %p (s: %p)\n", &this, signallee );
552
553 unlock_all( locks, count ); // unlock all the monitors
554 unpark( signallee ); // unpark the thread we signalled
555 park(); // everything is ready to go to sleep
556
557 // WOKE UP
558
559 __cfaabi_dbg_print_buffer_local( "Kernel : signal_block returned\n" );
560
561 monitor_restore; // restore the masks and recursions
562
563 return true;
564} // signal_block
565
566// Access the user_info of the thread waiting at the front of the queue
567uintptr_t front( condition & this ) libcfa_public {
568 verifyf( ! empty( this ),
569 "Attempt to access user data on an empty condition.\n"
570 "Possible cause is not checking if the condition is empty before reading stored data."
571 );
572 return ((typeof(this.blocked.head))this.blocked.head)->user_info;
573}
574
575//-----------------------------------------------------------------------------
576// External scheduling
577// cases to handle :
578// - target already there :
579// block and wake
580// - dtor already there
581// put thread on signaller stack
582// - non-blocking
583// return else
584// - timeout
585// return timeout
586// - block
587// setup mask
588// block
589void __waitfor_internal( const __waitfor_mask_t & mask, int duration ) libcfa_public {
590 // This statment doesn't have a contiguous list of monitors...
591 // Create one!
592 __lock_size_t max = count_max( mask );
593 monitor$ * mon_storage[max];
594 __builtin_memset( mon_storage, 0, sizeof( mon_storage ) );
595 __lock_size_t actual_count = aggregate( mon_storage, mask );
596
597 __cfaabi_dbg_print_buffer_decl( "Kernel : waitfor %"PRIdFAST16" (s: %"PRIdFAST16", m: %"PRIdFAST16")\n", actual_count, mask.size, (__lock_size_t)max);
598
599 if (actual_count == 0) return;
600
601 __cfaabi_dbg_print_buffer_local( "Kernel : waitfor internal proceeding\n" );
602
603 // Create storage for monitor context
604 monitor_ctx( mon_storage, actual_count ); // creates monitors, count, recursions, masks, locks
605
606 // Lock all monitors (aggregates the locks as well)
607 lock_all( monitors, locks, count );
608
609 {
610 // Check if the entry queue
611 thread$ * nxt; int index;
612 [nxt, index] = search_entry_queue( mask, monitors, count );
613
614 if ( nxt ) {
615 *mask.accepted = index;
616 __acceptable_t& accepted = mask[index];
617 if ( accepted.is_dtor ) {
618 __cfaabi_dbg_print_buffer_local( "Kernel : dtor already there\n" );
619 verifyf( accepted.size == 1, "ERROR: Accepted dtor has more than 1 mutex parameter." );
620
621 monitor$ * mon2dtor = accepted[0];
622 verifyf( mon2dtor->dtor_node, "ERROR: Accepted monitor has no dtor_node." );
623
624 __condition_criterion_t * dtor_crit = mon2dtor->dtor_node->criteria;
625 push( mon2dtor->signal_stack, dtor_crit );
626
627 unlock_all( locks, count );
628 }
629 else {
630 __cfaabi_dbg_print_buffer_local( "Kernel : thread present, baton-passing\n" );
631
632 // Create the node specific to this wait operation
633 wait_ctx_primed( active_thread(), 0 );
634
635 // Save monitor states
636 monitor_save;
637
638 __cfaabi_dbg_print_buffer_local( "Kernel : baton of %"PRIdFAST16" monitors : ", count );
639 #ifdef __CFA_DEBUG_PRINT__
640 for ( i; count ) {
641 __cfaabi_dbg_print_buffer_local( "%p %p ", monitors[i], monitors[i]->signal_stack.top );
642 }
643 #endif
644 __cfaabi_dbg_print_buffer_local( "\n" );
645
646 // Set the owners to be the next thread
647 __set_owner( monitors, count, nxt );
648
649 // unlock all the monitors
650 unlock_all( locks, count );
651
652 // unpark the thread we signalled
653 unpark( nxt );
654
655 //Everything is ready to go to sleep
656 park();
657
658 // We are back, restore the owners and recursions
659 monitor_restore;
660
661 __cfaabi_dbg_print_buffer_local( "Kernel : thread present, returned\n" );
662 }
663
664 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
665 return;
666 }
667 }
668
669
670 if ( duration == 0 ) {
671 __cfaabi_dbg_print_buffer_local( "Kernel : non-blocking, exiting\n" );
672
673 unlock_all( locks, count );
674
675 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
676 return;
677 }
678
679
680 verifyf( duration < 0, "Timeout on waitfor statments not supported yet." );
681
682 __cfaabi_dbg_print_buffer_local( "Kernel : blocking waitfor\n" );
683
684 // Create the node specific to this wait operation
685 wait_ctx_primed( active_thread(), 0 );
686
687 monitor_save;
688 set_mask( monitors, count, mask );
689
690 for ( i; count ) {
691 verify( monitors[i]->owner == active_thread() );
692 }
693
694 // unlock all the monitors
695 unlock_all( locks, count );
696
697 //Everything is ready to go to sleep
698 park();
699
700
701 // WE WOKE UP
702
703
704 //We are back, restore the masks and recursions
705 monitor_restore;
706
707 __cfaabi_dbg_print_buffer_local( "Kernel : exiting\n" );
708
709 __cfaabi_dbg_print_buffer_local( "Kernel : accepted %d\n", *mask.accepted);
710}
711
712//-----------------------------------------------------------------------------
713// Utilities
714
715static inline void __set_owner( monitor$ * this, thread$ * owner ) {
716 /* paranoid */ verify( this->lock.lock );
717
718 //Pass the monitor appropriately
719 this->owner = owner;
720
721 //We are passing the monitor to someone else, which means recursion level is not 0
722 this->recursion = owner ? 1 : 0;
723}
724
725static inline void __set_owner( monitor$ * monitors [], __lock_size_t count, thread$ * owner ) {
726 /* paranoid */ verify ( monitors[0]->lock.lock );
727 /* 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] );
728 monitors[0]->owner = owner;
729 monitors[0]->recursion = 1;
730 for ( i; 1~count ) {
731 /* paranoid */ verify ( monitors[i]->lock.lock );
732 /* 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] );
733 monitors[i]->owner = owner;
734 monitors[i]->recursion = 0;
735 }
736}
737
738static inline void set_mask( monitor$ * storage [], __lock_size_t count, const __waitfor_mask_t & mask ) {
739 for ( i; count) {
740 storage[i]->mask = mask;
741 }
742}
743
744static inline void reset_mask( monitor$ * this ) {
745 this->mask.accepted = 0p;
746 this->mask.data = 0p;
747 this->mask.size = 0;
748}
749
750static inline thread$ * next_thread( monitor$ * this ) {
751 //Check the signaller stack
752 __cfaabi_dbg_print_safe( "Kernel : mon %p AS-stack top %p\n", this, this->signal_stack.top);
753 __condition_criterion_t * urgent = pop( this->signal_stack );
754 if ( urgent ) {
755 //The signaller stack is not empty,
756 //regardless of if we are ready to baton pass,
757 //we need to set the monitor as in use
758 /* 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 );
759 __set_owner( this, urgent->owner->waiting_thread );
760
761 return check_condition( urgent );
762 }
763
764 // No signaller thread
765 // Get the next thread in the entry_queue
766 thread$ * new_owner = pop_head( this->entry_queue );
767 /* 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 );
768 /* paranoid */ verify( ! new_owner || new_owner->user_link.next == 0p );
769 __set_owner( this, new_owner );
770
771 return new_owner;
772}
773
774static inline bool is_accepted( monitor$ * this, const __monitor_group_t & group ) {
775 __acceptable_t * it = this->mask.data; // Optim
776 __lock_size_t count = this->mask.size;
777
778 // Check if there are any acceptable functions
779 if ( ! it ) return false;
780
781 // If this isn't the first monitor to test this, there is no reason to repeat the test.
782 if ( this != group[0] ) return group[0]->mask.accepted >= 0;
783
784 // For all acceptable functions check if this is the current function.
785 for ( __lock_size_t i = 0; i < count; i++, it++ ) {
786 if ( *it == group ) {
787 *this->mask.accepted = i;
788 return true;
789 }
790 }
791
792 // No function matched
793 return false;
794}
795
796static inline void init( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
797 for ( i; count ) {
798 (criteria[i]){ monitors[i], waiter };
799 }
800
801 waiter.criteria = criteria;
802}
803
804static inline void init_push( __lock_size_t count, monitor$ * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
805 for ( i; count ) {
806 (criteria[i]){ monitors[i], waiter };
807 __cfaabi_dbg_print_safe( "Kernel : target %p = %p\n", criteria[i].target, &criteria[i] );
808 push( criteria[i].target->signal_stack, &criteria[i] );
809 }
810
811 waiter.criteria = criteria;
812}
813
814static inline void lock_all( __spinlock_t * locks [], __lock_size_t count ) {
815 for ( i; count ) {
816 lock( *locks[i] __cfaabi_dbg_ctx2 );
817 }
818}
819
820static inline void lock_all( monitor$ * source [], __spinlock_t * /*out*/ locks [], __lock_size_t count ) {
821 for ( i; count ) {
822 __spinlock_t * l = &source[i]->lock;
823 lock( *l __cfaabi_dbg_ctx2 );
824 if (locks) locks[i] = l;
825 }
826}
827
828static inline void unlock_all( __spinlock_t * locks [], __lock_size_t count ) {
829 for ( i; count ) {
830 unlock( *locks[i] );
831 }
832}
833
834static inline void unlock_all( monitor$ * locks [], __lock_size_t count ) {
835 for ( i; count ) {
836 unlock( locks[i]->lock );
837 }
838}
839
840static inline void save(
841 monitor$ * ctx [],
842 __lock_size_t count,
843 __attribute((unused)) __spinlock_t * locks [],
844 unsigned int /*out*/ recursions [],
845 __waitfor_mask_t /*out*/ masks []
846) {
847 for ( i; count ) {
848 recursions[i] = ctx[i]->recursion;
849 masks[i] = ctx[i]->mask;
850 }
851}
852
853static inline void restore(
854 monitor$ * ctx [],
855 __lock_size_t count,
856 __spinlock_t * locks [],
857 unsigned int /*out*/ recursions [],
858 __waitfor_mask_t /*out*/ masks []
859) {
860 lock_all( locks, count );
861 for ( i; count ) {
862 ctx[i]->recursion = recursions[i];
863 ctx[i]->mask = masks[i];
864 }
865 unlock_all( locks, count );
866}
867
868// Function has 2 different behavior
869// 1 - Marks a monitors as being ready to run
870// 2 - Checks if all the monitors are ready to run
871// if so return the thread to run
872static inline thread$ * check_condition( __condition_criterion_t * target ) {
873 __condition_node_t * node = target->owner;
874 unsigned short count = node->count;
875 __condition_criterion_t * criteria = node->criteria;
876
877 bool ready2run = true;
878
879 for ( i; count ) {
880 // __cfaabi_dbg_print_safe( "Checking %p for %p\n", &criteria[i], target );
881 if ( &criteria[i] == target ) {
882 criteria[i].ready = true;
883 // __cfaabi_dbg_print_safe( "True\n" );
884 }
885
886 ready2run = criteria[i].ready && ready2run;
887 }
888
889 __cfaabi_dbg_print_safe( "Kernel : Runing %i (%p)\n", ready2run, ready2run ? (thread*)node->waiting_thread : (thread*)0p );
890 return ready2run ? node->waiting_thread : 0p;
891}
892
893static inline void brand_condition( condition & this ) {
894 thread$ * thrd = active_thread();
895 if ( ! this.monitors ) {
896 // __cfaabi_dbg_print_safe( "Branding\n" );
897 assertf( thrd->monitors.data != 0p, "No current monitor to brand condition %p", thrd->monitors.data );
898 this.monitor_count = thrd->monitors.size;
899
900 this.monitors = (monitor$ **)malloc( this.monitor_count * sizeof( *this.monitors ) );
901 for ( i; this.monitor_count ) {
902 this.monitors[i] = thrd->monitors[i];
903 }
904 }
905}
906
907static inline [thread$ *, int] search_entry_queue( const __waitfor_mask_t & mask, monitor$ * monitors [], __lock_size_t count ) {
908 __queue_t(thread$) & entry_queue = monitors[0]->entry_queue;
909 // For each acceptable (respect lexical priority in waitfor statement)
910 int i = 0;
911 __acceptable_t * end = end(mask);
912 __acceptable_t * begin = begin(mask);
913 for ( __acceptable_t * it = begin; it != end; it++, i++ ) {
914 #if defined( __CFA_WITH_VERIFY__ )
915 thread$ * prior = 0p;
916 #endif // __CFA_WITH_VERIFY__
917
918 for ( thread$ ** thrd_it = &entry_queue.head; (*thrd_it) != 1p; thrd_it = &get_next(**thrd_it) ) {
919 thread$ * curr = *thrd_it;
920
921 /* paranoid */ verifyf( ! prior || prior->user_link.next == curr, "search not making progress, from %p (%p) to %p",
922 prior, prior->user_link.next, curr );
923 /* paranoid */ verifyf( curr != prior, "search not making progress, from %p to %p", prior, curr );
924
925 // For each thread in the entry-queue check for a match
926 if ( *it == curr->monitors ) {
927 // If match, return it after removing from the entry queue
928 return [remove( entry_queue, thrd_it ), i];
929 } // if
930
931 #if defined( __CFA_WITH_VERIFY__ )
932 prior = curr;
933 #endif
934 } // for
935 } // for
936 return [0, -1];
937}
938
939forall( T & | sized( T ) )
940static inline __lock_size_t insert_unique( T * array [], __lock_size_t & size, T * val ) {
941 if ( ! val ) return size;
942
943 for ( __lock_size_t i; ~= size ) {
944 if ( array[i] == val ) return size;
945 }
946
947 array[size] = val;
948 return size += 1;
949}
950
951static inline __lock_size_t count_max( const __waitfor_mask_t & mask ) {
952 __lock_size_t max = 0;
953 for ( i; mask.size ) {
954 __acceptable_t & accepted = mask[i];
955 max += accepted.size;
956 }
957 return max;
958}
959
960static inline __lock_size_t aggregate( monitor$ * storage [], const __waitfor_mask_t & mask ) {
961 __lock_size_t size = 0;
962 for ( i; mask.size ) {
963 __acceptable_t & accepted = mask[i];
964 __libcfa_small_sort( accepted.data, accepted.size );
965 for ( __lock_size_t j = 0; j < accepted.size; j++) {
966 insert_unique( storage, size, accepted[j] );
967 }
968 }
969 // TODO insertion sort instead of this
970 __libcfa_small_sort( storage, size );
971 return size;
972}
973
974//-----------------------------------------------------------------------------
975// Enter routine for mutex stmt
976// Can't be accepted since a mutex stmt is effectively an anonymous routine
977// Thus we do not need a monitor group
978void lock( monitor$ * this ) libcfa_public {
979 thread$ * thrd = active_thread();
980
981 // Lock the monitor spinlock
982 lock( this->lock __cfaabi_dbg_ctx2 );
983
984 __cfaabi_dbg_print_safe( "Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner);
985
986 if ( unlikely(0 != (0x1 & (uintptr_t)this->owner)) ) {
987 abort( "Attempt by thread \"%.256s\" (%p) to access joined monitor %p.", thrd->self_cor.name, thrd, this );
988 } else if ( ! this->owner ) {
989 // No one has the monitor, just take it
990 __set_owner( this, thrd );
991
992 __cfaabi_dbg_print_safe( "Kernel : mon is free \n" );
993 } else if ( this->owner == thrd) {
994 // We already have the monitor, just note how many times we took it
995 this->recursion += 1;
996
997 __cfaabi_dbg_print_safe( "Kernel : mon already owned \n" );
998 } else {
999 __cfaabi_dbg_print_safe( "Kernel : blocking \n" );
1000
1001 // Some one else has the monitor, wait in line for it
1002 /* paranoid */ verify( thrd->user_link.next == 0p );
1003 append( this->entry_queue, thrd );
1004 /* paranoid */ verify( thrd->user_link.next == 1p );
1005
1006 unlock( this->lock );
1007 park();
1008
1009 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
1010
1011 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
1012 return;
1013 }
1014
1015 __cfaabi_dbg_print_safe( "Kernel : %10p Entered mon %p\n", thrd, this);
1016
1017 /* paranoid */ verifyf( active_thread() == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", active_thread(), this->owner, this->recursion, this );
1018 /* paranoid */ verify( this->lock.lock );
1019
1020 // Release the lock and leave
1021 unlock( this->lock );
1022 return;
1023}
1024
1025// Leave routine for mutex stmt
1026// Is just a wrapper around __leave for the is_lock trait to see
1027void unlock( monitor$ * this ) libcfa_public { __leave( this ); }
1028
1029// Local Variables: //
1030// mode: c //
1031// tab-width: 4 //
1032// End: //
Note: See TracBrowser for help on using the repository browser.