Index: libcfa/src/bits/weakso_locks.cfa =================================================================== --- libcfa/src/bits/weakso_locks.cfa (revision c4f411e21d92b3ef455f11eb19f9db0a2be169b1) +++ libcfa/src/bits/weakso_locks.cfa (revision 9eb7f07cfca64b294b593355d1385e8bd1d0166c) @@ -25,5 +25,5 @@ void unlock( blocking_lock & ) {} void on_notify( blocking_lock &, struct thread$ * ) {} -size_t on_wait( blocking_lock & ) { return 0; } +size_t on_wait( blocking_lock &, void (*pp_fn)( void * ), void * pp_datum ) { return 0; } void on_wakeup( blocking_lock &, size_t ) {} size_t wait_count( blocking_lock & ) { return 0; } Index: libcfa/src/bits/weakso_locks.hfa =================================================================== --- libcfa/src/bits/weakso_locks.hfa (revision c4f411e21d92b3ef455f11eb19f9db0a2be169b1) +++ libcfa/src/bits/weakso_locks.hfa (revision 9eb7f07cfca64b294b593355d1385e8bd1d0166c) @@ -57,5 +57,5 @@ void unlock( blocking_lock & this ) OPTIONAL_THREAD; void on_notify( blocking_lock & this, struct thread$ * t ) OPTIONAL_THREAD; -size_t on_wait( blocking_lock & this ) OPTIONAL_THREAD; +size_t on_wait( blocking_lock & this, void (*pp_fn)( void * ), void * pp_datum ) OPTIONAL_THREAD; void on_wakeup( blocking_lock & this, size_t ) OPTIONAL_THREAD; size_t wait_count( blocking_lock & this ) OPTIONAL_THREAD; @@ -75,5 +75,5 @@ static inline bool try_lock ( multiple_acquisition_lock & this ) { return try_lock( (blocking_lock &)this ); } static inline void unlock ( multiple_acquisition_lock & this ) { unlock ( (blocking_lock &)this ); } -static inline size_t on_wait ( multiple_acquisition_lock & this ) { return on_wait ( (blocking_lock &)this ); } +static inline size_t on_wait ( multiple_acquisition_lock & this, void (*pp_fn)( void * ), void * pp_datum ) { return on_wait ( (blocking_lock &)this, pp_fn, pp_datum ); } static inline void on_wakeup( multiple_acquisition_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); } static inline void on_notify( multiple_acquisition_lock & this, struct thread$ * t ){ on_notify( (blocking_lock &)this, t ); } Index: libcfa/src/concurrency/locks.cfa =================================================================== --- libcfa/src/concurrency/locks.cfa (revision c4f411e21d92b3ef455f11eb19f9db0a2be169b1) +++ libcfa/src/concurrency/locks.cfa (revision 9eb7f07cfca64b294b593355d1385e8bd1d0166c) @@ -171,5 +171,5 @@ } -size_t on_wait( blocking_lock & this ) with( this ) { +size_t on_wait( blocking_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) with( this ) { lock( lock __cfaabi_dbg_ctx2 ); /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this ); @@ -184,5 +184,5 @@ unlock( lock ); - park(); + pre_park_then_park( pp_fn, pp_datum ); return ret; @@ -396,12 +396,13 @@ } - static size_t block_and_get_recursion( info_thread(L) & i ) { + static size_t block_and_get_recursion( info_thread(L) & i, __cfa_pre_park pp_fn, void * pp_datum ) { size_t recursion_count = 0; if ( i.lock ) { // if lock was passed get recursion count to reset to after waking thread - recursion_count = on_wait( *i.lock ); // this call blocks - } else park( ); + recursion_count = on_wait( *i.lock, pp_fn, pp_datum ); // this call blocks + } else pre_park_then_park( pp_fn, pp_datum ); return recursion_count; } + static size_t block_and_get_recursion( info_thread(L) & i ) { return block_and_get_recursion( i, pre_park_noop, 0p ); } // helper for wait()'s' with no timeout @@ -424,4 +425,6 @@ queue_info_thread( this, i ); + static void cond_alarm_register( void * node_ptr ) { register_self( (alarm_node_t *)node_ptr ); } + // helper for wait()'s' with a timeout static void queue_info_thread_timeout( condition_variable(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) { @@ -433,8 +436,8 @@ // registers alarm outside cond lock to avoid deadlock - register_self( &node_wrap.alarm_node ); + // register_self( &node_wrap.alarm_node ); // blocks here - size_t recursion_count = block_and_get_recursion( info ); + size_t recursion_count = block_and_get_recursion( info, cond_alarm_register, (void *)(&node_wrap.alarm_node) ); // park(); @@ -503,5 +506,5 @@ info_thread( L ) i = { active_thread(), info, &l }; insert_last( blocked_threads, i ); - size_t recursion_count = on_wait( *i.lock ); // blocks here + size_t recursion_count = on_wait( *i.lock, pre_park_noop, 0p ); // blocks here // park( ); on_wakeup(*i.lock, recursion_count); @@ -552,5 +555,4 @@ // return recursion_count; // } - static void queue_info_thread_timeout( pthread_cond_var(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) { @@ -562,8 +564,8 @@ // registers alarm outside cond lock to avoid deadlock - register_self( &node_wrap.alarm_node ); + // register_self( &node_wrap.alarm_node ); // C_TODO: fix race: registers itself and then alarm handler calls on_notify before block_and_get_recursion is run // blocks here - size_t recursion_count = block_and_get_recursion( info ); + size_t recursion_count = block_and_get_recursion( info, cond_alarm_register, (void *)(&node_wrap.alarm_node) ); // park(); Index: libcfa/src/concurrency/locks.hfa =================================================================== --- libcfa/src/concurrency/locks.hfa (revision c4f411e21d92b3ef455f11eb19f9db0a2be169b1) +++ libcfa/src/concurrency/locks.hfa (revision 9eb7f07cfca64b294b593355d1385e8bd1d0166c) @@ -44,4 +44,27 @@ // - change messy big blocking lock from inheritance to composition to remove need for flags +typedef void (*__cfa_pre_park)( void * ); + +static inline void pre_park_noop( void * ) {} + +//----------------------------------------------------------------------------- +// is_blocking_lock +forall( L & | sized(L) ) +trait is_blocking_lock { + // For synchronization locks to use when acquiring + void on_notify( L &, struct thread$ * ); + + // For synchronization locks to use when releasing + size_t on_wait( L &, __cfa_pre_park pp_fn, void * pp_datum ); + + // to set recursion count after getting signalled; + void on_wakeup( L &, size_t recursion ); +}; + +static inline void pre_park_then_park( __cfa_pre_park pp_fn, void * pp_datum ) { + pp_fn( pp_datum ); + park(); +} + //----------------------------------------------------------------------------- // Semaphore @@ -69,5 +92,5 @@ static inline bool try_lock ( single_acquisition_lock & this ) { return try_lock( (blocking_lock &)this ); } static inline void unlock ( single_acquisition_lock & this ) { unlock ( (blocking_lock &)this ); } -static inline size_t on_wait ( single_acquisition_lock & this ) { return on_wait ( (blocking_lock &)this ); } +static inline size_t on_wait ( single_acquisition_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { return on_wait ( (blocking_lock &)this, pp_fn, pp_datum ); } static inline void on_wakeup( single_acquisition_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); } static inline void on_notify( single_acquisition_lock & this, struct thread$ * t ) { on_notify( (blocking_lock &)this, t ); } @@ -86,5 +109,5 @@ static inline bool try_lock ( owner_lock & this ) { return try_lock( (blocking_lock &)this ); } static inline void unlock ( owner_lock & this ) { unlock ( (blocking_lock &)this ); } -static inline size_t on_wait ( owner_lock & this ) { return on_wait ( (blocking_lock &)this ); } +static inline size_t on_wait ( owner_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { return on_wait ( (blocking_lock &)this, pp_fn, pp_datum ); } static inline void on_wakeup( owner_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); } static inline void on_notify( owner_lock & this, struct thread$ * t ) { on_notify( (blocking_lock &)this, t ); } @@ -218,5 +241,9 @@ static inline void on_notify( futex_mutex & f, thread$ * t){ unpark(t); } -static inline size_t on_wait( futex_mutex & f ) { unlock(f); park(); return 0; } +static inline size_t on_wait( futex_mutex & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} // to set recursion count after getting signalled; @@ -288,5 +315,9 @@ static inline void on_notify( go_mutex & f, thread$ * t){ unpark( t ); } -static inline size_t on_wait( go_mutex & f ) { unlock( f ); park(); return 0; } +static inline size_t on_wait( go_mutex & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( go_mutex & f, size_t recursion ) {} @@ -362,5 +393,9 @@ static inline void on_notify( exp_backoff_then_block_lock & this, struct thread$ * t ) { unpark( t ); } -static inline size_t on_wait( exp_backoff_then_block_lock & this ) { unlock( this ); park(); return 0; } +static inline size_t on_wait( exp_backoff_then_block_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( exp_backoff_then_block_lock & this, size_t recursion ) { lock( this ); } @@ -419,5 +454,9 @@ unlock( lock ); } -static inline size_t on_wait( fast_block_lock & this) { unlock(this); park(); return 0; } +static inline size_t on_wait( fast_block_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( fast_block_lock & this, size_t recursion ) { } @@ -497,5 +536,5 @@ } -static inline void on_notify(simple_owner_lock & this, thread$ * t ) with(this) { +static inline void on_notify( simple_owner_lock & this, thread$ * t ) with(this) { lock( lock __cfaabi_dbg_ctx2 ); // lock held @@ -512,5 +551,5 @@ } -static inline size_t on_wait( simple_owner_lock & this ) with(this) { +static inline size_t on_wait( simple_owner_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) with(this) { lock( lock __cfaabi_dbg_ctx2 ); /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this ); @@ -524,5 +563,6 @@ active_thread()->link_node = (void *)&node; unlock( lock ); - park(); + + pre_park_then_park( pp_fn, pp_datum ); return ret; @@ -624,5 +664,9 @@ unpark(t); } -static inline size_t on_wait( spin_queue_lock & this ) { unlock( this ); park(); return 0; } +static inline size_t on_wait( spin_queue_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( spin_queue_lock & this, size_t recursion ) { lock( this ); } @@ -665,5 +709,9 @@ static inline void on_notify( mcs_block_spin_lock & this, struct thread$ * t ) { unpark( t ); } -static inline size_t on_wait( mcs_block_spin_lock & this) { unlock( this ); park(); return 0; } +static inline size_t on_wait( mcs_block_spin_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( mcs_block_spin_lock & this, size_t recursion ) {lock( this ); } @@ -717,5 +765,9 @@ unpark(t); } -static inline size_t on_wait( block_spin_lock & this ) { unlock( this ); park(); return 0; } +static inline size_t on_wait( block_spin_lock & this, __cfa_pre_park pp_fn, void * pp_datum ) { + unlock( this ); + pre_park_then_park( pp_fn, pp_datum ); + return 0; +} static inline void on_wakeup( block_spin_lock & this, size_t recursion ) with(this) { // now we acquire the entire block_spin_lock upon waking up @@ -724,18 +776,4 @@ unlock( lock ); // Now we release the internal fast_spin_lock } - -//----------------------------------------------------------------------------- -// is_blocking_lock -forall( L & | sized(L) ) -trait is_blocking_lock { - // For synchronization locks to use when acquiring - void on_notify( L &, struct thread$ * ); - - // For synchronization locks to use when releasing - size_t on_wait( L & ); - - // to set recursion count after getting signalled; - void on_wakeup( L &, size_t recursion ); -}; //-----------------------------------------------------------------------------