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

ADTaaron-thesisarm-ehast-experimentalcleanup-dtorsdeferred_resndemanglerenumforall-pointer-decayjacob/cs343-translationjenkins-sandboxnew-astnew-ast-unique-exprnew-envno_listpersistent-indexerpthread-emulationqualifiedEnumresolv-newwith_gc
Last change on this file since e1c1829 was 4aa2fb2, checked in by Thierry Delisle <tdelisle@…>, 7 years ago

Added verify macro for asserts only present in debug

  • Property mode set to 100644
File size: 14.6 KB
Line 
1//                              -*- Mode: CFA -*-
2//
3// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo
4//
5// The contents of this file are covered under the licence agreement in the
6// file "LICENCE" distributed with Cforall.
7//
8// monitor_desc.c --
9//
10// Author           : Thierry Delisle
11// Created On       : Thd Feb 23 12:27:26 2017
12// Last Modified By : Thierry Delisle
13// Last Modified On : --
14// Update Count     : 0
15//
16
17#include "monitor"
18
19#include <stdlib>
20
21#include "kernel_private.h"
22#include "libhdr.h"
23
24//-----------------------------------------------------------------------------
25// Forward declarations
26static inline void set_owner( monitor_desc * this, thread_desc * owner );
27static inline thread_desc * next_thread( monitor_desc * this );
28
29static inline void lock_all( spinlock ** locks, unsigned short count );
30static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count );
31static inline void unlock_all( spinlock ** locks, unsigned short count );
32static inline void unlock_all( monitor_desc ** locks, unsigned short count );
33
34static inline void save_recursion   ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count );
35static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count );
36
37static inline thread_desc * check_condition( __condition_criterion_t * );
38static inline void brand_condition( condition * );
39static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val );
40
41//-----------------------------------------------------------------------------
42// Enter/Leave routines
43
44
45extern "C" {
46        void __enter_monitor_desc(monitor_desc * this) {
47                lock( &this->lock );
48                thread_desc * thrd = this_thread();
49
50                LIB_DEBUG_PRINT_SAFE("%p Entering %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
51
52                if( !this->owner ) {
53                        //No one has the monitor, just take it
54                        set_owner( this, thrd );
55                }
56                else if( this->owner == thrd) {
57                        //We already have the monitor, just not how many times we took it
58                        verify( this->recursion > 0 );
59                        this->recursion += 1;
60                }
61                else {
62                        //Some one else has the monitor, wait in line for it
63                        append( &this->entry_queue, thrd );
64                        LIB_DEBUG_PRINT_SAFE("%p Blocking on entry\n", thrd);
65                        ScheduleInternal( &this->lock );
66
67                        //ScheduleInternal will unlock spinlock, no need to unlock ourselves
68                        return; 
69                }
70
71                unlock( &this->lock );
72                return;
73        }
74
75        // leave pseudo code :
76        //      TODO
77        void __leave_monitor_desc(monitor_desc * this) {
78                lock( &this->lock );
79
80                LIB_DEBUG_PRINT_SAFE("%p Leaving %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
81                verifyf( this_thread() == this->owner, "Expected owner to be %p, got %p (r: %i)", this_thread(), this->owner, this->recursion );
82
83                //Leaving a recursion level, decrement the counter
84                this->recursion -= 1;
85
86                //If we haven't left the last level of recursion
87                //it means we don't need to do anything
88                if( this->recursion != 0) {
89                        unlock( &this->lock );
90                        return;
91                }
92
93                thread_desc * new_owner = next_thread( this );
94
95                //We can now let other threads in safely
96                unlock( &this->lock );
97
98                LIB_DEBUG_PRINT_SAFE("Next owner is %p\n", new_owner);
99
100                //We need to wake-up the thread
101                ScheduleThread( new_owner );
102        }
103}
104
105static inline void enter(monitor_desc ** monitors, int count) {
106        for(int i = 0; i < count; i++) {
107                __enter_monitor_desc( monitors[i] );
108        }
109}
110
111static inline void leave(monitor_desc ** monitors, int count) {
112        for(int i = count - 1; i >= 0; i--) {
113                __leave_monitor_desc( monitors[i] );
114        }
115}
116
117void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ) {
118        this->m = m;
119        this->count = count;
120        qsort(this->m, count);
121        enter( this->m, this->count );
122
123        this->prev_mntrs = this_thread()->current_monitors;
124        this->prev_count = this_thread()->current_monitor_count;
125
126        this_thread()->current_monitors      = m;
127        this_thread()->current_monitor_count = count;
128}
129
130void ^?{}( monitor_guard_t * this ) {
131        leave( this->m, this->count );
132
133        this_thread()->current_monitors      = this->prev_mntrs;
134        this_thread()->current_monitor_count = this->prev_count;
135}
136
137void ?{}(__condition_node_t * this, thread_desc * waiting_thread, unsigned short count, uintptr_t user_info ) {
138        this->waiting_thread = waiting_thread;
139        this->count = count;
140        this->next = NULL;
141        this->user_info = user_info;
142}
143
144void ?{}(__condition_criterion_t * this ) {
145        this->ready  = false;
146        this->target = NULL;
147        this->owner  = NULL;
148        this->next   = NULL;
149}
150
151void ?{}(__condition_criterion_t * this, monitor_desc * target, __condition_node_t * owner ) {
152        this->ready  = false;
153        this->target = target;
154        this->owner  = owner;
155        this->next   = NULL;
156}
157
158//-----------------------------------------------------------------------------
159// Internal scheduling
160void wait( condition * this, uintptr_t user_info = 0 ) {
161        LIB_DEBUG_PRINT_SAFE("Waiting\n");
162
163        brand_condition( this );
164
165        //Check that everything is as expected
166        assertf( this->monitors != NULL, "Waiting with no monitors (%p)", this->monitors );
167        verifyf( this->monitor_count != 0, "Waiting with 0 monitors (%i)", this->monitor_count );
168        verifyf( this->monitor_count < 32u, "Excessive monitor count (%i)", this->monitor_count );
169
170        unsigned short count = this->monitor_count;
171        unsigned int recursions[ count ];               //Save the current recursion levels to restore them later
172        spinlock *   locks     [ count ];               //We need to pass-in an array of locks to ScheduleInternal
173
174        LIB_DEBUG_PRINT_SAFE("count %i\n", count);
175
176        __condition_node_t waiter = { this_thread(), count, user_info };
177
178        __condition_criterion_t criteria[count];
179        for(int i = 0; i < count; i++) {
180                (&criteria[i]){ this->monitors[i], &waiter };
181                LIB_DEBUG_PRINT_SAFE( "Criterion %p\n", &criteria[i] );
182        }
183
184        waiter.criteria = criteria;
185        append( &this->blocked, &waiter );
186
187        lock_all( this->monitors, locks, count );
188        save_recursion( this->monitors, recursions, count );
189        //DON'T unlock, ask the kernel to do it
190
191        //Find the next thread(s) to run
192        unsigned short thread_count = 0;
193        thread_desc * threads[ count ];
194        for(int i = 0; i < count; i++) {
195                threads[i] = 0;
196        }
197
198        for( int i = 0; i < count; i++) {
199                thread_desc * new_owner = next_thread( this->monitors[i] );
200                thread_count = insert_unique( threads, thread_count, new_owner );
201        }
202
203        LIB_DEBUG_PRINT_SAFE("Will unblock: ");
204        for(int i = 0; i < thread_count; i++) {
205                LIB_DEBUG_PRINT_SAFE("%p ", threads[i]);
206        }
207        LIB_DEBUG_PRINT_SAFE("\n");
208
209        // Everything is ready to go to sleep
210        ScheduleInternal( locks, count, threads, thread_count );
211
212
213        //WE WOKE UP
214
215
216        //We are back, restore the owners and recursions
217        lock_all( locks, count );
218        restore_recursion( this->monitors, recursions, count );
219        unlock_all( locks, count );
220}
221
222bool signal( condition * this ) {
223        if( is_empty( this ) ) {
224                LIB_DEBUG_PRINT_SAFE("Nothing to signal\n");
225                return false;
226        }
227
228        //Check that everything is as expected
229        verify( this->monitors );
230        verify( this->monitor_count != 0 );
231
232        unsigned short count = this->monitor_count;
233       
234        //Some more checking in debug
235        LIB_DEBUG_DO(
236                thread_desc * this_thrd = this_thread();
237                if ( this->monitor_count != this_thrd->current_monitor_count ) {
238                        abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", this, this->monitor_count, this_thrd->current_monitor_count );
239                } // if
240
241                for(int i = 0; i < this->monitor_count; i++) {
242                        if ( this->monitors[i] != this_thrd->current_monitors[i] ) {
243                                abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->current_monitors[i] );
244                        } // if
245                }
246        );
247
248        //Lock all the monitors
249        lock_all( this->monitors, NULL, count );
250        LIB_DEBUG_PRINT_SAFE("Signalling");
251
252        //Pop the head of the waiting queue
253        __condition_node_t * node = pop_head( &this->blocked );
254
255        //Add the thread to the proper AS stack
256        for(int i = 0; i < count; i++) {
257                __condition_criterion_t * crit = &node->criteria[i];
258                LIB_DEBUG_PRINT_SAFE(" %p", crit->target);
259                assert( !crit->ready );
260                push( &crit->target->signal_stack, crit );
261        }
262
263        LIB_DEBUG_PRINT_SAFE("\n");
264
265        //Release
266        unlock_all( this->monitors, count );
267
268        return true;
269}
270
271bool signal_block( condition * this ) {
272        if( !this->blocked.head ) {
273                LIB_DEBUG_PRINT_SAFE("Nothing to signal\n");
274                return false;
275        }
276
277        //Check that everything is as expected
278        verifyf( this->monitors != NULL, "Waiting with no monitors (%p)", this->monitors );
279        verifyf( this->monitor_count != 0, "Waiting with 0 monitors (%i)", this->monitor_count );
280
281        unsigned short count = this->monitor_count;
282        unsigned int recursions[ count ];               //Save the current recursion levels to restore them later
283        spinlock *   locks     [ count ];               //We need to pass-in an array of locks to ScheduleInternal
284
285        lock_all( this->monitors, locks, count );
286
287        //create creteria
288        __condition_node_t waiter = { this_thread(), count, 0 };
289
290        __condition_criterion_t criteria[count];
291        for(int i = 0; i < count; i++) {
292                (&criteria[i]){ this->monitors[i], &waiter };
293                LIB_DEBUG_PRINT_SAFE( "Criterion %p\n", &criteria[i] );
294                push( &criteria[i].target->signal_stack, &criteria[i] );
295        }
296
297        waiter.criteria = criteria;
298
299        //save contexts
300        save_recursion( this->monitors, recursions, count );
301
302        //Find the thread to run
303        thread_desc * signallee = pop_head( &this->blocked )->waiting_thread;
304        for(int i = 0; i < count; i++) {
305                set_owner( this->monitors[i], signallee );
306        }
307
308        LIB_DEBUG_PRINT_SAFE( "Waiting on signal block\n" );
309
310        //Everything is ready to go to sleep
311        ScheduleInternal( locks, count, &signallee, 1 );
312
313
314
315
316        LIB_DEBUG_PRINT_SAFE( "Back from signal block\n" );
317
318        //We are back, restore the owners and recursions
319        lock_all( locks, count );
320        restore_recursion( this->monitors, recursions, count );
321        unlock_all( locks, count );
322
323        return true;
324}
325
326uintptr_t front( condition * this ) {
327        verifyf( !is_empty(this), 
328                "Attempt to access user data on an empty condition.\n"
329                "Possible cause is not checking if the condition is empty before reading stored data."
330        );
331        return this->blocked.head->user_info;
332}
333
334//-----------------------------------------------------------------------------
335// Internal scheduling
336void __accept_internal( unsigned short count, __acceptable_t * acceptables, void (*func)(void) ) {
337        // thread_desc * this = this_thread();
338
339        // unsigned short count = this->current_monitor_count;
340        // unsigned int recursions[ count ];            //Save the current recursion levels to restore them later
341        // spinlock *   locks     [ count ];            //We need to pass-in an array of locks to ScheduleInternal
342
343        // lock_all( this->current_monitors, locks, count );
344
345
346
347
348
349        // // // Everything is ready to go to sleep
350        // // ScheduleInternal( locks, count, threads, thread_count );
351
352
353        // //WE WOKE UP
354
355
356        // //We are back, restore the owners and recursions
357        // lock_all( locks, count );
358        // restore_recursion( this->monitors, recursions, count );
359        // unlock_all( locks, count );
360}
361
362//-----------------------------------------------------------------------------
363// Utilities
364
365static inline void set_owner( monitor_desc * this, thread_desc * owner ) {
366        //Pass the monitor appropriately
367        this->owner = owner;
368
369        //We are passing the monitor to someone else, which means recursion level is not 0
370        this->recursion = owner ? 1 : 0;
371}
372
373static inline thread_desc * next_thread( monitor_desc * this ) {
374        //Check the signaller stack
375        __condition_criterion_t * urgent = pop( &this->signal_stack );
376        if( urgent ) {
377                //The signaller stack is not empty,
378                //regardless of if we are ready to baton pass,
379                //we need to set the monitor as in use
380                set_owner( this,  urgent->owner->waiting_thread );
381
382                return check_condition( urgent );
383        }
384
385        // No signaller thread
386        // Get the next thread in the entry_queue
387        thread_desc * new_owner = pop_head( &this->entry_queue );
388        set_owner( this, new_owner );
389
390        return new_owner;
391}
392
393static inline void lock_all( spinlock ** locks, unsigned short count ) {
394        for( int i = 0; i < count; i++ ) {
395                lock( locks[i] );
396        }
397}
398
399static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count ) {
400        for( int i = 0; i < count; i++ ) {
401                spinlock * l = &source[i]->lock;
402                lock( l );
403                if(locks) locks[i] = l;
404        }
405}
406
407static inline void unlock_all( spinlock ** locks, unsigned short count ) {
408        for( int i = 0; i < count; i++ ) {
409                unlock( locks[i] );
410        }
411}
412
413static inline void unlock_all( monitor_desc ** locks, unsigned short count ) {
414        for( int i = 0; i < count; i++ ) {
415                unlock( &locks[i]->lock );
416        }
417}
418
419
420static inline void save_recursion   ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count ) {
421        for( int i = 0; i < count; i++ ) {
422                recursions[i] = ctx[i]->recursion;
423        }
424}
425
426static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count ) {
427        for( int i = 0; i < count; i++ ) {
428                ctx[i]->recursion = recursions[i];
429        }
430}
431
432// Function has 2 different behavior
433// 1 - Marks a monitors as being ready to run
434// 2 - Checks if all the monitors are ready to run
435//     if so return the thread to run
436static inline thread_desc * check_condition( __condition_criterion_t * target ) {
437        __condition_node_t * node = target->owner;
438        unsigned short count = node->count;
439        __condition_criterion_t * criteria = node->criteria;
440
441        bool ready2run = true;
442
443        for(    int i = 0; i < count; i++ ) {
444
445                LIB_DEBUG_PRINT_SAFE( "Checking %p for %p\n", &criteria[i], target );
446                if( &criteria[i] == target ) {
447                        criteria[i].ready = true;
448                        LIB_DEBUG_PRINT_SAFE( "True\n" );
449                }
450
451                ready2run = criteria[i].ready && ready2run;
452        }
453
454        LIB_DEBUG_PRINT_SAFE( "Runing %i\n", ready2run );
455        return ready2run ? node->waiting_thread : NULL;
456}
457
458static inline void brand_condition( condition * this ) {
459        thread_desc * thrd = this_thread();
460        if( !this->monitors ) {
461                LIB_DEBUG_PRINT_SAFE("Branding\n");
462                assertf( thrd->current_monitors != NULL, "No current monitor to brand condition", thrd->current_monitors );
463                this->monitor_count = thrd->current_monitor_count;
464
465                this->monitors = malloc( this->monitor_count * sizeof( *this->monitors ) );
466                for( int i = 0; i < this->monitor_count; i++ ) {
467                        this->monitors[i] = thrd->current_monitors[i];
468                }
469        }
470}
471
472static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val ) {
473        if( !val ) return end;
474
475        for(int i = 0; i <= end; i++) {
476                if( thrds[i] == val ) return end;
477        }
478
479        thrds[end] = val;
480        return end + 1;
481}
482
483void ?{}( __condition_blocked_queue_t * this ) {
484        this->head = NULL;
485        this->tail = &this->head;
486}
487
488void append( __condition_blocked_queue_t * this, __condition_node_t * c ) {
489        verify(this->tail != NULL);
490        *this->tail = c;
491        this->tail = &c->next;
492}
493
494__condition_node_t * pop_head( __condition_blocked_queue_t * this ) {
495        __condition_node_t * head = this->head;
496        if( head ) {
497                this->head = head->next;
498                if( !head->next ) {
499                        this->tail = &this->head;
500                }
501                head->next = NULL;
502        }
503        return head;
504}
Note: See TracBrowser for help on using the repository browser.