Index: libcfa/src/concurrency/locks.hfa =================================================================== --- libcfa/src/concurrency/locks.hfa (revision 058ece2989c03ebfdae409579c414e8364c40ead) +++ libcfa/src/concurrency/locks.hfa (revision b77f0e1fb94f6a4a2617000cc28b1371637e1fb8) @@ -30,4 +30,12 @@ #include "time.hfa" +#include + + +// futex headers +#include /* Definition of FUTEX_* constants */ +#include /* Definition of SYS_* constants */ +#include + //----------------------------------------------------------------------------- // Semaphore @@ -140,4 +148,73 @@ //----------------------------------------------------------------------------- +// futex_mutex + +// - No cond var support +// - Kernel thd blocking alternative to the spinlock +// - No ownership (will deadlock on reacq) +struct futex_mutex { + // lock state any state other than UNLOCKED is locked + // enum LockState { UNLOCKED = 0, UNCONTENDED = 1, CONTENDED = 2 }; + + // stores a lock state + int val; +}; + +// to use for FUTEX_WAKE and FUTEX_WAIT (other futex calls will need more params) +static int futex(int *uaddr, int futex_op, int val) { + return syscall(SYS_futex, uaddr, futex_op, val, NULL, NULL, 0); +} + +static inline void ?{}( futex_mutex & this ) with(this) { val = 0; } + +static inline bool internal_try_lock(futex_mutex & this, int & compare_val) with(this) { + return __atomic_compare_exchange_n((int*)&val, (int*)&compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE); +} + +static inline int internal_exchange(futex_mutex & this) with(this) { + return __atomic_exchange_n((int*)&val, 2, __ATOMIC_ACQUIRE); +} + +// if this is called recursively IT WILL DEADLOCK!!!!! +static inline void lock(futex_mutex & this) with(this) { + int state; + + + // linear backoff + for( int spin = 4; spin < 1024; spin += spin) { + state = 0; + // if unlocked, lock and return + if (internal_try_lock(this, state)) return; + if (2 == state) break; + for (int i = 0; i < spin; i++) Pause(); + } + // if (internal_try_lock(this, state)) return; + + // if not in contended state, set to be in contended state + if (state != 2) state = internal_exchange(this); + + // block and spin until we win the lock + while (state != 0) { + futex((int*)&val, FUTEX_WAIT, 2); // if val is not 2 this returns with EWOULDBLOCK + state = internal_exchange(this); + } +} + +static inline void unlock(futex_mutex & this) with(this) { + // if uncontended do atomice unlock and then return + if (__atomic_fetch_sub(&val, 1, __ATOMIC_RELEASE) == 1) return; // TODO: try acq/rel + + // otherwise threads are blocked so we must wake one + __atomic_store_n((int *)&val, 0, __ATOMIC_RELEASE); + futex((int *)&val, FUTEX_WAKE, 1); +} + +static inline void on_notify( futex_mutex & f, thread$ * t){ unpark(t); } +static inline size_t on_wait( futex_mutex & f ) {unlock(f); return 0;} + +// to set recursion count after getting signalled; +static inline void on_wakeup( futex_mutex & f, size_t recursion ) {} + +//----------------------------------------------------------------------------- // CLH Spinlock // - No recursive acquisition @@ -165,4 +242,13 @@ } +static inline void on_notify(clh_lock & this, struct thread$ * t ) { unpark(t); } +static inline size_t on_wait(clh_lock & this) { unlock(this); return 0; } +static inline void on_wakeup(clh_lock & this, size_t recursion ) { + #ifdef REACQ + lock(this); + #endif +} + + //----------------------------------------------------------------------------- // Linear backoff Spinlock @@ -171,7 +257,4 @@ __spinlock_t spinlock; - // Current thread owning the lock - struct thread$ * owner; - // List of blocked threads dlist( thread$ ) blocked_threads; @@ -179,31 +262,17 @@ // Used for comparing and exchanging volatile size_t lock_value; - - // used for linear backoff spinning - int spin_start; - int spin_end; - int spin_count; - - // after unsuccessful linear backoff yield this many times - int yield_count; -}; - -static inline void ?{}( linear_backoff_then_block_lock & this, int spin_start, int spin_end, int spin_count, int yield_count ) { +}; + +static inline void ?{}( linear_backoff_then_block_lock & this ) { this.spinlock{}; this.blocked_threads{}; this.lock_value = 0; - this.spin_start = spin_start; - this.spin_end = spin_end; - this.spin_count = spin_count; - this.yield_count = yield_count; -} -static inline void ?{}( linear_backoff_then_block_lock & this ) { this{4, 1024, 16, 0}; } +} static inline void ^?{}( linear_backoff_then_block_lock & this ) {} -static inline void ?{}( linear_backoff_then_block_lock & this, linear_backoff_then_block_lock this2 ) = void; -static inline void ?=?( linear_backoff_then_block_lock & this, linear_backoff_then_block_lock this2 ) = void; +// static inline void ?{}( linear_backoff_then_block_lock & this, linear_backoff_then_block_lock this2 ) = void; +// static inline void ?=?( linear_backoff_then_block_lock & this, linear_backoff_then_block_lock this2 ) = void; static inline bool internal_try_lock(linear_backoff_then_block_lock & this, size_t & compare_val) with(this) { if (__atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) { - owner = active_thread(); return true; } @@ -215,5 +284,4 @@ static inline bool try_lock_contention(linear_backoff_then_block_lock & this) with(this) { if (__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE) == 0) { - owner = active_thread(); return true; } @@ -234,8 +302,6 @@ static inline void lock(linear_backoff_then_block_lock & this) with(this) { - // if owner just return - if (active_thread() == owner) return; size_t compare_val = 0; - int spin = spin_start; + int spin = 4; // linear backoff for( ;; ) { @@ -244,5 +310,5 @@ if (2 == compare_val) break; for (int i = 0; i < spin; i++) Pause(); - if (spin >= spin_end) break; + if (spin >= 1024) break; spin += spin; } @@ -254,6 +320,4 @@ static inline void unlock(linear_backoff_then_block_lock & this) with(this) { - verify(lock_value > 0); - owner = 0p; if (__atomic_exchange_n(&lock_value, 0, __ATOMIC_RELEASE) == 1) return; lock( spinlock __cfaabi_dbg_ctx2 ); @@ -265,5 +329,9 @@ static inline void on_notify(linear_backoff_then_block_lock & this, struct thread$ * t ) { unpark(t); } static inline size_t on_wait(linear_backoff_then_block_lock & this) { unlock(this); return 0; } -static inline void on_wakeup(linear_backoff_then_block_lock & this, size_t recursion ) { lock(this); } +static inline void on_wakeup(linear_backoff_then_block_lock & this, size_t recursion ) { + #ifdef REACQ + lock(this); + #endif +} //----------------------------------------------------------------------------- @@ -306,5 +374,5 @@ assert(!(held && owner == active_thread())); #endif - if (held) { + if ( held ) { insert_last( blocked_threads, *active_thread() ); unlock( lock ); @@ -331,5 +399,13 @@ } -static inline void on_notify(fast_block_lock & this, struct thread$ * t ) { unpark(t); } +static inline void on_notify(fast_block_lock & this, struct thread$ * t ) with(this) { + #ifdef REACQ + lock( lock __cfaabi_dbg_ctx2 ); + insert_last( blocked_threads, *t ); + unlock( lock ); + #else + unpark(t); + #endif +} static inline size_t on_wait(fast_block_lock & this) { unlock(this); return 0; } static inline void on_wakeup(fast_block_lock & this, size_t recursion ) { } @@ -412,5 +488,4 @@ if ( owner != 0p ) { insert_last( blocked_threads, *t ); - unlock( lock ); } // lock not held @@ -419,6 +494,6 @@ recursion_count = 1; unpark( t ); - unlock( lock ); - } + } + unlock( lock ); } @@ -474,26 +549,23 @@ static inline void lock(spin_queue_lock & this) with(this) { mcs_spin_node node; - #ifdef __CFA_DEBUG__ - assert(!(held && owner == active_thread())); - #endif lock( lock, node ); while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause(); __atomic_store_n(&held, true, __ATOMIC_SEQ_CST); unlock( lock, node ); - #ifdef __CFA_DEBUG__ - owner = active_thread(); - #endif } static inline void unlock(spin_queue_lock & this) with(this) { - #ifdef __CFA_DEBUG__ - owner = 0p; - #endif __atomic_store_n(&held, false, __ATOMIC_RELEASE); } -static inline void on_notify(spin_queue_lock & this, struct thread$ * t ) { unpark(t); } +static inline void on_notify(spin_queue_lock & this, struct thread$ * t ) { + unpark(t); +} static inline size_t on_wait(spin_queue_lock & this) { unlock(this); return 0; } -static inline void on_wakeup(spin_queue_lock & this, size_t recursion ) { } +static inline void on_wakeup(spin_queue_lock & this, size_t recursion ) { + #ifdef REACQ + lock(this); + #endif +} @@ -511,9 +583,4 @@ // flag showing if lock is held volatile bool held; - - #ifdef __CFA_DEBUG__ - // for deadlock detection - struct thread$ * owner; - #endif }; @@ -529,20 +596,11 @@ static inline void lock(mcs_block_spin_lock & this) with(this) { mcs_node node; - #ifdef __CFA_DEBUG__ - assert(!(held && owner == active_thread())); - #endif lock( lock, node ); while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause(); __atomic_store_n(&held, true, __ATOMIC_SEQ_CST); unlock( lock, node ); - #ifdef __CFA_DEBUG__ - owner = active_thread(); - #endif } static inline void unlock(mcs_block_spin_lock & this) with(this) { - #ifdef __CFA_DEBUG__ - owner = 0p; - #endif __atomic_store_n(&held, false, __ATOMIC_SEQ_CST); } @@ -550,5 +608,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); return 0; } -static inline void on_wakeup(mcs_block_spin_lock & this, size_t recursion ) { } +static inline void on_wakeup(mcs_block_spin_lock & this, size_t recursion ) { + #ifdef REACQ + lock(this); + #endif +} //----------------------------------------------------------------------------- @@ -565,9 +627,4 @@ // flag showing if lock is held volatile bool held; - - #ifdef __CFA_DEBUG__ - // for deadlock detection - struct thread$ * owner; - #endif }; @@ -582,26 +639,43 @@ // if this is called recursively IT WILL DEADLOCK!!!!! static inline void lock(block_spin_lock & this) with(this) { - #ifdef __CFA_DEBUG__ - assert(!(held && owner == active_thread())); - #endif lock( lock ); while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause(); __atomic_store_n(&held, true, __ATOMIC_RELEASE); unlock( lock ); +} + +static inline void unlock(block_spin_lock & this) with(this) { + __atomic_store_n(&held, false, __ATOMIC_RELEASE); +} + +static inline void on_notify(block_spin_lock & this, struct thread$ * t ) with(this.lock) { + #ifdef REACQ + // first we acquire internal fast_block_lock + lock( lock __cfaabi_dbg_ctx2 ); + if ( held ) { // if internal fast_block_lock is held + insert_last( blocked_threads, *t ); + unlock( lock ); + return; + } + // if internal fast_block_lock is not held + held = true; #ifdef __CFA_DEBUG__ - owner = active_thread(); - #endif -} - -static inline void unlock(block_spin_lock & this) with(this) { - #ifdef __CFA_DEBUG__ - owner = 0p; - #endif - __atomic_store_n(&held, false, __ATOMIC_RELEASE); -} - -static inline void on_notify(block_spin_lock & this, struct thread$ * t ) { unpark(t); } + owner = t; + #endif + unlock( lock ); + + #endif + unpark(t); + +} static inline size_t on_wait(block_spin_lock & this) { unlock(this); return 0; } -static inline void on_wakeup(block_spin_lock & this, size_t recursion ) { } +static inline void on_wakeup(block_spin_lock & this, size_t recursion ) with(this) { + #ifdef REACQ + // now we acquire the entire block_spin_lock upon waking up + while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause(); + __atomic_store_n(&held, true, __ATOMIC_RELEASE); + unlock( lock ); // Now we release the internal fast_spin_lock + #endif +} //-----------------------------------------------------------------------------