Index: src/libcfa/Makefile.am =================================================================== --- src/libcfa/Makefile.am (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/Makefile.am (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -36,5 +36,5 @@ ${AM_V_GEN}@BACKEND_CC@ @CFA_FLAGS@ -D__CFA_DEBUG__ -O0 -c -o $@ $< -EXTRA_FLAGS = -g -Wall -Werror -Wno-unused-function -imacros libcfa-prelude.c @CFA_FLAGS@ +EXTRA_FLAGS = -g -Wall -Wno-unused-function -imacros libcfa-prelude.c @CFA_FLAGS@ AM_CCASFLAGS = @CFA_FLAGS@ Index: src/libcfa/Makefile.in =================================================================== --- src/libcfa/Makefile.in (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/Makefile.in (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -416,5 +416,5 @@ ARFLAGS = cr lib_LIBRARIES = $(am__append_1) $(am__append_2) -EXTRA_FLAGS = -g -Wall -Werror -Wno-unused-function -imacros libcfa-prelude.c @CFA_FLAGS@ +EXTRA_FLAGS = -g -Wall -Wno-unused-function -imacros libcfa-prelude.c @CFA_FLAGS@ AM_CCASFLAGS = @CFA_FLAGS@ headers = fstream iostream iterator limits rational stdlib \ Index: src/libcfa/concurrency/coroutine.c =================================================================== --- src/libcfa/concurrency/coroutine.c (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/coroutine.c (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -123,6 +123,4 @@ if(pageSize == 0ul) pageSize = sysconf( _SC_PAGESIZE ); - LIB_DEBUG_PRINT_SAFE("FRED"); - size_t cxtSize = libCeiling( sizeof(machine_context_t), 8 ); // minimum alignment Index: src/libcfa/concurrency/invoke.h =================================================================== --- src/libcfa/concurrency/invoke.h (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/invoke.h (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -84,4 +84,10 @@ }; + struct __waitfor_mask_t { + short * accepted; // the index of the accepted function, -1 if none + struct __acceptable_t * clauses; // list of acceptable functions, null if any + short size; // number of acceptable functions + }; + struct monitor_desc { struct spinlock lock; // spinlock to protect internal data @@ -90,23 +96,47 @@ struct __condition_stack_t signal_stack; // stack of conditions to run next once we exit the monitor unsigned int recursion; // monitor routines can be called recursively, we need to keep track of that + struct __waitfor_mask_t mask; // mask used to know if some thread is waiting for something while holding the monitor + }; - struct __acceptable_t * acceptables; // list of acceptable functions, null if any - unsigned short acceptable_count; // number of acceptable functions - short accepted_index; // the index of the accepted function, -1 if none - }; + struct __monitor_group_t { + struct monitor_desc ** list; // currently held monitors + short size; // number of currently held monitors + fptr_t func; // last function that acquired monitors + }; 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 + struct coroutine_desc self_cor; // coroutine body used to store context + struct monitor_desc self_mon; // monitor body used for mutual exclusion + struct monitor_desc * self_mon_p; // pointer to monitor with sufficient lifetime for current monitors + struct __monitor_group_t monitors; // monitors currently held by this thread // 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 + }; + + #ifdef __CFORALL__ + extern "Cforall" { + static inline monitor_desc * ?[?]( const __monitor_group_t & this, ptrdiff_t index ) { + return this.list[index]; + } + + static inline bool ?==?( const __monitor_group_t & lhs, const __monitor_group_t & rhs ) { + if( (lhs.list != 0) != (rhs.list != 0) ) return false; + if( lhs.size != rhs.size ) return false; + if( lhs.func != rhs.func ) return false; + + // Check that all the monitors match + for( int i = 0; i < lhs.size; i++ ) { + // If not a match, check next function + if( lhs[i] != rhs[i] ) return false; + } + + return true; + } + } + #endif #endif //_INVOKE_H_ Index: src/libcfa/concurrency/kernel.c =================================================================== --- src/libcfa/concurrency/kernel.c (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/kernel.c (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -106,5 +106,5 @@ void ?{}( thread_desc & this, current_stack_info_t * info) { - (this.cor){ info }; + (this.self_cor){ info }; } @@ -328,5 +328,5 @@ // if( !thrd ) return; verify( thrd ); - verify( thrd->cor.state != Halted ); + verify( thrd->self_cor.state != Halted ); verify( disable_preempt_count > 0 ); @@ -373,5 +373,5 @@ assert(thrd); disable_interrupts(); - assert( thrd->cor.state != Halted ); + assert( thrd->self_cor.state != Halted ); this_processor->finish.action_code = Schedule; this_processor->finish.thrd = thrd; @@ -466,5 +466,5 @@ this_processor = mainProcessor; this_thread = mainThread; - this_coroutine = &mainThread->cor; + this_coroutine = &mainThread->self_cor; // Enable preemption @@ -547,5 +547,5 @@ thread_desc * thrd = kernel_data; - int len = snprintf( abort_text, abort_text_size, "Error occurred while executing task %.256s (%p)", thrd->cor.name, thrd ); + int len = snprintf( abort_text, abort_text_size, "Error occurred while executing task %.256s (%p)", thrd->self_cor.name, thrd ); __lib_debug_write( STDERR_FILENO, abort_text, len ); Index: src/libcfa/concurrency/monitor =================================================================== --- src/libcfa/concurrency/monitor (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/monitor (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -22,4 +22,9 @@ #include "stdlib" +trait is_monitor(dtype T) { + monitor_desc * get_monitor( T & ); + void ^?{}( T & mutex ); +}; + static inline void ?{}(monitor_desc & this) { (this.lock){}; @@ -28,7 +33,7 @@ (this.signal_stack){}; this.recursion = 0; - this.acceptables = NULL; - this.acceptable_count = 0; - this.accepted_index = -1; + this.mask.accepted = NULL; + this.mask.clauses = NULL; + this.mask.size = 0; } @@ -100,11 +105,9 @@ struct __acceptable_t { - fptr_t func; - unsigned short count; - monitor_desc ** monitors; + __monitor_group_t; bool is_dtor; }; -int __accept_internal( unsigned short count, __acceptable_t * acceptables ); +void __waitfor_internal( const __waitfor_mask_t & mask, int duration ); // Local Variables: // Index: src/libcfa/concurrency/monitor.c =================================================================== --- src/libcfa/concurrency/monitor.c (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/monitor.c (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -23,7 +23,11 @@ //----------------------------------------------------------------------------- // Forward declarations -static inline void set_owner( monitor_desc * this, thread_desc * owner ); +static inline void set_owner ( monitor_desc * this, thread_desc * owner ); +static inline void set_owner ( monitor_desc ** storage, short count, thread_desc * owner ); +static inline void set_mask ( monitor_desc ** storage, short count, const __waitfor_mask_t & mask ); +static inline void reset_mask( monitor_desc * this ); + static inline thread_desc * next_thread( monitor_desc * this ); -static inline int is_accepted( thread_desc * owner, monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() ); +static inline bool is_accepted( monitor_desc * this, const __monitor_group_t & monitors ); static inline void lock_all( spinlock ** locks, unsigned short count ); @@ -32,33 +36,41 @@ static inline void unlock_all( monitor_desc ** locks, unsigned short count ); -static inline void save_recursion ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count ); -static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count ); +static inline void save ( monitor_desc ** ctx, short count, spinlock ** locks, unsigned int * /*out*/ recursions, __waitfor_mask_t * /*out*/ masks ); +static inline void restore( monitor_desc ** ctx, short count, spinlock ** locks, unsigned int * /*in */ recursions, __waitfor_mask_t * /*in */ masks ); static inline void init ( int count, monitor_desc ** monitors, __condition_node_t * waiter, __condition_criterion_t * criteria ); static inline void init_push( int count, monitor_desc ** monitors, __condition_node_t * waiter, __condition_criterion_t * criteria ); -static inline thread_desc * check_condition( __condition_criterion_t * ); -static inline void brand_condition( condition * ); -static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val ); - -static inline thread_desc * search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count ); +static inline thread_desc * check_condition ( __condition_criterion_t * ); +static inline void brand_condition ( condition * ); +static inline [thread_desc *, int] search_entry_queue( const __waitfor_mask_t &, monitor_desc ** monitors, int count ); + +forall(dtype T | sized( T )) +static inline short insert_unique( T ** array, short & size, T * val ); +static inline short count_max ( const __waitfor_mask_t & mask ); +static inline short aggregate ( monitor_desc ** storage, const __waitfor_mask_t & mask ); //----------------------------------------------------------------------------- // Useful defines -#define wait_ctx(thrd, user_info) /* Create the necessary information to use the signaller stack */ \ - __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \ - __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \ - init( count, monitors, &waiter, criteria ); /* Link everything together */ \ - -#define wait_ctx_primed(thrd, user_info) /* Create the necessary information to use the signaller stack */ \ - __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \ - __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \ - init_push( count, monitors, &waiter, criteria ); /* Link everything together and push it to the AS-Stack */ \ - -#define monitor_ctx( mons, cnt ) /* Define that create the necessary struct for internal/external scheduling operations */ \ - monitor_desc ** monitors = mons; /* Save the targeted monitors */ \ - unsigned short count = cnt; /* Save the count to a local variable */ \ - unsigned int recursions[ count ]; /* Save the current recursion levels to restore them later */ \ - spinlock * locks [ count ]; /* We need to pass-in an array of locks to BlockInternal */ \ +#define wait_ctx(thrd, user_info) /* Create the necessary information to use the signaller stack */ \ + __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \ + __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \ + init( count, monitors, &waiter, criteria ); /* Link everything together */ \ + +#define wait_ctx_primed(thrd, user_info) /* Create the necessary information to use the signaller stack */ \ + __condition_node_t waiter = { thrd, count, user_info }; /* Create the node specific to this wait operation */ \ + __condition_criterion_t criteria[count]; /* Create the creteria this wait operation needs to wake up */ \ + init_push( count, monitors, &waiter, criteria ); /* Link everything together and push it to the AS-Stack */ \ + +#define monitor_ctx( mons, cnt ) /* Define that create the necessary struct for internal/external scheduling operations */ \ + monitor_desc ** monitors = mons; /* Save the targeted monitors */ \ + unsigned short count = cnt; /* Save the count to a local variable */ \ + unsigned int recursions[ count ]; /* Save the current recursion levels to restore them later */ \ + __waitfor_mask_t masks[ count ]; /* Save the current waitfor masks to restore them later */ \ + spinlock * locks [ count ]; /* We need to pass-in an array of locks to BlockInternal */ \ + +#define monitor_save save ( monitors, count, locks, recursions, masks ) +#define monitor_restore restore( monitors, count, locks, recursions, masks ) + //----------------------------------------------------------------------------- @@ -68,5 +80,5 @@ extern "C" { // Enter single monitor - static void __enter_monitor_desc( monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() ) { + static void __enter_monitor_desc( monitor_desc * this, const __monitor_group_t & group ) { // Lock the monitor spinlock, lock_yield to reduce contention lock_yield( &this->lock DEBUG_CTX2 ); @@ -75,5 +87,4 @@ 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 @@ -89,7 +100,10 @@ LIB_DEBUG_PRINT_SAFE("Kernel : mon already owned \n"); } - else if( (this->accepted_index = is_accepted( thrd, this, group, group_cnt, func)) >= 0 ) { + else if( is_accepted( this, group) ) { // Some one was waiting for us, enter set_owner( this, thrd ); + + // Reset mask + reset_mask( this ); LIB_DEBUG_PRINT_SAFE("Kernel : mon accepts \n"); @@ -120,5 +134,7 @@ lock_yield( &this->lock DEBUG_CTX2 ); - verifyf( this_thread == this->owner, "Expected owner to be %p, got %p (r: %i)", this_thread, this->owner, this->recursion ); + LIB_DEBUG_PRINT_SAFE("Kernel : %10p Leaving mon %p (%p)\n", this_thread, this, this->owner); + + verifyf( this_thread == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", this_thread, this->owner, this->recursion, this ); // Leaving a recursion level, decrement the counter @@ -146,5 +162,5 @@ // Should never return void __leave_thread_monitor( thread_desc * thrd ) { - monitor_desc * this = &thrd->mon; + monitor_desc * this = &thrd->self_mon; // Lock the monitor now @@ -153,7 +169,7 @@ disable_interrupts(); - thrd->cor.state = Halted; - - verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i)", thrd, this->owner, this->recursion ); + thrd->self_cor.state = Halted; + + verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", thrd, this->owner, this->recursion, this ); // Leaving a recursion level, decrement the counter @@ -178,7 +194,7 @@ // Enter multiple monitor // relies on the monitor array being sorted -static inline void enter(monitor_desc ** monitors, int count, void (*func)() ) { - for(int i = 0; i < count; i++) { - __enter_monitor_desc( monitors[i], monitors, count, func ); +static inline void enter( __monitor_group_t monitors ) { + for(int i = 0; i < monitors.size; i++) { + __enter_monitor_desc( monitors.list[i], monitors ); } } @@ -203,15 +219,20 @@ // 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; + this.prev_mntrs = this_thread->monitors.list; + this.prev_count = this_thread->monitors.size; + this.prev_func = this_thread->monitors.func; // Update thread context (needed for conditions) - this_thread->current_monitors = m; - this_thread->current_monitor_count = count; - this_thread->current_monitor_func = func; + this_thread->monitors.list = m; + this_thread->monitors.size = count; + this_thread->monitors.func = func; + + // LIB_DEBUG_PRINT_SAFE("MGUARD : enter %d\n", count); // Enter the monitors in order - enter( this.m, this.count, func ); + __monitor_group_t group = {this.m, this.count, func}; + enter( group ); + + // LIB_DEBUG_PRINT_SAFE("MGUARD : entered\n"); } @@ -219,11 +240,15 @@ // Dtor for monitor guard void ^?{}( monitor_guard_t & this ) { + // LIB_DEBUG_PRINT_SAFE("MGUARD : leaving %d\n", this.count); + // Leave the monitors in order leave( this.m, this.count ); + // LIB_DEBUG_PRINT_SAFE("MGUARD : left\n"); + // Restore thread context - this_thread->current_monitors = this.prev_mntrs; - this_thread->current_monitor_count = this.prev_count; - this_thread->current_monitor_func = this.prev_func; + this_thread->monitors.list = this.prev_mntrs; + this_thread->monitors.size = this.prev_count; + this_thread->monitors.func = this.prev_func; } @@ -271,14 +296,9 @@ append( &this->blocked, &waiter ); - // Lock all monitors (aggregates the lock them as well) + // Lock all monitors (aggregates the locks as well) lock_all( monitors, locks, count ); - // DON'T unlock, ask the kernel to do it - - // Save monitor state - save_recursion( monitors, recursions, count ); - // Find the next thread(s) to run - unsigned short thread_count = 0; + short thread_count = 0; thread_desc * threads[ count ]; for(int i = 0; i < count; i++) { @@ -286,8 +306,11 @@ } + // Save monitor states + monitor_save; + // Remove any duplicate threads for( int i = 0; i < count; i++) { thread_desc * new_owner = next_thread( monitors[i] ); - thread_count = insert_unique( threads, thread_count, new_owner ); + insert_unique( threads, thread_count, new_owner ); } @@ -295,12 +318,6 @@ BlockInternal( locks, count, threads, thread_count ); - - // WE WOKE UP - - // We are back, restore the owners and recursions - lock_all( locks, count ); - restore_recursion( monitors, recursions, count ); - unlock_all( locks, count ); + monitor_restore; } @@ -315,11 +332,11 @@ LIB_DEBUG_DO( thread_desc * this_thrd = this_thread; - if ( this->monitor_count != this_thrd->current_monitor_count ) { - 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 ); + if ( this->monitor_count != this_thrd->monitors.size ) { + abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", this, this->monitor_count, this_thrd->monitors.size ); } for(int i = 0; i < this->monitor_count; i++) { - if ( this->monitors[i] != this_thrd->current_monitors[i] ) { - abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->current_monitors[i] ); + if ( this->monitors[i] != this_thrd->monitors.list[i] ) { + abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->monitors.list[i] ); } } @@ -364,11 +381,9 @@ //save contexts - save_recursion( monitors, recursions, count ); + monitor_save; //Find the thread to run thread_desc * signallee = pop_head( &this->blocked )->waiting_thread; - for(int i = 0; i < count; i++) { - set_owner( monitors[i], signallee ); - } + set_owner( monitors, count, signallee ); //Everything is ready to go to sleep @@ -379,8 +394,6 @@ - //We are back, restore the owners and recursions - lock_all( locks, count ); - restore_recursion( monitors, recursions, count ); - unlock_all( locks, count ); + //We are back, restore the masks and recursions + monitor_restore; return true; @@ -397,59 +410,109 @@ //----------------------------------------------------------------------------- -// Internal scheduling -int __accept_internal( unsigned short acc_count, __acceptable_t * acceptables ) { - thread_desc * thrd = this_thread; +// External scheduling +// cases to handle : +// - target already there : +// block and wake +// - dtor already there +// put thread on signaller stack +// - non-blocking +// return else +// - timeout +// return timeout +// - block +// setup mask +// block +void __waitfor_internal( const __waitfor_mask_t & mask, int duration ) { + // This statment doesn't have a contiguous list of monitors... + // Create one! + short max = count_max( mask ); + monitor_desc * mon_storage[max]; + short actual_count = aggregate( mon_storage, mask ); + + if(actual_count == 0) return; + + LIB_DEBUG_PRINT_SAFE("Kernel : waitfor internal proceeding\n"); // Create storage for monitor context - monitor_ctx( acceptables->monitors, acceptables->count ); - - // Lock all monitors (aggregates the lock them as well) + monitor_ctx( mon_storage, actual_count ); + + // Lock all monitors (aggregates the locks as well) lock_all( monitors, locks, count ); + { + // Check if the entry queue + thread_desc * next; int index; + [next, index] = search_entry_queue( mask, monitors, count ); + + if( next ) { + *mask.accepted = index; + if( mask.clauses[index].is_dtor ) { + #warning case not implemented + } + else { + LIB_DEBUG_PRINT_SAFE("Kernel : thread present, baton-passing\n"); + + // Create the node specific to this wait operation + wait_ctx_primed( this_thread, 0 ); + + // Save monitor states + monitor_save; + + // Set the owners to be the next thread + set_owner( monitors, count, next ); + + // Everything is ready to go to sleep + BlockInternal( locks, count, &next, 1 ); + + // We are back, restore the owners and recursions + monitor_restore; + + LIB_DEBUG_PRINT_SAFE("Kernel : thread present, returned\n"); + } + + LIB_DEBUG_PRINT_SAFE("Kernel : accepted %d\n", *mask.accepted); + + return; + } + } + + + if( duration == 0 ) { + LIB_DEBUG_PRINT_SAFE("Kernel : non-blocking, exiting\n"); + + unlock_all( locks, count ); + + LIB_DEBUG_PRINT_SAFE("Kernel : accepted %d\n", *mask.accepted); + return; + } + + + verifyf( duration < 0, "Timeout on waitfor statments not supported yet."); + + LIB_DEBUG_PRINT_SAFE("Kernel : blocking waitfor\n"); + // Create the node specific to this wait operation - wait_ctx_primed( thrd, 0 ); - - // Check if the entry queue - thread_desc * next = search_entry_queue( acceptables, acc_count, monitors, count ); - - LIB_DEBUG_PRINT_SAFE("Owner(s) :"); + wait_ctx_primed( this_thread, 0 ); + + monitor_save; + set_mask( monitors, count, mask ); + 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 - for(int i = 0; i < count; i++) { - monitors[i]->acceptables = acceptables; - monitors[i]->acceptable_count = acc_count; - } - } - 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 - BlockInternal( locks, count, &next, next ? 1 : 0 ); - - - //WE WOKE UP - - - //We are back, restore the owners and recursions - lock_all( locks, count ); - restore_recursion( monitors, recursions, count ); - int acc_idx = monitors[0]->accepted_index; - unlock_all( locks, count ); - - return acc_idx; + verify( monitors[i]->owner == this_thread ); + } + + //Everything is ready to go to sleep + BlockInternal( locks, count ); + + + // WE WOKE UP + + + //We are back, restore the masks and recursions + monitor_restore; + + LIB_DEBUG_PRINT_SAFE("Kernel : exiting\n"); + + LIB_DEBUG_PRINT_SAFE("Kernel : accepted %d\n", *mask.accepted); } @@ -458,4 +521,6 @@ static inline void set_owner( monitor_desc * this, thread_desc * owner ) { + // LIB_DEBUG_PRINT_SAFE("Kernal : Setting owner of %p to %p ( was %p)\n", this, owner, this->owner ); + //Pass the monitor appropriately this->owner = owner; @@ -463,4 +528,22 @@ //We are passing the monitor to someone else, which means recursion level is not 0 this->recursion = owner ? 1 : 0; +} + +static inline void set_owner( monitor_desc ** monitors, short count, thread_desc * owner ) { + for( int i = 0; i < count; i++ ) { + set_owner( monitors[i], owner ); + } +} + +static inline void set_mask( monitor_desc ** storage, short count, const __waitfor_mask_t & mask ) { + for(int i = 0; i < count; i++) { + storage[i]->mask = mask; + } +} + +static inline void reset_mask( monitor_desc * this ) { + this->mask.accepted = NULL; + this->mask.clauses = NULL; + this->mask.size = 0; } @@ -485,38 +568,24 @@ } -static inline int is_accepted( thread_desc * owner, monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() ) { - __acceptable_t* accs = this->acceptables; // Optim - int acc_cnt = this->acceptable_count; +static inline bool is_accepted( monitor_desc * this, const __monitor_group_t & group ) { + __acceptable_t * it = this->mask.clauses; // Optim + int count = this->mask.size; // Check if there are any acceptable functions - if( !accs ) return -1; + if( !it ) return false; // If this isn't the first monitor to test this, there is no reason to repeat the test. - if( this != group[0] ) return group[0]->accepted_index; + if( this != group[0] ) return group[0]->mask.accepted >= 0; // For all acceptable functions check if this is the current function. - OUT_LOOP: - for( int i = 0; i < acc_cnt; i++ ) { - __acceptable_t * acc = &accs[i]; - - // if function matches, check the monitors - if( acc->func == func ) { - - // If the group count is different then it can't be a match - if( acc->count != group_cnt ) return -1; - - // Check that all the monitors match - for( int j = 0; j < group_cnt; j++ ) { - // If not a match, check next function - if( acc->monitors[j] != group[j] ) continue OUT_LOOP; - } - - // It's a complete match, accept the call - return i; + for( short i = 0; i < count; i++, it++ ) { + if( *it == group ) { + *this->mask.accepted = i; + return true; } } // No function matched - return -1; + return false; } @@ -564,15 +633,18 @@ } - -static inline void save_recursion ( monitor_desc ** ctx, unsigned int * /*out*/ recursions, unsigned short count ) { +static inline void save( monitor_desc ** ctx, short count, __attribute((unused)) spinlock ** locks, unsigned int * /*out*/ recursions, __waitfor_mask_t * /*out*/ masks ) { for( int i = 0; i < count; i++ ) { recursions[i] = ctx[i]->recursion; - } -} - -static inline void restore_recursion( monitor_desc ** ctx, unsigned int * /*in */ recursions, unsigned short count ) { + masks[i] = ctx[i]->mask; + } +} + +static inline void restore( monitor_desc ** ctx, short count, spinlock ** locks, unsigned int * /*out*/ recursions, __waitfor_mask_t * /*out*/ masks ) { + lock_all( locks, count ); for( int i = 0; i < count; i++ ) { ctx[i]->recursion = recursions[i]; - } + ctx[i]->mask = masks[i]; + } + unlock_all( locks, count ); } @@ -607,35 +679,15 @@ if( !this->monitors ) { // LIB_DEBUG_PRINT_SAFE("Branding\n"); - assertf( thrd->current_monitors != NULL, "No current monitor to brand condition %p", thrd->current_monitors ); - this->monitor_count = thrd->current_monitor_count; + assertf( thrd->monitors.list != NULL, "No current monitor to brand condition %p", thrd->monitors.list ); + this->monitor_count = thrd->monitors.size; this->monitors = malloc( this->monitor_count * sizeof( *this->monitors ) ); for( int i = 0; i < this->monitor_count; i++ ) { - this->monitors[i] = thrd->current_monitors[i]; - } - } -} - -static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val ) { - if( !val ) return end; - - for(int i = 0; i <= end; i++) { - if( thrds[i] == val ) return end; - } - - thrds[end] = val; - return end + 1; -} - - -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 ) { + this->monitors[i] = thrd->monitors.list[i]; + } + } +} + +static inline [thread_desc *, int] search_entry_queue( const __waitfor_mask_t & mask, monitor_desc ** monitors, int count ) { __thread_queue_t * entry_queue = &monitors[0]->entry_queue; @@ -644,21 +696,55 @@ for( thread_desc ** thrd_it = &entry_queue->head; *thrd_it; - thrd_it = &(*thrd_it)->next) - { + 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++ ) { + int i = 0; + __acceptable_t * end = mask.clauses + mask.size; + for( __acceptable_t * it = mask.clauses; it != end; it++, i++ ) { // Check if we have a match - if( match( acc_it, *thrd_it ) ) { + if( *it == (*thrd_it)->monitors ) { // If we have a match return it // after removeing it from the entry queue - return remove( entry_queue, thrd_it ); + return [remove( entry_queue, thrd_it ), i]; } } } - return NULL; -} + return [0, -1]; +} + +forall(dtype T | sized( T )) +static inline short insert_unique( T ** array, short & size, T * val ) { + if( !val ) return size; + + for(int i = 0; i <= size; i++) { + if( array[i] == val ) return size; + } + + array[size] = val; + size = size + 1; + return size; +} + +static inline short count_max( const __waitfor_mask_t & mask ) { + short max = 0; + for( int i = 0; i < mask.size; i++ ) { + max += mask.clauses[i].size; + } + return max; +} + +static inline short aggregate( monitor_desc ** storage, const __waitfor_mask_t & mask ) { + short size = 0; + for( int i = 0; i < mask.size; i++ ) { + for( int j = 0; j < mask.clauses[i].size; j++) { + insert_unique( storage, size, mask.clauses[i].list[j] ); + } + } + qsort( storage, size ); + return size; +} + void ?{}( __condition_blocked_queue_t & this ) { this.head = NULL; Index: src/libcfa/concurrency/preemption.c =================================================================== --- src/libcfa/concurrency/preemption.c (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/preemption.c (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -328,4 +328,18 @@ siginfo_t info; int sig = sigwaitinfo( &mask, &info ); + + if( sig < 0 ) { + //Error! + int err = errno; + switch( err ) { + case EAGAIN : + case EINTR : + continue; + case EINVAL : + abortf("Timeout was invalid."); + default: + abortf("Unhandled error %d", err); + } + } // If another signal arrived something went wrong Index: src/libcfa/concurrency/thread =================================================================== --- src/libcfa/concurrency/thread (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/thread (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -36,18 +36,18 @@ forall( dtype T | is_thread(T) ) static inline coroutine_desc* get_coroutine(T & this) { - return &get_thread(this)->cor; + return &get_thread(this)->self_cor; } forall( dtype T | is_thread(T) ) static inline monitor_desc* get_monitor(T & this) { - return &get_thread(this)->mon; + return &get_thread(this)->self_mon; } static inline coroutine_desc* get_coroutine(thread_desc * this) { - return &this->cor; + return &this->self_cor; } static inline monitor_desc* get_monitor(thread_desc * this) { - return &this->mon; + return &this->self_mon; } Index: src/libcfa/concurrency/thread.c =================================================================== --- src/libcfa/concurrency/thread.c (revision b46267011fc777d8d76dd6ebf5547ec8a41dbeba) +++ src/libcfa/concurrency/thread.c (revision a6c5d7c3b627f0180da89e9ae78c01a821292827) @@ -33,16 +33,16 @@ void ?{}(thread_desc& this) { - (this.cor){}; - this.cor.name = "Anonymous Coroutine"; - this.mon.owner = &this; - this.mon.recursion = 1; + (this.self_cor){}; + this.self_cor.name = "Anonymous Coroutine"; + this.self_mon.owner = &this; + this.self_mon.recursion = 1; + this.self_mon_p = &this.self_mon; this.next = NULL; - this.current_monitors = &this.mon; - this.current_monitor_count = 1; + (this.monitors){ &this.self_mon_p, 1, (fptr_t)0 }; } void ^?{}(thread_desc& this) { - ^(this.cor){}; + ^(this.self_cor){}; }