Changeset 24d6572 for libcfa/src/concurrency/locks.cfa

Timestamp:

Jun 12, 2023, 2:45:32 PM (2 years ago)

Author:

Fangren Yu <f37yu@…>

Branches:

ast-experimental, master

Children:

62d62db

Parents:

34b4268 (diff), 251ce80 (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' into ast-experimental

File:

• libcfa/src/concurrency/locks.cfa (modified) (16 diffs)

libcfa/src/concurrency/locks.cfa

@@ -5 +5 @@
 // file "LICENCE" distributed with Cforall.
 //
-// locks.hfa -- LIBCFATHREAD
+// locks.cfa -- LIBCFATHREAD
 // Runtime locks that used with the runtime thread system.
 //
@@ -16 +16 @@
 
 #define __cforall_thread__
-#define _GNU_SOURCE
 
 #include "locks.hfa"
@@ -80 +79 @@
         // lock is held by some other thread
         if ( owner != 0p && owner != thrd ) {
-                insert_last( blocked_threads, *thrd );
+        select_node node;
+                insert_last( blocked_threads, node );
                 wait_count++;
                 unlock( lock );
                 park( );
-        }
-        // multi acquisition lock is held by current thread
-        else if ( owner == thrd && multi_acquisition ) {
+        return;
+        } else if ( owner == thrd && multi_acquisition ) { // multi acquisition lock is held by current thread
                 recursion_count++;
-                unlock( lock );
-        }
-        // lock isn't held
-        else {
+        } else {  // lock isn't held
                 owner = thrd;
                 recursion_count = 1;
-                unlock( lock );
-        }
+        }
+    unlock( lock );
 }
 
@@ -118 +114 @@
 }
 
-static void pop_and_set_new_owner( blocking_lock & this ) with( this ) {
-        thread$ * t = &try_pop_front( blocked_threads );
-        owner = t;
-        recursion_count = ( t ? 1 : 0 );
-        if ( t ) wait_count--;
-        unpark( t );
+static inline void pop_node( blocking_lock & this ) with( this ) {
+    __handle_waituntil_OR( blocked_threads );
+    select_node * node = &try_pop_front( blocked_threads );
+    if ( node ) {
+        wait_count--;
+        owner = node->blocked_thread;
+        recursion_count = 1;
+        // if ( !node->clause_status || __make_select_node_available( *node ) ) unpark( node->blocked_thread );
+        wake_one( blocked_threads, *node );
+    } else {
+        owner = 0p;
+        recursion_count = 0;
+    }
 }
 
@@ -135 +138 @@
         recursion_count--;
         if ( recursion_count == 0 ) {
-                pop_and_set_new_owner( this );
+                pop_node( this );
         }
         unlock( lock );
@@ -148 +151 @@
         // lock held
         if ( owner != 0p ) {
-                insert_last( blocked_threads, *t );
+                insert_last( blocked_threads, *(select_node *)t->link_node );
                 wait_count++;
-                unlock( lock );
         }
         // lock not held
@@ -157 +159 @@
                 recursion_count = 1;
                 unpark( t );
-                unlock( lock );
-        }
-}
-
-size_t on_wait( blocking_lock & this ) with( this ) {
+        }
+    unlock( lock );
+}
+
+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 );
@@ -168 +170 @@
         size_t ret = recursion_count;
 
-        pop_and_set_new_owner( this );
+        pop_node( this );
+
+    select_node node;
+    active_thread()->link_node = (void *)&node;
         unlock( lock );
+
+    pre_park_then_park( pp_fn, pp_datum );
+
         return ret;
 }
@@ -176 +184 @@
         recursion_count = recursion;
 }
+
+// waituntil() support
+bool register_select( blocking_lock & this, select_node & node ) with(this) {
+    lock( lock __cfaabi_dbg_ctx2 );
+        thread$ * thrd = active_thread();
+
+        // single acquisition lock is held by current thread
+        /* paranoid */ verifyf( owner != thrd || multi_acquisition, "Single acquisition lock holder (%p) attempted to reacquire the lock %p resulting in a deadlock.", owner, &this );
+
+    if ( !node.park_counter && ( (owner == thrd && multi_acquisition) || owner == 0p ) ) { // OR special case
+        if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+           unlock( lock );
+           return false;
+        }
+    }
+
+        // lock is held by some other thread
+        if ( owner != 0p && owner != thrd ) {
+                insert_last( blocked_threads, node );
+                wait_count++;
+                unlock( lock );
+        return false;
+        } else if ( owner == thrd && multi_acquisition ) { // multi acquisition lock is held by current thread
+                recursion_count++;
+        } else {  // lock isn't held
+                owner = thrd;
+                recursion_count = 1;
+        }
+
+    if ( node.park_counter ) __make_select_node_available( node );
+    unlock( lock );
+    return true;
+}
+
+bool unregister_select( blocking_lock & this, select_node & node ) with(this) {
+    lock( lock __cfaabi_dbg_ctx2 );
+    if ( node`isListed ) {
+        remove( node );
+        wait_count--;
+        unlock( lock );
+        return false;
+    }
+    
+    if ( owner == active_thread() ) {
+        /* paranoid */ verifyf( recursion_count == 1 || multi_acquisition, "Thread %p attempted to unlock owner lock %p in waituntil unregister, which is not recursive but has a recursive count of %zu", active_thread(), &this, recursion_count );
+        // if recursion count is zero release lock and set new owner if one is waiting
+        recursion_count--;
+        if ( recursion_count == 0 ) {
+            pop_node( this );
+        }
+    }
+        unlock( lock );
+    return false;
+}
+
+void on_selected( blocking_lock & this, select_node & node ) {}
 
 //-----------------------------------------------------------------------------
@@ -312 +376 @@
         int counter( condition_variable(L) & this ) with(this) { return count; }
 
-        static size_t queue_and_get_recursion( condition_variable(L) & this, info_thread(L) * i ) with(this) {
+        static void enqueue_thread( condition_variable(L) & this, info_thread(L) * i ) with(this) {
                 // add info_thread to waiting queue
                 insert_last( blocked_threads, *i );
                 count++;
-                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 );
-                }
-                return recursion_count;
-        }
+        }
+
+    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, 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
         static void queue_info_thread( condition_variable(L) & this, info_thread(L) & i ) with(this) {
                 lock( lock __cfaabi_dbg_ctx2 );
-                size_t recursion_count = queue_and_get_recursion(this, &i);
+        enqueue_thread( this, &i );
                 unlock( lock );
 
                 // blocks here
-                park( );
+        size_t recursion_count = block_and_get_recursion( i );
 
                 // resets recursion count here after waking
-                if (i.lock) on_wakeup(*i.lock, recursion_count);
+                if ( i.lock ) on_wakeup( *i.lock, recursion_count );
         }
 
@@ -341 +409 @@
                 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) {
                 lock( lock __cfaabi_dbg_ctx2 );
-                size_t recursion_count = queue_and_get_recursion(this, &info);
+        enqueue_thread( this, &info );
                 alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
                 unlock( lock );
 
-                // registers alarm outside cond lock to avoid deadlock
-                register_self( &node_wrap.alarm_node );
-
-                // blocks here
-                park();
+                // blocks here and registers alarm node before blocking after releasing locks to avoid deadlock
+        size_t recursion_count = block_and_get_recursion( info, cond_alarm_register, (void *)(&node_wrap.alarm_node) );
+                // park();
 
                 // unregisters alarm so it doesn't go off if this happens first
@@ -358 +426 @@
 
                 // resets recursion count here after waking
-                if (info.lock) on_wakeup(*info.lock, recursion_count);
+                if ( info.lock ) on_wakeup( *info.lock, recursion_count );
         }
 
@@ -418 +486 @@
                 info_thread( L ) i = { active_thread(), info, &l };
                 insert_last( blocked_threads, i );
-                size_t recursion_count = on_wait( *i.lock );
-                park( );
+                size_t recursion_count = on_wait( *i.lock, pre_park_noop, 0p ); // blocks here
+                // park( );
                 on_wakeup(*i.lock, recursion_count);
         }
@@ -460 +528 @@
         bool empty ( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
 
-        static size_t queue_and_get_recursion( pthread_cond_var(L) & this, info_thread(L) * i ) with(this) {
-                // add info_thread to waiting queue
-                insert_last( blocked_threads, *i );
-                size_t recursion_count = 0;
-                recursion_count = on_wait( *i->lock );
-                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) {
                 lock( lock __cfaabi_dbg_ctx2 );
-                size_t recursion_count = queue_and_get_recursion(this, &info);
+        insert_last( blocked_threads, info );
                 pthread_alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
                 unlock( lock );
 
-                // registers alarm outside cond lock to avoid deadlock
-                register_self( &node_wrap.alarm_node );
-
-                // blocks here
-                park();
-
-                // unregisters alarm so it doesn't go off if this happens first
+                // blocks here and registers alarm node before blocking after releasing locks to avoid deadlock
+        size_t recursion_count = block_and_get_recursion( info, cond_alarm_register, (void *)(&node_wrap.alarm_node) );
+
+                // unregisters alarm so it doesn't go off if signal happens first
                 unregister_self( &node_wrap.alarm_node );
 
                 // resets recursion count here after waking
-                if (info.lock) on_wakeup(*info.lock, recursion_count);
+                if ( info.lock ) on_wakeup( *info.lock, recursion_count );
         }
 
@@ -494 +551 @@
                 lock( lock __cfaabi_dbg_ctx2 );
                 info_thread( L ) i = { active_thread(), info, &l };
-                size_t recursion_count = queue_and_get_recursion(this, &i);
-                unlock( lock );
-                park( );
-                on_wakeup(*i.lock, recursion_count);
+        insert_last( blocked_threads, i );
+                unlock( lock );
+
+        // blocks here
+                size_t recursion_count = block_and_get_recursion( i );
+
+                on_wakeup( *i.lock, recursion_count );
         }
 
@@ -585 +645 @@
         return thrd != 0p;
 }
+