Changeset b110bcc for libcfa/src/concurrency/locks.hfa

Timestamp:

Apr 21, 2023, 5:36:12 PM (2 years ago)

Author:

JiadaL <j82liang@…>

Branches:

ADT, master

Children:

28f8f15, 6e4c44d

Parents:

2ed94a9 (diff), 699a97d (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

File:

• libcfa/src/concurrency/locks.hfa (modified) (10 diffs)

libcfa/src/concurrency/locks.hfa

@@ -32 +32 @@
 #include <fstream.hfa>
 
-
 // futex headers
 #include <linux/futex.h>      /* Definition of FUTEX_* constants */
@@ -155 +154 @@
 // futex_mutex
 
-// - No cond var support
 // - Kernel thd blocking alternative to the spinlock
 // - No ownership (will deadlock on reacq)
@@ -185 +183 @@
         int state;
 
-        
-        // // linear backoff omitted for now
-        // 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();
-        // }
-
-        // no contention try to acquire
-        if (internal_try_lock(this, state)) return;
+        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();
+        }
+
+        // // no contention try to acquire
+        // if (internal_try_lock(this, state)) return;
         
         // if not in contended state, set to be in contended state
@@ -209 +205 @@
 
 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
+        // if uncontended do atomic unlock and then return
+    if (__atomic_exchange_n(&val, 0, __ATOMIC_RELEASE) == 1) return;
         
         // otherwise threads are blocked so we must wake one
-        __atomic_store_n((int *)&val, 0, __ATOMIC_RELEASE);
         futex((int *)&val, FUTEX_WAKE, 1);
 }
@@ -222 +217 @@
 // to set recursion count after getting signalled;
 static inline void on_wakeup( futex_mutex & f, size_t recursion ) {}
+
+//-----------------------------------------------------------------------------
+// go_mutex
+
+// - Kernel thd blocking alternative to the spinlock
+// - No ownership (will deadlock on reacq)
+// - Golang's flavour of mutex
+// - Impl taken from Golang: src/runtime/lock_futex.go
+struct go_mutex {
+        // lock state any state other than UNLOCKED is locked
+        // enum LockState { UNLOCKED = 0, LOCKED = 1, SLEEPING = 2 };
+        
+        // stores a lock state
+        int val; 
+};
+
+static inline void  ?{}( go_mutex & this ) with(this) { val = 0; }
+
+static inline bool internal_try_lock(go_mutex & this, int & compare_val, int new_val ) with(this) {
+        return __atomic_compare_exchange_n((int*)&val, (int*)&compare_val, new_val, false, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE);
+}
+
+static inline int internal_exchange(go_mutex & this, int swap ) with(this) {
+        return __atomic_exchange_n((int*)&val, swap, __ATOMIC_ACQUIRE);
+}
+
+// if this is called recursively IT WILL DEADLOCK!!!!!
+static inline void lock(go_mutex & this) with(this) {
+        int state, init_state;
+
+    // speculative grab
+    state = internal_exchange(this, 1);
+    if ( !state ) return; // state == 0
+    init_state = state;
+    for (;;) {
+        for( int i = 0; i < 4; i++ ) {
+            while( !val ) { // lock unlocked
+                state = 0;
+                if (internal_try_lock(this, state, init_state)) return;
+            }
+            for (int i = 0; i < 30; i++) Pause();
+        }
+
+        while( !val ) { // lock unlocked
+            state = 0;
+            if (internal_try_lock(this, state, init_state)) return;
+        }
+        sched_yield();
+        
+        // if not in contended state, set to be in contended state
+        state = internal_exchange(this, 2);
+        if ( !state ) return; // state == 0
+        init_state = 2;
+        futex((int*)&val, FUTEX_WAIT, 2); // if val is not 2 this returns with EWOULDBLOCK
+    }
+}
+
+static inline void unlock( go_mutex & this ) with(this) {
+        // if uncontended do atomic unlock and then return
+    if (__atomic_exchange_n(&val, 0, __ATOMIC_RELEASE) == 1) return;
+        
+        // otherwise threads are blocked so we must wake one
+        futex((int *)&val, FUTEX_WAKE, 1);
+}
+
+static inline void on_notify( go_mutex & f, thread$ * t){ unpark(t); }
+static inline size_t on_wait( go_mutex & f ) {unlock(f); return 0;}
+static inline void on_wakeup( go_mutex & f, size_t recursion ) {}
 
 //-----------------------------------------------------------------------------
@@ -253 +316 @@
 static inline void on_wakeup(clh_lock & this, size_t recursion ) { lock(this); }
 
-
 //-----------------------------------------------------------------------------
 // Exponential backoff then block lock
@@ -272 +334 @@
         this.lock_value = 0;
 }
-static inline void ^?{}( exp_backoff_then_block_lock & this ) {}
-// static inline void ?{}( exp_backoff_then_block_lock & this, exp_backoff_then_block_lock this2 ) = void;
-// static inline void ?=?( exp_backoff_then_block_lock & this, exp_backoff_then_block_lock this2 ) = void;
+
+static inline void  ^?{}( exp_backoff_then_block_lock & this ){}
 
 static inline bool internal_try_lock(exp_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)) {
-                return true;
-        }
-        return false;
+        return __atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
 }
 
@@ -286 +344 @@
 
 static inline bool try_lock_contention(exp_backoff_then_block_lock & this) with(this) {
-        if (__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE) == 0) {
-                return true;
-        }
-        return false;
+        return !__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE);
 }
 
 static inline bool block(exp_backoff_then_block_lock & this) with(this) {
-        lock( spinlock __cfaabi_dbg_ctx2 ); // TODO change to lockfree queue (MPSC)
-        if (lock_value != 2) {
-                unlock( spinlock );
-                return true;
-        }
-        insert_last( blocked_threads, *active_thread() );
-        unlock( spinlock );
+    lock( spinlock __cfaabi_dbg_ctx2 );
+    if (__atomic_load_n( &lock_value, __ATOMIC_SEQ_CST) != 2) {
+        unlock( spinlock );
+        return true;
+    }
+    insert_last( blocked_threads, *active_thread() );
+    unlock( spinlock );
         park( );
         return true;
@@ -307 +362 @@
         size_t compare_val = 0;
         int spin = 4;
+
         // linear backoff
         for( ;; ) {
@@ -324 +380 @@
 static inline void unlock(exp_backoff_then_block_lock & this) with(this) {
     if (__atomic_exchange_n(&lock_value, 0, __ATOMIC_RELEASE) == 1) return;
-        lock( spinlock __cfaabi_dbg_ctx2 );
-        thread$ * t = &try_pop_front( blocked_threads );
-        unlock( spinlock );
-        unpark( t );
+    lock( spinlock __cfaabi_dbg_ctx2 );
+    thread$ * t = &try_pop_front( blocked_threads );
+    unlock( spinlock );
+    unpark( t );
 }