Index: src/libcfa/concurrency/invoke.h =================================================================== --- src/libcfa/concurrency/invoke.h (revision 97e32964dc1e5125b35bd5db0234e6b6f272fe14) +++ src/libcfa/concurrency/invoke.h (revision 90c4df010b647b1dd75cbb19ce37bdaa11617886) @@ -28,4 +28,6 @@ #define thread_local _Thread_local + typedef void (*fptr_t)(); + struct spinlock { volatile int lock; @@ -50,4 +52,5 @@ void append( struct __thread_queue_t *, struct thread_desc * ); struct thread_desc * pop_head( struct __thread_queue_t * ); + struct thread_desc * remove( struct __thread_queue_t *, struct thread_desc ** ); void ?{}( struct __condition_stack_t * ); @@ -91,13 +94,19 @@ unsigned short acceptable_count; // number of acceptable functions short accepted_index; // the index of the accepted function, -1 if none - void (*pre_accept)(void); // function to run before an accept + fptr_t pre_accept; // function to run before an accept }; struct thread_desc { + // Core threading fields struct coroutine_desc cor; // coroutine body used to store context struct monitor_desc mon; // monitor body used for mutual exclusion + + // Link lists fields struct thread_desc * next; // instrusive link field for threads + + // Current status related to monitors struct monitor_desc ** current_monitors; // currently held monitors unsigned short current_monitor_count; // number of currently held monitors + fptr_t current_monitor_func; // last function that acquired monitors }; Index: src/libcfa/concurrency/kernel.c =================================================================== --- src/libcfa/concurrency/kernel.c (revision 97e32964dc1e5125b35bd5db0234e6b6f272fe14) +++ src/libcfa/concurrency/kernel.c (revision 90c4df010b647b1dd75cbb19ce37bdaa11617886) @@ -668,4 +668,23 @@ } +thread_desc * remove( __thread_queue_t * this, thread_desc ** it ) { + thread_desc * thrd = *it; + verify( thrd ); + + (*it) = thrd->next; + + if( this->tail == &thrd->next ) { + this->tail = it; + } + + thrd->next = NULL; + + verify( (this->head == NULL) == (&this->head == this->tail) ); + verify( *this->tail == NULL ); + return thrd; +} + + + void ?{}( __condition_stack_t * this ) { this->top = NULL; Index: src/libcfa/concurrency/monitor =================================================================== --- src/libcfa/concurrency/monitor (revision 97e32964dc1e5125b35bd5db0234e6b6f272fe14) +++ src/libcfa/concurrency/monitor (revision 90c4df010b647b1dd75cbb19ce37bdaa11617886) @@ -39,4 +39,5 @@ monitor_desc ** prev_mntrs; unsigned short prev_count; + fptr_t prev_func; }; @@ -99,8 +100,6 @@ // External scheduling -typedef void (*void_fptr_t)(void); - struct __acceptable_t { - void_fptr_t func; + fptr_t func; unsigned short count; monitor_desc ** monitors; Index: src/libcfa/concurrency/monitor.c =================================================================== --- src/libcfa/concurrency/monitor.c (revision 97e32964dc1e5125b35bd5db0234e6b6f272fe14) +++ src/libcfa/concurrency/monitor.c (revision 90c4df010b647b1dd75cbb19ce37bdaa11617886) @@ -73,8 +73,12 @@ thread_desc * thrd = this_thread; + LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner); + this->accepted_index = -1; if( !this->owner ) { // No one has the monitor, just take it set_owner( this, thrd ); + + LIB_DEBUG_PRINT_SAFE("Kernel : mon is free \n"); } else if( this->owner == thrd) { @@ -82,17 +86,27 @@ verify( this->recursion > 0 ); this->recursion += 1; + + LIB_DEBUG_PRINT_SAFE("Kernel : mon already owned \n"); } else if( (this->accepted_index = is_accepted( thrd, this, group, group_cnt, func)) >= 0 ) { // Some one was waiting for us, enter set_owner( this, thrd ); + + LIB_DEBUG_PRINT_SAFE("Kernel : mon accepts \n"); } else { + LIB_DEBUG_PRINT_SAFE("Kernel : blocking \n"); + // Some one else has the monitor, wait in line for it append( &this->entry_queue, thrd ); BlockInternal( &this->lock ); + LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered mon %p\n", thrd, this); + // BlockInternal will unlock spinlock, no need to unlock ourselves return; } + + LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered mon %p\n", thrd, this); // Release the lock and leave @@ -194,14 +208,16 @@ qsort(this->m, count); - // Enter the monitors in order - enter( this->m, this->count, func ); - // Save previous thread context this->prev_mntrs = this_thread->current_monitors; this->prev_count = this_thread->current_monitor_count; + this->prev_func = this_thread->current_monitor_func; // Update thread context (needed for conditions) this_thread->current_monitors = m; this_thread->current_monitor_count = count; + this_thread->current_monitor_func = func; + + // Enter the monitors in order + enter( this->m, this->count, func ); } @@ -214,4 +230,5 @@ this_thread->current_monitors = this->prev_mntrs; this_thread->current_monitor_count = this->prev_count; + this_thread->current_monitor_func = this->prev_func; } @@ -402,4 +419,12 @@ thread_desc * next = search_entry_queue( acceptables, acc_count, monitors, count ); + LIB_DEBUG_PRINT_SAFE("Owner(s) :"); + for(int i = 0; i < count; i++) { + LIB_DEBUG_PRINT_SAFE(" %p", monitors[i]->owner ); + } + LIB_DEBUG_PRINT_SAFE("\n"); + + LIB_DEBUG_PRINT_SAFE("Passing mon to %p\n", next); + if( !next ) { // Update acceptables on the current monitors @@ -409,6 +434,13 @@ } } + else { + for(int i = 0; i < count; i++) { + set_owner( monitors[i], next ); + } + } + save_recursion( monitors, recursions, count ); + // Everything is ready to go to sleep @@ -602,5 +634,34 @@ } +static inline bool match( __acceptable_t * acc, thread_desc * thrd ) { + verify( thrd ); + verify( acc ); + if( acc->func != thrd->current_monitor_func ) return false; + + return true; +} + static inline thread_desc * search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count ) { + + __thread_queue_t * entry_queue = &monitors[0]->entry_queue; + + // For each thread in the entry-queue + for( thread_desc ** thrd_it = &entry_queue->head; + *thrd_it; + thrd_it = &(*thrd_it)->next) + { + // For each acceptable check if it matches + __acceptable_t * acc_end = acceptables + acc_count; + for( __acceptable_t * acc_it = acceptables; acc_it != acc_end; acc_it++ ) { + // Check if we have a match + if( match( acc_it, *thrd_it ) ) { + + // If we have a match return it + // after removeing it from the entry queue + return remove( entry_queue, thrd_it ); + } + } + } + return NULL; } Index: src/libcfa/concurrency/preemption.c =================================================================== --- src/libcfa/concurrency/preemption.c (revision 97e32964dc1e5125b35bd5db0234e6b6f272fe14) +++ src/libcfa/concurrency/preemption.c (revision 90c4df010b647b1dd75cbb19ce37bdaa11617886) @@ -332,5 +332,5 @@ assertf(sig == SIGALRM, "Kernel Internal Error, sigwait: Unexpected signal %d (%d : %d)\n", sig, info.si_code, info.si_value.sival_int); - LIB_DEBUG_PRINT_SAFE("Kernel : Caught alarm from %d with %d\n", info.si_code, info.si_value.sival_int ); + // LIB_DEBUG_PRINT_SAFE("Kernel : Caught alarm from %d with %d\n", info.si_code, info.si_value.sival_int ); // Switch on the code (a.k.a. the sender) to switch( info.si_code ) @@ -340,5 +340,5 @@ case SI_TIMER: case SI_KERNEL: - LIB_DEBUG_PRINT_SAFE("Kernel : Preemption thread tick\n"); + // LIB_DEBUG_PRINT_SAFE("Kernel : Preemption thread tick\n"); lock( &event_kernel->lock DEBUG_CTX2 ); tick_preemption();