source: src/libcfa/concurrency/monitor.c @ 0c78741

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

Implementation of internal scheduling in CFA

  • Property mode set to 100644
File size: 10.9 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 "kernel_private.h"
20#include "libhdr.h"
21
22//-----------------------------------------------------------------------------
23// Forward declarations
24static inline void set_owner( monitor_desc * this, thread_desc * owner );
25static inline thread_desc * next_thread( monitor_desc * this );
26
27static inline void lock_all( spinlock ** locks, unsigned short count );
28static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count );
29static inline void unlock_all( spinlock ** locks, unsigned short count );
30static inline void unlock_all( monitor_desc ** locks, unsigned short count );
31
32static inline void save_recursion   ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count );
33static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count );
34
35static inline thread_desc * check_condition( __condition_criterion_t * );
36static inline void brand_condition( condition * );
37static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val );
38
39//-----------------------------------------------------------------------------
40// Enter/Leave routines
41
42
43extern "C" {
44        void __enter_monitor_desc(monitor_desc * this) {
45                lock( &this->lock );
46                thread_desc * thrd = this_thread();
47
48                LIB_DEBUG_PRINT_SAFE("%p Entering %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
49
50                if( !this->owner ) {
51                        //No one has the monitor, just take it
52                        set_owner( this, thrd );
53                }
54                else if( this->owner == thrd) {
55                        //We already have the monitor, just not how many times we took it
56                        assert( this->recursion > 0 );
57                        this->recursion += 1;
58                }
59                else {
60                        //Some one else has the monitor, wait in line for it
61                        append( &this->entry_queue, thrd );
62                        ScheduleInternal( &this->lock );
63
64                        //ScheduleInternal will unlock spinlock, no need to unlock ourselves
65                        return; 
66                }
67
68                unlock( &this->lock );
69                return;
70        }
71
72        // leave pseudo code :
73        //      TODO
74        void __leave_monitor_desc(monitor_desc * this) {
75                lock( &this->lock );
76
77                thread_desc * thrd = this_thread();
78
79                LIB_DEBUG_PRINT_SAFE("%p Leaving %p (o: %p, r: %i)\n", thrd, this, this->owner, this->recursion);
80                assertf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i)", thrd, this->owner, this->recursion );
81
82                //Leaving a recursion level, decrement the counter
83                this->recursion -= 1;
84
85                //If we haven't left the last level of recursion
86                //it means we don't need to do anything
87                if( this->recursion != 0) {
88                        unlock( &this->lock );
89                        return;
90                }
91
92                thread_desc * new_owner = next_thread( this );
93
94                //We can now let other threads in safely
95                unlock( &this->lock );
96
97                //We need to wake-up the thread
98                ScheduleThread( new_owner );
99        }
100}
101
102static inline void enter(monitor_desc ** monitors, int count) {
103        for(int i = 0; i < count; i++) {
104                __enter_monitor_desc( monitors[i] );
105        }
106}
107
108static inline void leave(monitor_desc ** monitors, int count) {
109        for(int i = count - 1; i >= 0; i--) {
110                __leave_monitor_desc( monitors[i] );
111        }
112}
113
114void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ) {
115        this->m = m;
116        this->count = count;
117        qsort(this->m, count);
118        enter( this->m, this->count );
119
120        this->prev_mntrs = this_thread()->current_monitors;
121        this->prev_count = this_thread()->current_monitor_count;
122
123        this_thread()->current_monitors      = m;
124        this_thread()->current_monitor_count = count;
125}
126
127void ^?{}( monitor_guard_t * this ) {
128        leave( this->m, this->count );
129
130        this_thread()->current_monitors      = this->prev_mntrs;
131        this_thread()->current_monitor_count = this->prev_count;
132}
133
134//-----------------------------------------------------------------------------
135// Internal scheduling
136void wait( condition * this ) {
137        LIB_DEBUG_PRINT_SAFE("Waiting\n");
138
139        brand_condition( this );
140
141        //Check that everything is as expected
142        assertf( this->monitors != NULL, "Waiting with no monitors (%p)", this->monitors );
143        assertf( this->monitor_count != 0, "Waiting with 0 monitors (%i)", this->monitor_count );
144
145        unsigned short count = this->monitor_count;
146        unsigned int blarg_recursions[ count ];         //Save the current recursion levels to restore them later
147        spinlock *   blarg_locks     [ count ];         //We need to pass-in an array of locks to ScheduleInternal
148
149        LIB_DEBUG_PRINT_SAFE("count %i\n", count);
150
151        __condition_node_t waiter;
152        waiter.waiting_thread = this_thread();
153        waiter.count = count;
154        waiter.next = NULL;
155
156        __condition_criterion_t criteria[count];
157        for(int i = 0; i < count; i++) {
158                criteria[i].ready  = false;
159                criteria[i].target = this->monitors[i];
160                criteria[i].owner  = &waiter;
161                criteria[i].next   = NULL;
162                LIB_DEBUG_PRINT_SAFE( "Criterion %p\n", &criteria[i] );
163        }
164
165        waiter.criteria = criteria;
166        append( &this->blocked, &waiter );
167
168        lock_all( this->monitors, blarg_locks, count );
169        save_recursion( this->monitors, blarg_recursions, count );
170        //DON'T unlock, ask the kernel to do it
171
172        //Find the next thread(s) to run
173        unsigned short thread_count = count;
174        thread_desc * threads[ count ];
175
176        for( int i = 0; i < count; i++) {
177                thread_desc * new_owner = next_thread( this->monitors[i] );
178                thread_count = insert_unique( threads, i, new_owner );
179        }
180
181        LIB_DEBUG_PRINT_SAFE("Will unblock: ");
182        for(int i = 0; i < thread_count; i++) {
183                LIB_DEBUG_PRINT_SAFE("%p ", threads[i]);
184        }
185        LIB_DEBUG_PRINT_SAFE("\n");
186
187        //Everything is ready to go to sleep
188        ScheduleInternal( blarg_locks, count, threads, thread_count );
189
190
191        //WE WOKE UP
192
193
194        //We are back, restore the owners and recursions
195        lock_all( blarg_locks, count );
196        restore_recursion( this->monitors, blarg_recursions, count );
197        unlock_all( blarg_locks, count );
198}
199
200void signal( condition * this ) {
201        if( !this->blocked.head ) {
202                LIB_DEBUG_PRINT_SAFE("Nothing to signal\n");
203                return;
204        }
205
206        //Check that everything is as expected
207        assert( this->monitors );
208        assert( this->monitor_count != 0 );
209
210        unsigned short count = this->monitor_count;
211       
212        LIB_DEBUG_DO(
213                thread_desc * this_thrd = this_thread();
214                if ( this->monitor_count != this_thrd->current_monitor_count ) {
215                        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 );
216                } // if
217
218                for(int i = 0; i < this->monitor_count; i++) {
219                        if ( this->monitors[i] != this_thrd->current_monitors[i] ) {
220                                abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->current_monitors[i] );
221                        } // if
222                }
223        );
224
225        lock_all( this->monitors, NULL, count );
226        LIB_DEBUG_PRINT_SAFE("Signalling");
227
228        __condition_node_t * node = pop_head( &this->blocked );
229        for(int i = 0; i < count; i++) {
230                __condition_criterion_t * crit = &node->criteria[i];
231                LIB_DEBUG_PRINT_SAFE(" %p", crit->target);
232                assert( !crit->ready );
233                push( &crit->target->signal_stack, crit );
234        }
235
236        LIB_DEBUG_PRINT_SAFE("\n");
237
238        unlock_all( this->monitors, count );
239}
240
241//-----------------------------------------------------------------------------
242// Utilities
243
244static inline void set_owner( monitor_desc * this, thread_desc * owner ) {
245        //Pass the monitor appropriately
246        this->owner = owner;
247
248        //We are passing the monitor to someone else, which means recursion level is not 0
249        this->recursion = owner ? 1 : 0;
250}
251
252static inline thread_desc * next_thread( monitor_desc * this ) {
253        //Check the signaller stack
254        __condition_criterion_t * urgent = pop( &this->signal_stack );
255        if( urgent ) {
256                //The signaller stack is not empty,
257                //regardless of if we are ready to baton pass,
258                //we need to set the monitor as in use
259                set_owner( this,  urgent->owner->waiting_thread );
260
261                return check_condition( urgent );
262        }
263
264        // No signaller thread
265        // Get the next thread in the entry_queue
266        thread_desc * new_owner = pop_head( &this->entry_queue );
267        set_owner( this, new_owner );
268
269        return new_owner;
270}
271
272static inline void lock_all( spinlock ** locks, unsigned short count ) {
273        for( int i = 0; i < count; i++ ) {
274                lock( locks[i] );
275        }
276}
277
278static inline void lock_all( monitor_desc ** source, spinlock ** /*out*/ locks, unsigned short count ) {
279        for( int i = 0; i < count; i++ ) {
280                spinlock * l = &source[i]->lock;
281                lock( l );
282                if(locks) locks[i] = l;
283        }
284}
285
286static inline void unlock_all( spinlock ** locks, unsigned short count ) {
287        for( int i = 0; i < count; i++ ) {
288                unlock( locks[i] );
289        }
290}
291
292static inline void unlock_all( monitor_desc ** locks, unsigned short count ) {
293        for( int i = 0; i < count; i++ ) {
294                unlock( &locks[i]->lock );
295        }
296}
297
298
299static inline void save_recursion   ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count ) {
300        for( int i = 0; i < count; i++ ) {
301                recursions[i] = ctx[i]->recursion;
302        }
303}
304
305static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count ) {
306        for( int i = 0; i < count; i++ ) {
307                ctx[i]->recursion = recursions[i];
308        }
309}
310
311// Function has 2 different behavior
312// 1 - Marks a monitors as being ready to run
313// 2 - Checks if all the monitors are ready to run
314//     if so return the thread to run
315static inline thread_desc * check_condition( __condition_criterion_t * target ) {
316        __condition_node_t * node = target->owner;
317        unsigned short count = node->count;
318        __condition_criterion_t * criteria = node->criteria;
319
320        bool ready2run = true;
321
322        for(    int i = 0; i < count; i++ ) {
323                LIB_DEBUG_PRINT_SAFE( "Checking %p for %p\n", &criteria[i], target );
324                if( &criteria[i] == target ) {
325                        criteria[i].ready = true;
326                        LIB_DEBUG_PRINT_SAFE( "True\n" );
327                }
328
329                ready2run = criteria[i].ready && ready2run;
330        }
331
332        LIB_DEBUG_PRINT_SAFE( "Runing %i\n", ready2run );
333        return ready2run ? node->waiting_thread : NULL;
334}
335
336static inline void brand_condition( condition * this ) {
337        thread_desc * thrd = this_thread();
338        if( !this->monitors ) {
339                LIB_DEBUG_PRINT_SAFE("Branding\n");
340                assertf( thrd->current_monitors != NULL, "No current monitor to brand condition", thrd->current_monitors );
341                this->monitors = thrd->current_monitors;
342                this->monitor_count = thrd->current_monitor_count;
343        }
344}
345
346static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val ) {
347        for(int i = 0; i < end; i++) {
348                if( thrds[i] == val ) return end;
349        }
350
351        thrds[end] = val;
352        return end + 1;
353}
354
355void ?{}( __condition_blocked_queue_t * this ) {
356        this->head = NULL;
357        this->tail = &this->head;
358}
359
360void append( __condition_blocked_queue_t * this, __condition_node_t * c ) {
361        assert(this->tail != NULL);
362        *this->tail = c;
363        this->tail = &c->next;
364}
365
366__condition_node_t * pop_head( __condition_blocked_queue_t * this ) {
367        __condition_node_t * head = this->head;
368        if( head ) {
369                this->head = head->next;
370                if( !head->next ) {
371                        this->tail = &this->head;
372                }
373                head->next = NULL;
374        }
375        return head;
376}
Note: See TracBrowser for help on using the repository browser.