Changes in / [4be3669:8731d8c]
- Files:
-
- 1 added
- 7 edited
Legend:
- Unmodified
- Added
- Removed
-
doc/proposals/concurrency/thePlan.md
r4be3669 r8731d8c 10 10 done - Multi monitors calls, 11 11 done - Monitors as a language feature (not calling enter/leave by hand) 12 13 _Phase 3_ : Monitor features 12 14 Internal scheduling 15 External scheduling 13 16 14 _Phase 3_ : Kernel features17 _Phase 4_ : Kernel features 15 18 Preemption 16 19 Detach thread 17 20 Cluster migration 18 19 _Phase 4_ : Monitor features20 External scheduling21 21 22 22 _Phase 5_ : Performance -
src/libcfa/concurrency/invoke.h
r4be3669 r8731d8c 33 33 }; 34 34 35 struct simple_thread_list {35 struct __thread_queue_t { 36 36 struct thread_desc * head; 37 37 struct thread_desc ** tail; 38 38 }; 39 39 40 struct __thread_stack_t { 41 struct thread_desc * top; 42 }; 43 40 44 #ifdef __CFORALL__ 41 45 extern "Cforall" { 42 void ?{}( struct simple_thread_list * ); 43 void append( struct simple_thread_list *, struct thread_desc * ); 44 struct thread_desc * pop_head( struct simple_thread_list * ); 46 void ?{}( struct __thread_queue_t * ); 47 void append( struct __thread_queue_t *, struct thread_desc * ); 48 struct thread_desc * pop_head( struct __thread_queue_t * ); 49 50 void ?{}( struct __thread_stack_t * ); 51 void push( struct __thread_stack_t *, struct thread_desc * ); 52 struct thread_desc * pop( struct __thread_stack_t * ); 45 53 46 54 void ?{}(spinlock * this); … … 50 58 51 59 struct coStack_t { 52 unsigned int size; // size of stack53 void *storage; // pointer to stack54 void *limit; // stack grows towards stack limit55 void *base; // base of stack56 void *context; // address of cfa_context_t57 void *top; // address of top of storage58 bool userStack; // whether or not the user allocated the stack60 unsigned int size; // size of stack 61 void *storage; // pointer to stack 62 void *limit; // stack grows towards stack limit 63 void *base; // base of stack 64 void *context; // address of cfa_context_t 65 void *top; // address of top of storage 66 bool userStack; // whether or not the user allocated the stack 59 67 }; 60 68 … … 62 70 63 71 struct coroutine_desc { 64 struct coStack_t stack; // stack information of the coroutine65 const char *name; // textual name for coroutine/task, initialized by uC++ generated code66 int errno_; // copy of global UNIX variable errno67 enum coroutine_state state; // current execution status for coroutine68 struct coroutine_desc *starter; // first coroutine to resume this one69 struct coroutine_desc *last; // last coroutine to resume this one72 struct coStack_t stack; // stack information of the coroutine 73 const char *name; // textual name for coroutine/task, initialized by uC++ generated code 74 int errno_; // copy of global UNIX variable errno 75 enum coroutine_state state; // current execution status for coroutine 76 struct coroutine_desc *starter; // first coroutine to resume this one 77 struct coroutine_desc *last; // last coroutine to resume this one 70 78 }; 71 79 72 80 struct monitor_desc { 73 struct spinlock lock; 74 struct thread_desc * owner; 75 struct simple_thread_list entry_queue; 76 unsigned int recursion; 81 struct spinlock lock; // spinlock to protect internal data 82 struct thread_desc * owner; // current owner of the monitor 83 struct __thread_queue_t entry_queue; // queue of threads that are blocked waiting for the monitor 84 struct __thread_stack_t signal_stack; // stack of threads to run next once we exit the monitor 85 struct monitor_desc * stack_owner; // if bulk acquiring was used we need to synchronize signals with an other monitor 86 unsigned int recursion; // monitor routines can be called recursively, we need to keep track of that 77 87 }; 78 88 79 89 struct thread_desc { 80 struct coroutine_desc cor; // coroutine body used to store context 81 struct monitor_desc mon; // monitor body used for mutual exclusion 82 struct thread_desc * next; // instrusive link field for threads 90 struct coroutine_desc cor; // coroutine body used to store context 91 struct monitor_desc mon; // monitor body used for mutual exclusion 92 struct thread_desc * next; // instrusive link field for threads 93 struct monitor_desc ** current_monitors; // currently held monitors 94 unsigned short current_monitor_count; // number of currently held monitors 83 95 }; 84 96 -
src/libcfa/concurrency/kernel
r4be3669 r8731d8c 32 32 33 33 struct signal_once { 34 volatile bool cond ition;34 volatile bool cond; 35 35 struct spinlock lock; 36 struct simple_thread_list blocked;36 struct __thread_queue_t blocked; 37 37 }; 38 38 … … 46 46 // Cluster 47 47 struct cluster { 48 simple_thread_list ready_queue;48 __thread_queue_t ready_queue; 49 49 spinlock lock; 50 50 }; -
src/libcfa/concurrency/kernel.c
r4be3669 r8731d8c 299 299 // Scheduler routines 300 300 void ScheduleThread( thread_desc * thrd ) { 301 if( !thrd ) return; 302 301 303 assertf( thrd->next == NULL, "Expected null got %p", thrd->next ); 302 304 … … 473 475 474 476 void ?{}( signal_once * this ) { 475 this->cond ition= false;477 this->cond = false; 476 478 } 477 479 void ^?{}( signal_once * this ) { … … 481 483 void wait( signal_once * this ) { 482 484 lock( &this->lock ); 483 if( !this->cond ition) {485 if( !this->cond ) { 484 486 append( &this->blocked, this_thread() ); 485 487 ScheduleInternal( &this->lock ); … … 492 494 lock( &this->lock ); 493 495 { 494 this->cond ition= true;496 this->cond = true; 495 497 496 498 thread_desc * it; … … 504 506 //----------------------------------------------------------------------------- 505 507 // Queues 506 void ?{}( simple_thread_list * this ) {508 void ?{}( __thread_queue_t * this ) { 507 509 this->head = NULL; 508 510 this->tail = &this->head; 509 511 } 510 512 511 void append( simple_thread_list * this, thread_desc * t ) {513 void append( __thread_queue_t * this, thread_desc * t ) { 512 514 assert(this->tail != NULL); 513 515 *this->tail = t; … … 515 517 } 516 518 517 thread_desc * pop_head( simple_thread_list * this ) {519 thread_desc * pop_head( __thread_queue_t * this ) { 518 520 thread_desc * head = this->head; 519 521 if( head ) { … … 526 528 return head; 527 529 } 530 531 void ?{}( __thread_stack_t * this ) { 532 this->top = NULL; 533 } 534 535 void push( __thread_stack_t * this, thread_desc * t ) { 536 assert(t->next != NULL); 537 t->next = this->top; 538 this->top = t; 539 } 540 541 thread_desc * pop( __thread_stack_t * this ) { 542 thread_desc * top = this->top; 543 if( top ) { 544 this->top = top->next; 545 top->next = NULL; 546 } 547 return top; 548 } 528 549 // Local Variables: // 529 550 // mode: c // -
src/libcfa/concurrency/monitor
r4be3669 r8731d8c 18 18 #define MONITOR_H 19 19 20 #include <stddef.h> 21 20 22 #include "assert" 21 23 #include "invoke.h" … … 23 25 24 26 static inline void ?{}(monitor_desc * this) { 25 this->owner = 0; 27 this->owner = NULL; 28 this->stack_owner = NULL; 26 29 this->recursion = 0; 27 30 } 28 29 //Array entering routine30 void enter(monitor_desc **, int count);31 void leave(monitor_desc **, int count);32 31 33 32 struct monitor_guard_t { 34 33 monitor_desc ** m; 35 34 int count; 35 monitor_desc ** prev_mntrs; 36 unsigned short prev_count; 36 37 }; 37 38 … … 40 41 } 41 42 42 static inline void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ) { 43 this->m = m; 44 this->count = count; 45 qsort(this->m, count); 46 enter( this->m, this->count ); 43 void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ); 44 void ^?{}( monitor_guard_t * this ); 45 46 //----------------------------------------------------------------------------- 47 // Internal scheduling 48 struct condition { 49 __thread_queue_t blocked; 50 monitor_desc ** monitors; 51 unsigned short monitor_count; 52 }; 53 54 static inline void ?{}( condition * this ) { 55 this->monitors = NULL; 56 this->monitor_count = 0; 47 57 } 48 58 49 static inline void ^?{}( monitor_guard_t * this ) { 50 leave( this->m, this->count ); 51 } 52 53 59 void wait( condition * this ); 60 void signal( condition * this ); 54 61 #endif //MONITOR_H -
src/libcfa/concurrency/monitor.c
r4be3669 r8731d8c 18 18 19 19 #include "kernel_private.h" 20 #include "libhdr.h" 21 22 void set_owner( monitor_desc * this, thread_desc * owner ) { 23 //Pass the monitor appropriately 24 this->owner = owner; 25 26 //We are passing the monitor to someone else, which means recursion level is not 0 27 this->recursion = owner ? 1 : 0; 28 } 20 29 21 30 extern "C" { 22 void __enter_monitor_desc(monitor_desc * this ) {31 void __enter_monitor_desc(monitor_desc * this, monitor_desc * leader) { 23 32 lock( &this->lock ); 24 33 thread_desc * thrd = this_thread(); 25 34 35 // //Update the stack owner 36 // this->stack_owner = leader; 37 38 LIB_DEBUG_PRINT_SAFE("Entering %p (o: %p, r: %i)\n", this, this->owner, this->recursion); 39 26 40 if( !this->owner ) { 27 41 //No one has the monitor, just take it 28 this->owner = thrd; 29 this->recursion = 1; 42 set_owner( this, thrd ); 30 43 } 31 44 else if( this->owner == thrd) { … … 44 57 45 58 unlock( &this->lock ); 46 } 47 48 void __leave_monitor_desc(monitor_desc * this) { 59 return; 60 } 61 62 // leave pseudo code : 63 // decrement level 64 // leve == 0 ? 65 // no : done 66 // yes : 67 // signal stack empty ? 68 // has leader : 69 // bulk acquiring means we don't own the signal stack 70 // ignore it but don't release the monitor 71 // yes : 72 // next in entry queue is new owner 73 // no : 74 // top of the signal stack is the owner 75 // context switch to him right away 76 // 77 void __leave_monitor_desc(monitor_desc * this, monitor_desc * leader) { 49 78 lock( &this->lock ); 50 79 80 LIB_DEBUG_PRINT_SAFE("Leaving %p (o: %p, r: %i)\n", this, this->owner, this->recursion); 81 51 82 thread_desc * thrd = this_thread(); 52 assert ( thrd == this->owner);83 assertf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i)", this->owner, thrd, this->recursion ); 53 84 54 85 //Leaving a recursion level, decrement the counter 55 86 this->recursion -= 1; 56 87 57 //If we left the last level of recursion it means we are changing who owns the monitor 88 //If we haven't left the last level of recursion 89 //it means we don't need to do anything 90 if( this->recursion != 0) { 91 // this->stack_owner = leader; 92 unlock( &this->lock ); 93 return; 94 } 95 96 // //If we don't own the signal stack then just leave it to the owner 97 // if( this->stack_owner ) { 98 // this->stack_owner = leader; 99 // unlock( &this->lock ); 100 // return; 101 // } 102 103 //We are the stack owner and have left the last recursion level. 104 //We are in charge of passing the monitor 58 105 thread_desc * new_owner = 0; 59 if( this->recursion == 0) { 60 //Get the next thread in the list 61 new_owner = this->owner = pop_head( &this->entry_queue ); 62 63 //We are passing the monitor to someone else, which means recursion level is not 0 64 this->recursion = new_owner ? 1 : 0; 65 } 66 106 107 //Check the signaller stack 108 new_owner = pop( &this->signal_stack ); 109 if( new_owner ) { 110 //The signaller stack is not empty, 111 //transfer control immediately 112 set_owner( this, new_owner ); 113 // this->stack_owner = leader; 114 ScheduleInternal( &this->lock, new_owner ); 115 return; 116 } 117 118 // No signaller thread 119 // Get the next thread in the entry_queue 120 new_owner = pop_head( &this->entry_queue ); 121 set_owner( this, new_owner ); 122 123 // //Update the stack owner 124 // this->stack_owner = leader; 125 126 //We can now let other threads in safely 67 127 unlock( &this->lock ); 68 128 69 //If we have a new owner, we need to wake-up the thread 70 if( new_owner ) { 71 ScheduleThread( new_owner ); 72 } 73 } 74 } 75 76 void enter(monitor_desc ** monitors, int count) { 77 for(int i = 0; i < count; i++) { 78 __enter_monitor_desc( monitors[i] ); 79 } 80 } 81 82 void leave(monitor_desc ** monitors, int count) { 83 for(int i = count - 1; i >= 0; i--) { 84 __leave_monitor_desc( monitors[i] ); 85 } 86 } 129 //We need to wake-up the thread 130 ScheduleThread( new_owner ); 131 } 132 } 133 134 static inline void enter(monitor_desc ** monitors, int count) { 135 __enter_monitor_desc( monitors[0], NULL ); 136 for(int i = 1; i < count; i++) { 137 __enter_monitor_desc( monitors[i], monitors[0] ); 138 } 139 } 140 141 static inline void leave(monitor_desc ** monitors, int count) { 142 __leave_monitor_desc( monitors[0], NULL ); 143 for(int i = count - 1; i >= 1; i--) { 144 __leave_monitor_desc( monitors[i], monitors[0] ); 145 } 146 } 147 148 void ?{}( monitor_guard_t * this, monitor_desc ** m, int count ) { 149 this->m = m; 150 this->count = count; 151 qsort(this->m, count); 152 enter( this->m, this->count ); 153 154 this->prev_mntrs = this_thread()->current_monitors; 155 this->prev_count = this_thread()->current_monitor_count; 156 157 this_thread()->current_monitors = m; 158 this_thread()->current_monitor_count = count; 159 } 160 161 void ^?{}( monitor_guard_t * this ) { 162 leave( this->m, this->count ); 163 164 this_thread()->current_monitors = this->prev_mntrs; 165 this_thread()->current_monitor_count = this->prev_count; 166 } 167 168 //----------------------------------------------------------------------------- 169 // Internal scheduling 170 void wait( condition * this ) { 171 assertf(false, "NO SUPPORTED"); 172 // LIB_DEBUG_FPRINTF("Waiting\n"); 173 thread_desc * this_thrd = this_thread(); 174 175 if( !this->monitors ) { 176 this->monitors = this_thrd->current_monitors; 177 this->monitor_count = this_thrd->current_monitor_count; 178 } 179 180 unsigned short count = this->monitor_count; 181 182 //Check that everything is as expected 183 assert( this->monitors != NULL ); 184 assert( this->monitor_count != 0 ); 185 186 unsigned int recursions[ count ]; //Save the current recursion levels to restore them later 187 spinlock * locks [ count ]; //We need to pass-in an array of locks to ScheduleInternal 188 189 // LIB_DEBUG_FPRINTF("Getting ready to wait\n"); 190 191 //Loop on all the monitors and release the owner 192 for( unsigned int i = 0; i < count; i++ ) { 193 monitor_desc * cur = this->monitors[i]; 194 195 assert( cur ); 196 197 // LIB_DEBUG_FPRINTF("cur %p lock %p\n", cur, &cur->lock); 198 199 //Store the locks for later 200 locks[i] = &cur->lock; 201 202 //Protect the monitors 203 lock( locks[i] ); 204 { 205 //Save the recursion levels 206 recursions[i] = cur->recursion; 207 208 //Release the owner 209 cur->recursion = 0; 210 cur->owner = NULL; 211 } 212 //Release the monitor 213 unlock( locks[i] ); 214 } 215 216 // LIB_DEBUG_FPRINTF("Waiting now\n"); 217 218 //Everything is ready to go to sleep 219 ScheduleInternal( locks, count ); 220 221 222 //WE WOKE UP 223 224 225 //We are back, restore the owners and recursions 226 for( unsigned int i = 0; i < count; i++ ) { 227 monitor_desc * cur = this->monitors[i]; 228 229 //Protect the monitors 230 lock( locks[i] ); 231 { 232 //Release the owner 233 cur->owner = this_thrd; 234 cur->recursion = recursions[i]; 235 } 236 //Release the monitor 237 unlock( locks[i] ); 238 } 239 } 240 241 static void __signal_internal( condition * this ) { 242 assertf(false, "NO SUPPORTED"); 243 if( !this->blocked.head ) return; 244 245 //Check that everything is as expected 246 assert( this->monitors ); 247 assert( this->monitor_count != 0 ); 248 249 LIB_DEBUG_DO( 250 if ( this->monitors != this_thread()->current_monitors ) { 251 abortf( "Signal on condition %p made outside of the correct monitor(s)", this ); 252 } // if 253 ); 254 255 monitor_desc * owner = this->monitors[0]; 256 lock( &owner->lock ); 257 { 258 thread_desc * unblock = pop_head( &this->blocked ); 259 push( &owner->signal_stack, unblock ); 260 } 261 unlock( &owner->lock ); 262 } 263 264 void signal( condition * this ) { 265 __signal_internal( this ); 266 } -
src/libcfa/concurrency/thread.c
r4be3669 r8731d8c 39 39 this->mon.recursion = 1; 40 40 this->next = NULL; 41 42 this->current_monitors = NULL; 43 this->current_monitor_count = 0; 41 44 } 42 45
Note: See TracChangeset
for help on using the changeset viewer.