source: src/libcfa/concurrency/monitor.c @ f5478c8

aaron-thesisarm-ehcleanup-dtorsdeferred_resndemanglerjacob/cs343-translationjenkins-sandboxnew-astnew-ast-unique-exprnew-envno_listpersistent-indexerresolv-newwith_gc
Last change on this file since f5478c8 was f5478c8, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

  • Property mode set to 100644
File size: 28.5 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_desc.c --
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 : Mon Jul 31 14:59:05 2017
13// Update Count     : 3
14//
15
16#include "monitor"
17
18#include <stdlib>
19#include <inttypes.h>
20
21#include "libhdr.h"
22#include "kernel_private.h"
23
24#include "bits/algorithms.h"
25
26//-----------------------------------------------------------------------------
27// Forward declarations
28static inline void set_owner ( monitor_desc * this, thread_desc * owner );
29static inline void set_owner ( monitor_desc * storage [], __lock_size_t count, thread_desc * owner );
30static inline void set_mask  ( monitor_desc * storage [], __lock_size_t count, const __waitfor_mask_t & mask );
31static inline void reset_mask( monitor_desc * this );
32
33static inline thread_desc * next_thread( monitor_desc * this );
34static inline bool is_accepted( monitor_desc * this, const __monitor_group_t & monitors );
35
36static inline void lock_all  ( __spinlock_t * locks [], __lock_size_t count );
37static inline void lock_all  ( monitor_desc * source [], __spinlock_t * /*out*/ locks [], __lock_size_t count );
38static inline void unlock_all( __spinlock_t * locks [], __lock_size_t count );
39static inline void unlock_all( monitor_desc * locks [], __lock_size_t count );
40
41static inline void save   ( monitor_desc * ctx [], __lock_size_t count, __spinlock_t * locks [], unsigned int /*out*/ recursions [], __waitfor_mask_t /*out*/ masks [] );
42static inline void restore( monitor_desc * ctx [], __lock_size_t count, __spinlock_t * locks [], unsigned int /*in */ recursions [], __waitfor_mask_t /*in */ masks [] );
43
44static inline void init     ( __lock_size_t count, monitor_desc * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] );
45static inline void init_push( __lock_size_t count, monitor_desc * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] );
46
47static inline thread_desc *        check_condition   ( __condition_criterion_t * );
48static inline void                 brand_condition   ( condition & );
49static inline [thread_desc *, int] search_entry_queue( const __waitfor_mask_t &, monitor_desc * monitors [], __lock_size_t count );
50
51forall(dtype T | sized( T ))
52static inline __lock_size_t insert_unique( T * array [], __lock_size_t & size, T * val );
53static inline __lock_size_t count_max    ( const __waitfor_mask_t & mask );
54static inline __lock_size_t aggregate    ( monitor_desc * storage [], const __waitfor_mask_t & mask );
55
56#ifndef __CFA_LOCK_NO_YIELD
57#define DO_LOCK lock_yield
58#else
59#define DO_LOCK lock
60#endif
61
62//-----------------------------------------------------------------------------
63// Useful defines
64#define wait_ctx(thrd, user_info)                               /* Create the necessary information to use the signaller stack                         */ \
65        __condition_node_t waiter = { thrd, count, user_info };   /* Create the node specific to this wait operation                                     */ \
66        __condition_criterion_t criteria[count];                  /* Create the creteria this wait operation needs to wake up                            */ \
67        init( count, monitors, waiter, criteria );                /* Link everything together                                                            */ \
68
69#define wait_ctx_primed(thrd, user_info)                        /* Create the necessary information to use the signaller stack                         */ \
70        __condition_node_t waiter = { thrd, count, user_info };   /* Create the node specific to this wait operation                                     */ \
71        __condition_criterion_t criteria[count];                  /* Create the creteria this wait operation needs to wake up                            */ \
72        init_push( count, monitors, waiter, criteria );           /* Link everything together and push it to the AS-Stack                                */ \
73
74#define monitor_ctx( mons, cnt )                                /* Define that create the necessary struct for internal/external scheduling operations */ \
75        monitor_desc ** monitors = mons;                          /* Save the targeted monitors                                                          */ \
76        __lock_size_t count = cnt;                                /* Save the count to a local variable                                                  */ \
77        unsigned int recursions[ count ];                         /* Save the current recursion levels to restore them later                             */ \
78        __waitfor_mask_t masks [ count ];                         /* Save the current waitfor masks to restore them later                                */ \
79        __spinlock_t *   locks [ count ];                         /* We need to pass-in an array of locks to BlockInternal                               */ \
80
81#define monitor_save    save   ( monitors, count, locks, recursions, masks )
82#define monitor_restore restore( monitors, count, locks, recursions, masks )
83
84
85//-----------------------------------------------------------------------------
86// Enter/Leave routines
87
88
89extern "C" {
90        // Enter single monitor
91        static void __enter_monitor_desc( monitor_desc * this, const __monitor_group_t & group ) {
92                // Lock the monitor spinlock
93                DO_LOCK( this->lock DEBUG_CTX2 );
94                thread_desc * thrd = this_thread;
95
96                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner);
97
98                if( !this->owner ) {
99                        // No one has the monitor, just take it
100                        set_owner( this, thrd );
101
102                        LIB_DEBUG_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                        LIB_DEBUG_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                        LIB_DEBUG_PRINT_SAFE("Kernel :  mon accepts \n");
118                }
119                else {
120                        LIB_DEBUG_PRINT_SAFE("Kernel :  blocking \n");
121
122                        // Some one else has the monitor, wait in line for it
123                        append( this->entry_queue, thrd );
124                        BlockInternal( &this->lock );
125
126                        LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered  mon %p\n", thrd, this);
127
128                        // BlockInternal will unlock spinlock, no need to unlock ourselves
129                        return;
130                }
131
132                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered  mon %p\n", thrd, this);
133
134                // Release the lock and leave
135                unlock( this->lock );
136                return;
137        }
138
139        static void __enter_monitor_dtor( monitor_desc * this, fptr_t func ) {
140                // Lock the monitor spinlock
141                DO_LOCK( this->lock DEBUG_CTX2 );
142                thread_desc * thrd = this_thread;
143
144                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entering dtor for mon %p (%p)\n", thrd, this, this->owner);
145
146
147                if( !this->owner ) {
148                        LIB_DEBUG_PRINT_SAFE("Kernel : Destroying free mon %p\n", this);
149
150                        // No one has the monitor, just take it
151                        set_owner( this, thrd );
152
153                        unlock( this->lock );
154                        return;
155                }
156                else if( this->owner == thrd) {
157                        // We already have the monitor... but where about to destroy it so the nesting will fail
158                        // Abort!
159                        abortf("Attempt to destroy monitor %p by thread \"%.256s\" (%p) in nested mutex.");
160                }
161
162                __lock_size_t count = 1;
163                monitor_desc ** monitors = &this;
164                __monitor_group_t group = { &this, 1, func };
165                if( is_accepted( this, group) ) {
166                        LIB_DEBUG_PRINT_SAFE("Kernel :  mon accepts dtor, block and signal it \n");
167
168                        // Wake the thread that is waiting for this
169                        __condition_criterion_t * urgent = pop( this->signal_stack );
170                        verify( urgent );
171
172                        // Reset mask
173                        reset_mask( this );
174
175                        // Create the node specific to this wait operation
176                        wait_ctx_primed( this_thread, 0 )
177
178                        // Some one else has the monitor, wait for him to finish and then run
179                        BlockInternal( &this->lock, urgent->owner->waiting_thread );
180
181                        // Some one was waiting for us, enter
182                        set_owner( this, thrd );
183                }
184                else {
185                        LIB_DEBUG_PRINT_SAFE("Kernel :  blocking \n");
186
187                        wait_ctx( this_thread, 0 )
188                        this->dtor_node = &waiter;
189
190                        // Some one else has the monitor, wait in line for it
191                        append( this->entry_queue, thrd );
192                        BlockInternal( &this->lock );
193
194                        // BlockInternal will unlock spinlock, no need to unlock ourselves
195                        return;
196                }
197
198                LIB_DEBUG_PRINT_SAFE("Kernel : Destroying %p\n", this);
199
200        }
201
202        // Leave single monitor
203        void __leave_monitor_desc( monitor_desc * this ) {
204                // Lock the monitor spinlock, DO_LOCK to reduce contention
205                DO_LOCK( this->lock DEBUG_CTX2 );
206
207                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Leaving mon %p (%p)\n", this_thread, this, this->owner);
208
209                verifyf( this_thread == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", this_thread, this->owner, this->recursion, this );
210
211                // Leaving a recursion level, decrement the counter
212                this->recursion -= 1;
213
214                // If we haven't left the last level of recursion
215                // it means we don't need to do anything
216                if( this->recursion != 0) {
217                        LIB_DEBUG_PRINT_SAFE("Kernel :  recursion still %d\n", this->recursion);
218                        unlock( this->lock );
219                        return;
220                }
221
222                // Get the next thread, will be null on low contention monitor
223                thread_desc * new_owner = next_thread( this );
224
225                // We can now let other threads in safely
226                unlock( this->lock );
227
228                //We need to wake-up the thread
229                WakeThread( new_owner );
230        }
231
232        // Leave single monitor for the last time
233        void __leave_dtor_monitor_desc( monitor_desc * this ) {
234                LIB_DEBUG_DO(
235                        if( this_thread != this->owner ) {
236                                abortf("Destroyed monitor %p has inconsistent owner, expected %p got %p.\n", this, this_thread, this->owner);
237                        }
238                        if( this->recursion != 1 ) {
239                                abortf("Destroyed monitor %p has %d outstanding nested calls.\n", this, this->recursion - 1);
240                        }
241                )
242        }
243
244        // Leave the thread monitor
245        // last routine called by a thread.
246        // Should never return
247        void __leave_thread_monitor( thread_desc * thrd ) {
248                monitor_desc * this = &thrd->self_mon;
249
250                // Lock the monitor now
251                DO_LOCK( this->lock DEBUG_CTX2 );
252
253                disable_interrupts();
254
255                thrd->self_cor.state = Halted;
256
257                verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", thrd, this->owner, this->recursion, this );
258
259                // Leaving a recursion level, decrement the counter
260                this->recursion -= 1;
261
262                // If we haven't left the last level of recursion
263                // it must mean there is an error
264                if( this->recursion != 0) { abortf("Thread internal monitor has unbalanced recursion"); }
265
266                // Fetch the next thread, can be null
267                thread_desc * new_owner = next_thread( this );
268
269                // Leave the thread, this will unlock the spinlock
270                // Use leave thread instead of BlockInternal which is
271                // specialized for this case and supports null new_owner
272                LeaveThread( &this->lock, new_owner );
273
274                // Control flow should never reach here!
275        }
276}
277
278// Enter multiple monitor
279// relies on the monitor array being sorted
280static inline void enter( __monitor_group_t monitors ) {
281        for( __lock_size_t i = 0; i < monitors.size; i++) {
282                __enter_monitor_desc( monitors[i], monitors );
283        }
284}
285
286// Leave multiple monitor
287// relies on the monitor array being sorted
288static inline void leave(monitor_desc * monitors [], __lock_size_t count) {
289        for( __lock_size_t i = count - 1; i >= 0; i--) {
290                __leave_monitor_desc( monitors[i] );
291        }
292}
293
294// Ctor for monitor guard
295// Sorts monitors before entering
296void ?{}( monitor_guard_t & this, monitor_desc * m [], __lock_size_t count, fptr_t func ) {
297        // Store current array
298        this.m = m;
299        this.count = count;
300
301        // Sort monitors based on address -> TODO use a sort specialized for small numbers
302        __libcfa_small_sort(this.m, count);
303
304        // Save previous thread context
305        this.prev = this_thread->monitors;
306
307        // Update thread context (needed for conditions)
308        (this_thread->monitors){m, count, func};
309
310        // LIB_DEBUG_PRINT_SAFE("MGUARD : enter %d\n", count);
311
312        // Enter the monitors in order
313        __monitor_group_t group = {this.m, this.count, func};
314        enter( group );
315
316        // LIB_DEBUG_PRINT_SAFE("MGUARD : entered\n");
317}
318
319
320// Dtor for monitor guard
321void ^?{}( monitor_guard_t & this ) {
322        // LIB_DEBUG_PRINT_SAFE("MGUARD : leaving %d\n", this.count);
323
324        // Leave the monitors in order
325        leave( this.m, this.count );
326
327        // LIB_DEBUG_PRINT_SAFE("MGUARD : left\n");
328
329        // Restore thread context
330        this_thread->monitors = this.prev;
331}
332
333// Ctor for monitor guard
334// Sorts monitors before entering
335void ?{}( monitor_dtor_guard_t & this, monitor_desc * m [], fptr_t func ) {
336        // Store current array
337        this.m = *m;
338
339        // Save previous thread context
340        this.prev = this_thread->monitors;
341
342        // Update thread context (needed for conditions)
343        (this_thread->monitors){m, 1, func};
344
345        __enter_monitor_dtor( this.m, func );
346}
347
348// Dtor for monitor guard
349void ^?{}( monitor_dtor_guard_t & this ) {
350        // Leave the monitors in order
351        __leave_dtor_monitor_desc( this.m );
352
353        // Restore thread context
354        this_thread->monitors = this.prev;
355}
356
357//-----------------------------------------------------------------------------
358// Internal scheduling types
359void ?{}(__condition_node_t & this, thread_desc * waiting_thread, __lock_size_t count, uintptr_t user_info ) {
360        this.waiting_thread = waiting_thread;
361        this.count = count;
362        this.next = NULL;
363        this.user_info = user_info;
364}
365
366void ?{}(__condition_criterion_t & this ) {
367        this.ready  = false;
368        this.target = NULL;
369        this.owner  = NULL;
370        this.next   = NULL;
371}
372
373void ?{}(__condition_criterion_t & this, monitor_desc * target, __condition_node_t & owner ) {
374        this.ready  = false;
375        this.target = target;
376        this.owner  = &owner;
377        this.next   = NULL;
378}
379
380//-----------------------------------------------------------------------------
381// Internal scheduling
382void wait( condition & this, uintptr_t user_info = 0 ) {
383        brand_condition( this );
384
385        // Check that everything is as expected
386        assertf( this.monitors != NULL, "Waiting with no monitors (%p)", this.monitors );
387        verifyf( this.monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", this.monitor_count );
388        verifyf( this.monitor_count < 32u, "Excessive monitor count (%"PRIiFAST16")", this.monitor_count );
389
390        // Create storage for monitor context
391        monitor_ctx( this.monitors, this.monitor_count );
392
393        // Create the node specific to this wait operation
394        wait_ctx( this_thread, user_info );
395
396        // Append the current wait operation to the ones already queued on the condition
397        // We don't need locks for that since conditions must always be waited on inside monitor mutual exclusion
398        append( this.blocked, &waiter );
399
400        // Lock all monitors (aggregates the locks as well)
401        lock_all( monitors, locks, count );
402
403        // Find the next thread(s) to run
404        __lock_size_t thread_count = 0;
405        thread_desc * threads[ count ];
406        __builtin_memset( threads, 0, sizeof( threads ) );
407
408        // Save monitor states
409        monitor_save;
410
411        // Remove any duplicate threads
412        for( __lock_size_t i = 0; i < count; i++) {
413                thread_desc * new_owner = next_thread( monitors[i] );
414                insert_unique( threads, thread_count, new_owner );
415        }
416
417        // Everything is ready to go to sleep
418        BlockInternal( locks, count, threads, thread_count );
419
420        // We are back, restore the owners and recursions
421        monitor_restore;
422}
423
424bool signal( condition & this ) {
425        if( is_empty( this ) ) { return false; }
426
427        //Check that everything is as expected
428        verify( this.monitors );
429        verify( this.monitor_count != 0 );
430
431        //Some more checking in debug
432        LIB_DEBUG_DO(
433                thread_desc * this_thrd = this_thread;
434                if ( this.monitor_count != this_thrd->monitors.size ) {
435                        abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", &this, this.monitor_count, this_thrd->monitors.size );
436                }
437
438                for(int i = 0; i < this.monitor_count; i++) {
439                        if ( this.monitors[i] != this_thrd->monitors[i] ) {
440                                abortf( "Signal on condition %p made with different monitor, expected %p got %i", &this, this.monitors[i], this_thrd->monitors[i] );
441                        }
442                }
443        );
444
445        __lock_size_t count = this.monitor_count;
446
447        // Lock all monitors
448        lock_all( this.monitors, NULL, count );
449
450        //Pop the head of the waiting queue
451        __condition_node_t * node = pop_head( this.blocked );
452
453        //Add the thread to the proper AS stack
454        for(int i = 0; i < count; i++) {
455                __condition_criterion_t * crit = &node->criteria[i];
456                assert( !crit->ready );
457                push( crit->target->signal_stack, crit );
458        }
459
460        //Release
461        unlock_all( this.monitors, count );
462
463        return true;
464}
465
466bool signal_block( condition & this ) {
467        if( !this.blocked.head ) { return false; }
468
469        //Check that everything is as expected
470        verifyf( this.monitors != NULL, "Waiting with no monitors (%p)", this.monitors );
471        verifyf( this.monitor_count != 0, "Waiting with 0 monitors (%"PRIiFAST16")", this.monitor_count );
472
473        // Create storage for monitor context
474        monitor_ctx( this.monitors, this.monitor_count );
475
476        // Lock all monitors (aggregates the locks them as well)
477        lock_all( monitors, locks, count );
478
479        // Create the node specific to this wait operation
480        wait_ctx_primed( this_thread, 0 )
481
482        //save contexts
483        monitor_save;
484
485        //Find the thread to run
486        thread_desc * signallee = pop_head( this.blocked )->waiting_thread;
487        set_owner( monitors, count, signallee );
488
489        LIB_DEBUG_PRINT_BUFFER_DECL( "Kernel : signal_block condition %p (s: %p)\n", &this, signallee );
490
491        //Everything is ready to go to sleep
492        BlockInternal( locks, count, &signallee, 1 );
493
494
495        // WE WOKE UP
496
497
498        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel :   signal_block returned\n" );
499
500        //We are back, restore the masks and recursions
501        monitor_restore;
502
503        return true;
504}
505
506// Access the user_info of the thread waiting at the front of the queue
507uintptr_t front( condition & this ) {
508        verifyf( !is_empty(this),
509                "Attempt to access user data on an empty condition.\n"
510                "Possible cause is not checking if the condition is empty before reading stored data."
511        );
512        return ((typeof(this.blocked.head))this.blocked.head)->user_info;
513}
514
515//-----------------------------------------------------------------------------
516// External scheduling
517// cases to handle :
518//      - target already there :
519//              block and wake
520//      - dtor already there
521//              put thread on signaller stack
522//      - non-blocking
523//              return else
524//      - timeout
525//              return timeout
526//      - block
527//              setup mask
528//              block
529void __waitfor_internal( const __waitfor_mask_t & mask, int duration ) {
530        // This statment doesn't have a contiguous list of monitors...
531        // Create one!
532        __lock_size_t max = count_max( mask );
533        monitor_desc * mon_storage[max];
534        __builtin_memset( mon_storage, 0, sizeof( mon_storage ) );
535        __lock_size_t actual_count = aggregate( mon_storage, mask );
536
537        LIB_DEBUG_PRINT_BUFFER_DECL( "Kernel : waitfor %d (s: %d, m: %d)\n", actual_count, mask.size, (__lock_size_t)max);
538
539        if(actual_count == 0) return;
540
541        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : waitfor internal proceeding\n");
542
543        // Create storage for monitor context
544        monitor_ctx( mon_storage, actual_count );
545
546        // Lock all monitors (aggregates the locks as well)
547        lock_all( monitors, locks, count );
548
549        {
550                // Check if the entry queue
551                thread_desc * next; int index;
552                [next, index] = search_entry_queue( mask, monitors, count );
553
554                if( next ) {
555                        *mask.accepted = index;
556                        __acceptable_t& accepted = mask[index];
557                        if( accepted.is_dtor ) {
558                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : dtor already there\n");
559                                verifyf( accepted.size == 1,  "ERROR: Accepted dtor has more than 1 mutex parameter." );
560
561                                monitor_desc * mon2dtor = accepted[0];
562                                verifyf( mon2dtor->dtor_node, "ERROR: Accepted monitor has no dtor_node." );
563
564                                __condition_criterion_t * dtor_crit = mon2dtor->dtor_node->criteria;
565                                push( mon2dtor->signal_stack, dtor_crit );
566
567                                unlock_all( locks, count );
568                        }
569                        else {
570                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : thread present, baton-passing\n");
571
572                                // Create the node specific to this wait operation
573                                wait_ctx_primed( this_thread, 0 );
574
575                                // Save monitor states
576                                monitor_save;
577
578                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel :  baton of %d monitors : ", count );
579                                #ifdef __CFA_DEBUG_PRINT__
580                                        for( int i = 0; i < count; i++) {
581                                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "%p %p ", monitors[i], monitors[i]->signal_stack.top );
582                                        }
583                                #endif
584                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "\n");
585
586                                // Set the owners to be the next thread
587                                set_owner( monitors, count, next );
588
589                                // Everything is ready to go to sleep
590                                BlockInternal( locks, count, &next, 1 );
591
592                                // We are back, restore the owners and recursions
593                                monitor_restore;
594
595                                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : thread present, returned\n");
596                        }
597
598                        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : accepted %d\n", *mask.accepted);
599                        return;
600                }
601        }
602
603
604        if( duration == 0 ) {
605                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : non-blocking, exiting\n");
606
607                unlock_all( locks, count );
608
609                LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : accepted %d\n", *mask.accepted);
610                return;
611        }
612
613
614        verifyf( duration < 0, "Timeout on waitfor statments not supported yet.");
615
616        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : blocking waitfor\n");
617
618        // Create the node specific to this wait operation
619        wait_ctx_primed( this_thread, 0 );
620
621        monitor_save;
622        set_mask( monitors, count, mask );
623
624        for( __lock_size_t i = 0; i < count; i++) {
625                verify( monitors[i]->owner == this_thread );
626        }
627
628        //Everything is ready to go to sleep
629        BlockInternal( locks, count );
630
631
632        // WE WOKE UP
633
634
635        //We are back, restore the masks and recursions
636        monitor_restore;
637
638        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : exiting\n");
639
640        LIB_DEBUG_PRINT_BUFFER_LOCAL( "Kernel : accepted %d\n", *mask.accepted);
641}
642
643//-----------------------------------------------------------------------------
644// Utilities
645
646static inline void set_owner( monitor_desc * this, thread_desc * owner ) {
647        // LIB_DEBUG_PRINT_SAFE("Kernal :   Setting owner of %p to %p ( was %p)\n", this, owner, this->owner );
648
649        //Pass the monitor appropriately
650        this->owner = owner;
651
652        //We are passing the monitor to someone else, which means recursion level is not 0
653        this->recursion = owner ? 1 : 0;
654}
655
656static inline void set_owner( monitor_desc * monitors [], __lock_size_t count, thread_desc * owner ) {
657        monitors[0]->owner     = owner;
658        monitors[0]->recursion = 1;
659        for( __lock_size_t i = 1; i < count; i++ ) {
660                monitors[i]->owner     = owner;
661                monitors[i]->recursion = 0;
662        }
663}
664
665static inline void set_mask( monitor_desc * storage [], __lock_size_t count, const __waitfor_mask_t & mask ) {
666        for( __lock_size_t i = 0; i < count; i++) {
667                storage[i]->mask = mask;
668        }
669}
670
671static inline void reset_mask( monitor_desc * this ) {
672        this->mask.accepted = NULL;
673        this->mask.data = NULL;
674        this->mask.size = 0;
675}
676
677static inline thread_desc * next_thread( monitor_desc * this ) {
678        //Check the signaller stack
679        LIB_DEBUG_PRINT_SAFE("Kernel :  mon %p AS-stack top %p\n", this, this->signal_stack.top);
680        __condition_criterion_t * urgent = pop( this->signal_stack );
681        if( urgent ) {
682                //The signaller stack is not empty,
683                //regardless of if we are ready to baton pass,
684                //we need to set the monitor as in use
685                set_owner( this,  urgent->owner->waiting_thread );
686
687                return check_condition( urgent );
688        }
689
690        // No signaller thread
691        // Get the next thread in the entry_queue
692        thread_desc * new_owner = pop_head( this->entry_queue );
693        set_owner( this, new_owner );
694
695        return new_owner;
696}
697
698static inline bool is_accepted( monitor_desc * this, const __monitor_group_t & group ) {
699        __acceptable_t * it = this->mask.data; // Optim
700        __lock_size_t count = this->mask.size;
701
702        // Check if there are any acceptable functions
703        if( !it ) return false;
704
705        // If this isn't the first monitor to test this, there is no reason to repeat the test.
706        if( this != group[0] ) return group[0]->mask.accepted >= 0;
707
708        // For all acceptable functions check if this is the current function.
709        for( __lock_size_t i = 0; i < count; i++, it++ ) {
710                if( *it == group ) {
711                        *this->mask.accepted = i;
712                        return true;
713                }
714        }
715
716        // No function matched
717        return false;
718}
719
720static inline void init( __lock_size_t count, monitor_desc * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
721        for( __lock_size_t i = 0; i < count; i++) {
722                (criteria[i]){ monitors[i], waiter };
723        }
724
725        waiter.criteria = criteria;
726}
727
728static inline void init_push( __lock_size_t count, monitor_desc * monitors [], __condition_node_t & waiter, __condition_criterion_t criteria [] ) {
729        for( __lock_size_t i = 0; i < count; i++) {
730                (criteria[i]){ monitors[i], waiter };
731                LIB_DEBUG_PRINT_SAFE( "Kernel :  target %p = %p\n", criteria[i].target, &criteria[i] );
732                push( criteria[i].target->signal_stack, &criteria[i] );
733        }
734
735        waiter.criteria = criteria;
736}
737
738static inline void lock_all( __spinlock_t * locks [], __lock_size_t count ) {
739        for( __lock_size_t i = 0; i < count; i++ ) {
740                DO_LOCK( *locks[i] DEBUG_CTX2 );
741        }
742}
743
744static inline void lock_all( monitor_desc * source [], __spinlock_t * /*out*/ locks [], __lock_size_t count ) {
745        for( __lock_size_t i = 0; i < count; i++ ) {
746                __spinlock_t * l = &source[i]->lock;
747                DO_LOCK( *l DEBUG_CTX2 );
748                if(locks) locks[i] = l;
749        }
750}
751
752static inline void unlock_all( __spinlock_t * locks [], __lock_size_t count ) {
753        for( __lock_size_t i = 0; i < count; i++ ) {
754                unlock( *locks[i] );
755        }
756}
757
758static inline void unlock_all( monitor_desc * locks [], __lock_size_t count ) {
759        for( __lock_size_t i = 0; i < count; i++ ) {
760                unlock( locks[i]->lock );
761        }
762}
763
764static inline void save(
765        monitor_desc * ctx [],
766        __lock_size_t count,
767        __attribute((unused)) __spinlock_t * locks [],
768        unsigned int /*out*/ recursions [],
769        __waitfor_mask_t /*out*/ masks []
770) {
771        for( __lock_size_t i = 0; i < count; i++ ) {
772                recursions[i] = ctx[i]->recursion;
773                masks[i]      = ctx[i]->mask;
774        }
775}
776
777static inline void restore(
778        monitor_desc * ctx [],
779        __lock_size_t count,
780        __spinlock_t * locks [],
781        unsigned int /*out*/ recursions [],
782        __waitfor_mask_t /*out*/ masks []
783) {
784        lock_all( locks, count );
785        for( __lock_size_t i = 0; i < count; i++ ) {
786                ctx[i]->recursion = recursions[i];
787                ctx[i]->mask      = masks[i];
788        }
789        unlock_all( locks, count );
790}
791
792// Function has 2 different behavior
793// 1 - Marks a monitors as being ready to run
794// 2 - Checks if all the monitors are ready to run
795//     if so return the thread to run
796static inline thread_desc * check_condition( __condition_criterion_t * target ) {
797        __condition_node_t * node = target->owner;
798        unsigned short count = node->count;
799        __condition_criterion_t * criteria = node->criteria;
800
801        bool ready2run = true;
802
803        for(    int i = 0; i < count; i++ ) {
804
805                // LIB_DEBUG_PRINT_SAFE( "Checking %p for %p\n", &criteria[i], target );
806                if( &criteria[i] == target ) {
807                        criteria[i].ready = true;
808                        // LIB_DEBUG_PRINT_SAFE( "True\n" );
809                }
810
811                ready2run = criteria[i].ready && ready2run;
812        }
813
814        LIB_DEBUG_PRINT_SAFE( "Kernel :  Runing %i (%p)\n", ready2run, ready2run ? node->waiting_thread : NULL );
815        return ready2run ? node->waiting_thread : NULL;
816}
817
818static inline void brand_condition( condition & this ) {
819        thread_desc * thrd = this_thread;
820        if( !this.monitors ) {
821                // LIB_DEBUG_PRINT_SAFE("Branding\n");
822                assertf( thrd->monitors.data != NULL, "No current monitor to brand condition %p", thrd->monitors.data );
823                this.monitor_count = thrd->monitors.size;
824
825                this.monitors = (monitor_desc **)malloc( this.monitor_count * sizeof( *this.monitors ) );
826                for( int i = 0; i < this.monitor_count; i++ ) {
827                        this.monitors[i] = thrd->monitors[i];
828                }
829        }
830}
831
832static inline [thread_desc *, int] search_entry_queue( const __waitfor_mask_t & mask, monitor_desc * monitors [], __lock_size_t count ) {
833
834        __queue_t(thread_desc) & entry_queue = monitors[0]->entry_queue;
835
836        // For each thread in the entry-queue
837        for(    thread_desc ** thrd_it = &entry_queue.head;
838                *thrd_it;
839                thrd_it = &(*thrd_it)->next
840        ) {
841                // For each acceptable check if it matches
842                int i = 0;
843                __acceptable_t * end   = end  (mask);
844                __acceptable_t * begin = begin(mask);
845                for( __acceptable_t * it = begin; it != end; it++, i++ ) {
846                        // Check if we have a match
847                        if( *it == (*thrd_it)->monitors ) {
848
849                                // If we have a match return it
850                                // after removeing it from the entry queue
851                                return [remove( entry_queue, thrd_it ), i];
852                        }
853                }
854        }
855
856        return [0, -1];
857}
858
859forall(dtype T | sized( T ))
860static inline __lock_size_t insert_unique( T * array [], __lock_size_t & size, T * val ) {
861        if( !val ) return size;
862
863        for( __lock_size_t i = 0; i <= size; i++) {
864                if( array[i] == val ) return size;
865        }
866
867        array[size] = val;
868        size = size + 1;
869        return size;
870}
871
872static inline __lock_size_t count_max( const __waitfor_mask_t & mask ) {
873        __lock_size_t max = 0;
874        for( __lock_size_t i = 0; i < mask.size; i++ ) {
875                __acceptable_t & accepted = mask[i];
876                max += accepted.size;
877        }
878        return max;
879}
880
881static inline __lock_size_t aggregate( monitor_desc * storage [], const __waitfor_mask_t & mask ) {
882        __lock_size_t size = 0;
883        for( __lock_size_t i = 0; i < mask.size; i++ ) {
884                __acceptable_t & accepted = mask[i];
885                __libcfa_small_sort( accepted.data, accepted.size );
886                for( __lock_size_t j = 0; j < accepted.size; j++) {
887                        insert_unique( storage, size, accepted[j] );
888                }
889        }
890        // TODO insertion sort instead of this
891        __libcfa_small_sort( storage, size );
892        return size;
893}
894
895// Local Variables: //
896// mode: c //
897// tab-width: 4 //
898// End: //
Note: See TracBrowser for help on using the repository browser.