Changeset 044ae62 for libcfa

Timestamp:

May 29, 2023, 11:44:29 AM (3 years ago)

Author:

JiadaL <j82liang@…>

Branches:

ADT

Children:

fa2c005

Parents:

3a513d89 (diff), 2b78949 (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 ADT

Location:

libcfa/src

Files:

• Makefile.am (modified) (2 diffs)
• bits/weakso_locks.cfa (modified) (2 diffs)
• bits/weakso_locks.hfa (modified) (4 diffs)
• concurrency/channel.hfa (modified) (21 diffs)
• concurrency/future.hfa (modified) (8 diffs)
• concurrency/invoke.h (modified) (1 diff)
• concurrency/locks.cfa (modified) (15 diffs)
• concurrency/locks.hfa (modified) (39 diffs)
• concurrency/mutex_stmt.hfa (modified) (2 diffs)
• concurrency/preemption.cfa (modified) (1 diff)
• concurrency/select.cfa (added)
• concurrency/select.hfa (modified) (1 diff)
• concurrency/thread.cfa (modified) (3 diffs)

libcfa/src/Makefile.am

@@ -115 +115 @@
         concurrency/kernel/fwd.hfa \
         concurrency/mutex_stmt.hfa \
-    concurrency/select.hfa \
     concurrency/channel.hfa \
     concurrency/actor.hfa 
@@ -128 +127 @@
         concurrency/monitor.hfa \
         concurrency/mutex.hfa \
+    concurrency/select.hfa \
         concurrency/thread.hfa
 

libcfa/src/bits/weakso_locks.cfa

@@ -15 +15 @@
 // Update Count     :
 //
-
 #include "bits/weakso_locks.hfa"
-
 #pragma GCC visibility push(default)
 
@@ -27 +25 @@
 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; }
+bool register_select( blocking_lock & this, select_node & node ) { return false; }
+bool unregister_select( blocking_lock & this, select_node & node ) { return false; }
+bool on_selected( blocking_lock & this, select_node & node ) { return true; }
+

libcfa/src/bits/weakso_locks.hfa

@@ -23 +23 @@
 #include "containers/list.hfa"
 
-struct thread$;
+struct select_node;
 
 //-----------------------------------------------------------------------------
@@ -32 +32 @@
 
         // List of blocked threads
-        dlist( thread$ ) blocked_threads;
+        dlist( select_node ) blocked_threads;
 
         // Count of current blocked threads
@@ -57 +57 @@
 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;
+bool register_select( blocking_lock & this, select_node & node ) OPTIONAL_THREAD;
+bool unregister_select( blocking_lock & this, select_node & node ) OPTIONAL_THREAD;
+bool on_selected( blocking_lock & this, select_node & node ) OPTIONAL_THREAD;
 
 //----------
@@ -72 +75 @@
 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 ); }
+static inline bool   register_select( multiple_acquisition_lock & this, select_node & node ) { return register_select( (blocking_lock &)this, node ); }
+static inline bool   unregister_select( multiple_acquisition_lock & this, select_node & node ) { return unregister_select( (blocking_lock &)this, node ); }
+static inline bool   on_selected( multiple_acquisition_lock & this, select_node & node ) { return on_selected( (blocking_lock &)this, node ); }

libcfa/src/concurrency/channel.hfa

@@ +1 @@
+//
+// Cforall Version 1.0.0 Copyright (C) 2021 University of Waterloo
+//
+// The contents of this file are covered under the licence agreement in the
+// file "LICENCE" distributed with Cforall.
+//
+// channel.hfa -- LIBCFATHREAD
+// Runtime locks that used with the runtime thread system.
+//
+// Author           : Colby Alexander Parsons
+// Created On       : Thu Jan 21 19:46:50 2022
+// Last Modified By :
+// Last Modified On :
+// Update Count     :
+//
+
 #pragma once
 
 #include <locks.hfa>
 #include <list.hfa>
-#include <mutex_stmt.hfa>
-
-// link field used for threads waiting on channel
-struct wait_link {
-    // used to put wait_link on a dl queue
-    inline dlink(wait_link);
-
-    // waiting thread
-    struct thread$ * t;
-
-    // shadow field
-    void * elem;
-};
-P9_EMBEDDED( wait_link, dlink(wait_link) )
-
-static inline void ?{}( wait_link & this, thread$ * t, void * elem ) {
-    this.t = t;
-    this.elem = elem;
-}
-
-// wake one thread from the list
-static inline void wake_one( dlist( wait_link ) & queue ) {
-    wait_link & popped = try_pop_front( queue );
-    unpark( popped.t );
-}
+#include "select.hfa"
 
 // returns true if woken due to shutdown
 // blocks thread on list and releases passed lock
-static inline bool block( dlist( wait_link ) & queue, void * elem_ptr, go_mutex & lock ) {
-    wait_link w{ active_thread(), elem_ptr };
-    insert_last( queue, w );
+static inline bool block( dlist( select_node ) & queue, void * elem_ptr, go_mutex & lock ) {
+    select_node sn{ active_thread(), elem_ptr };
+    insert_last( queue, sn );
     unlock( lock );
     park();
-    return w.elem == 0p;
+    return sn.extra == 0p;
+}
+
+// Waituntil support (un)register_select helper routine
+// Sets select node avail if not special OR case and then unlocks
+static inline void __set_avail_then_unlock( select_node & node, go_mutex & mutex_lock ) {
+    if ( node.park_counter ) __make_select_node_available( node );
+    unlock( mutex_lock );
 }
 
@@ -59 +58 @@
     size_t size, front, back, count;
     T * buffer;
-    dlist( wait_link ) prods, cons; // lists of blocked threads
-    go_mutex mutex_lock;            // MX lock
-    bool closed;                    // indicates channel close/open
-    #ifdef CHAN_STATS
-    size_t blocks, operations;      // counts total ops and ops resulting in a blocked thd
+    dlist( select_node ) prods, cons; // lists of blocked threads
+    go_mutex mutex_lock;              // MX lock
+    bool closed;                      // indicates channel close/open
+    #ifdef CHAN_STATS
+    size_t p_blocks, p_ops, c_blocks, c_ops;      // counts total ops and ops resulting in a blocked thd
     #endif
 };
@@ -70 +69 @@
     size = _size;
     front = back = count = 0;
-    buffer = aalloc( size );
+    if ( size != 0 ) buffer = aalloc( size );
     prods{};
     cons{};
@@ -76 +75 @@
     closed = false;
     #ifdef CHAN_STATS
-    blocks = 0;
-    operations = 0;
+    p_blocks = 0;
+    p_ops = 0;
+    c_blocks = 0;
+    c_ops = 0;
     #endif
 }
@@ -84 +85 @@
 static inline void ^?{}( channel(T) &c ) with(c) {
     #ifdef CHAN_STATS
-    printf("Channel %p Blocks: %lu, Operations: %lu, %.2f%% of ops blocked\n", &c, blocks, operations, ((double)blocks)/operations * 100);
-    #endif
-    verifyf( cons`isEmpty && prods`isEmpty, "Attempted to delete channel with waiting threads (Deadlock).\n" );
-    delete( buffer );
-}
-static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
-static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
+    printf("Channel %p Blocks: %lu,\t\tOperations: %lu,\t%.2f%% of ops blocked\n", &c, p_blocks + c_blocks, p_ops + c_ops, ((double)p_blocks + c_blocks)/(p_ops + c_ops) * 100);
+    printf("Channel %p Consumer Blocks: %lu,\tConsumer Ops: %lu,\t%.2f%% of Consumer ops blocked\n", &c, p_blocks, p_ops, ((double)p_blocks)/p_ops * 100);
+    printf("Channel %p Producer Blocks: %lu,\tProducer Ops: %lu,\t%.2f%% of Producer ops blocked\n", &c, c_blocks, c_ops, ((double)c_blocks)/c_ops * 100);
+    #endif
+    verifyf( __handle_waituntil_OR( cons ) || __handle_waituntil_OR( prods ) || cons`isEmpty && prods`isEmpty, 
+        "Attempted to delete channel with waiting threads (Deadlock).\n" );
+    if ( size != 0 ) delete( buffer );
+}
+static inline size_t get_count( channel(T) & chan ) with(chan) { return __atomic_load_n( &count, __ATOMIC_RELAXED ); }
+static inline size_t get_size( channel(T) & chan ) with(chan) { return __atomic_load_n( &size, __ATOMIC_RELAXED ); }
 static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
 static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
@@ -102 +106 @@
     // flush waiting consumers and producers
     while ( has_waiting_consumers( chan ) ) {
-        cons`first.elem = 0p;
+        if( !__handle_waituntil_OR( cons ) ) // ensure we only signal special OR case threads when they win the race
+            break;  // if __handle_waituntil_OR returns false cons is empty so break
+        cons`first.extra = 0p;
         wake_one( cons );
     }
     while ( has_waiting_producers( chan ) ) {
-        prods`first.elem = 0p;
+        if( !__handle_waituntil_OR( prods ) ) // ensure we only signal special OR case threads when they win the race
+            break;  // if __handle_waituntil_OR returns false prods is empty so break
+        prods`first.extra = 0p;
         wake_one( prods );
     }
@@ -114 +122 @@
 static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
 
+// used to hand an element to a blocked consumer and signal it
+static inline void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
+    memcpy( cons`first.extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
+    wake_one( cons );
+}
+
+// used to hand an element to a blocked producer and signal it
+static inline void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
+    memcpy( (void *)&retval, prods`first.extra, sizeof(T) );
+    wake_one( prods );
+}
+
 static inline void flush( channel(T) & chan, T elem ) with(chan) {
     lock( mutex_lock );
     while ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
+        __cons_handoff( chan, elem );
     }
     unlock( mutex_lock );
@@ -125 +144 @@
 // handles buffer insert
 static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
-    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
     count += 1;
     back++;
@@ -131 +150 @@
 }
 
-// does the buffer insert or hands elem directly to consumer if one is waiting
-static inline void __do_insert( channel(T) & chan, T & elem ) with(chan) {
-    if ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    } else __buf_insert( chan, elem );
-}
-
 // needed to avoid an extra copy in closed case
 static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
     lock( mutex_lock );
     #ifdef CHAN_STATS
-    operations++;
-    #endif
+    p_ops++;
+    #endif
+
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( chan, elem );
+        unlock( mutex_lock );
+        return true;
+    }
+
     if ( count == size ) { unlock( mutex_lock ); return false; }
-    __do_insert( chan, elem );
+
+    __buf_insert( chan, elem );
     unlock( mutex_lock );
     return true;
@@ -157 +177 @@
 // handles closed case of insert routine
 static inline void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
-    channel_closed except{&channel_closed_vt, &elem, &chan };
+    channel_closed except{ &channel_closed_vt, &elem, &chan };
     throwResume except; // throw closed resumption
     if ( !__internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
@@ -172 +192 @@
 
     #ifdef CHAN_STATS
-    if ( !closed ) operations++;
+    if ( !closed ) p_ops++;
     #endif
 
@@ -182 +202 @@
     }
 
-    // have to check for the zero size channel case
-    if ( size == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T));
-        wake_one( cons );
-        unlock( mutex_lock );
-        return true;
+    // buffer count must be zero if cons are blocked (also handles zero-size case)
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( chan, elem );
+        unlock( mutex_lock );
+        return;
     }
 
@@ -193 +213 @@
     if ( count == size ) {
         #ifdef CHAN_STATS
-        blocks++;
+        p_blocks++;
         #endif
 
@@ -202 +222 @@
     } // if
 
-    if ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    } else __buf_insert( chan, elem );
-    
-    unlock( mutex_lock );
-    return;
-}
-
-// handles buffer remove
-static inline void __buf_remove( channel(T) & chan, T & retval ) with(chan) {
-    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    __buf_insert( chan, elem );
+    unlock( mutex_lock );
+}
+
+// does the buffer remove and potentially does waiting producer work
+static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
+    memcpy( (void *)&retval, (void *)&buffer[front], sizeof(T) );
     count -= 1;
     front = (front + 1) % size;
-}
-
-// does the buffer remove and potentially does waiting producer work
-static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
-    __buf_remove( chan, retval );
     if (count == size - 1 && !prods`isEmpty ) {
-        __buf_insert( chan, *(T *)prods`first.elem );  // do waiting producer work
+        if ( !__handle_waituntil_OR( prods ) ) return;
+        __buf_insert( chan, *(T *)prods`first.extra );  // do waiting producer work
         wake_one( prods );
     }
@@ -231 +242 @@
     lock( mutex_lock );
     #ifdef CHAN_STATS
-    operations++;
-    #endif
+    c_ops++;
+    #endif
+
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, retval );
+        unlock( mutex_lock );
+        return true;
+    }
+
     if ( count == 0 ) { unlock( mutex_lock ); return false; }
+
     __do_remove( chan, retval );
     unlock( mutex_lock );
@@ -244 +264 @@
 static inline [T, bool] try_remove( channel(T) & chan ) {
     T retval;
-    return [ retval, __internal_try_remove( chan, retval ) ];
-}
-
-static inline T try_remove( channel(T) & chan, T elem ) {
+    bool success = __internal_try_remove( chan, retval );
+    return [ retval, success ];
+}
+
+static inline T try_remove( channel(T) & chan ) {
     T retval;
     __internal_try_remove( chan, retval );
@@ -255 +276 @@
 // handles closed case of insert routine
 static inline void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
-    channel_closed except{&channel_closed_vt, 0p, &chan };
+    channel_closed except{ &channel_closed_vt, 0p, &chan };
     throwResume except; // throw resumption
     if ( !__internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
@@ -269 +290 @@
 
     #ifdef CHAN_STATS
-    if ( !closed ) operations++;
+    if ( !closed ) c_ops++;
     #endif
 
@@ -279 +300 @@
 
     // have to check for the zero size channel case
-    if ( size == 0 && !prods`isEmpty ) {
-        memcpy((void *)&retval, (void *)prods`first.elem, sizeof(T));
-        wake_one( prods );
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, retval );
         unlock( mutex_lock );
         return retval;
@@ -287 +308 @@
 
     // wait if buffer is empty, work will be completed by someone else
-    if (count == 0) {
+    if ( count == 0 ) {
         #ifdef CHAN_STATS
-        blocks++;
+        c_blocks++;
         #endif
         // check for if woken due to close
@@ -299 +320 @@
     // Remove from buffer
     __do_remove( chan, retval );
-
     unlock( mutex_lock );
     return retval;
 }
+
+///////////////////////////////////////////////////////////////////////////////////////////
+// The following is support for waituntil (select) statements
+///////////////////////////////////////////////////////////////////////////////////////////
+static inline bool unregister_chan( channel(T) & chan, select_node & node ) with(chan) {
+    if ( !node`isListed && !node.park_counter ) return false; // handle special OR case
+    lock( mutex_lock );
+    if ( node`isListed ) { // op wasn't performed
+        remove( node );
+        unlock( mutex_lock );
+        return false;
+    }
+    unlock( mutex_lock );
+
+    // only return true when not special OR case, not exceptional calse and status is SAT
+    return ( node.extra == 0p || !node.park_counter ) ? false : *node.clause_status == __SELECT_SAT;
+}
+
+// type used by select statement to capture a chan read as the selected operation
+struct chan_read {
+    T & ret;
+    channel(T) & chan;
+};
+
+static inline void ?{}( chan_read(T) & cr, channel(T) & chan, T & ret ) {
+    &cr.chan = &chan;
+    &cr.ret = &ret;
+}
+static inline chan_read(T) ?<<?( T & ret, channel(T) & chan ) { chan_read(T) cr{ chan, ret }; return cr; }
+
+static inline void __handle_select_closed_read( chan_read(T) & this, select_node & node ) with(this.chan, this) {
+    __closed_remove( chan, ret );
+    // if we get here then the insert succeeded
+    __make_select_node_available( node );
+}
+
+static inline bool register_select( chan_read(T) & this, select_node & node ) with(this.chan, this) {
+    lock( mutex_lock );
+    node.extra = &ret; // set .extra so that if it == 0p later in on_selected it is due to channel close
+
+    #ifdef CHAN_STATS
+    if ( !closed ) c_ops++;
+    #endif
+
+    if ( !node.park_counter ) {
+        // are we special case OR and front of cons is also special case OR
+        if ( !unlikely(closed) && !prods`isEmpty && prods`first.clause_status && !prods`first.park_counter ) {
+            if ( !__make_select_node_pending( node ) ) {
+                unlock( mutex_lock );
+                return false;
+            }
+            
+            if ( __handle_waituntil_OR( prods ) ) {
+                __prods_handoff( chan, ret );
+                __make_select_node_sat( node ); // need to to mark SAT now that we know operation is done or else threads could get stuck in __mark_select_node
+                unlock( mutex_lock );
+                return true;
+            }
+            __make_select_node_unsat( node );
+        }
+        // check if we can complete operation. If so race to establish winner in special OR case
+        if ( count != 0 || !prods`isEmpty || unlikely(closed) ) {
+            if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                unlock( mutex_lock );
+                return false;
+            }
+        }
+    }
+
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __handle_select_closed_read( this, node );
+        return true;
+    }
+
+    // have to check for the zero size channel case
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, ret );
+        __set_avail_then_unlock( node, mutex_lock );
+        return true;
+    }
+
+    // wait if buffer is empty, work will be completed by someone else
+    if ( count == 0 ) {
+        #ifdef CHAN_STATS
+        c_blocks++;
+        #endif
+        
+        insert_last( cons, node );
+        unlock( mutex_lock );
+        return false;
+    }
+
+    // Remove from buffer
+    __do_remove( chan, ret );
+    __set_avail_then_unlock( node, mutex_lock );
+    return true;
+}
+static inline bool unregister_select( chan_read(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
+static inline bool on_selected( chan_read(T) & this, select_node & node ) with(this) {
+    if ( node.extra == 0p ) // check if woken up due to closed channel
+        __closed_remove( chan, ret );
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+
+// type used by select statement to capture a chan write as the selected operation
+struct chan_write {
+    T elem;
+    channel(T) & chan;
+};
+
+static inline void ?{}( chan_write(T) & cw, channel(T) & chan, T elem ) {
+    &cw.chan = &chan;
+    memcpy( (void *)&cw.elem, (void *)&elem, sizeof(T) );
+}
+static inline chan_write(T) ?>>?( T elem, channel(T) & chan ) { chan_write(T) cw{ chan, elem }; return cw; }
+
+static inline void __handle_select_closed_write( chan_write(T) & this, select_node & node ) with(this.chan, this) {
+    __closed_insert( chan, elem );
+    // if we get here then the insert succeeded
+    __make_select_node_available( node );
+}
+
+static inline bool register_select( chan_write(T) & this, select_node & node ) with(this.chan, this) {
+    lock( mutex_lock );
+    node.extra = &elem; // set .extra so that if it == 0p later in on_selected it is due to channel close
+
+    #ifdef CHAN_STATS
+    if ( !closed ) p_ops++;
+    #endif
+
+    // special OR case handling
+    if ( !node.park_counter ) {
+        // are we special case OR and front of cons is also special case OR
+        if ( !unlikely(closed) && !cons`isEmpty && cons`first.clause_status && !cons`first.park_counter ) {
+            if ( !__make_select_node_pending( node ) ) {
+                unlock( mutex_lock );
+                return false;
+            }
+            
+            if ( __handle_waituntil_OR( cons ) ) {
+                __cons_handoff( chan, elem );
+                __make_select_node_sat( node ); // need to to mark SAT now that we know operation is done or else threads could get stuck in __mark_select_node
+                unlock( mutex_lock );
+                return true;
+            }
+            __make_select_node_unsat( node );
+        }
+        // check if we can complete operation. If so race to establish winner in special OR case
+        if ( count != size || !cons`isEmpty || unlikely(closed) ) {
+            if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                unlock( mutex_lock );
+                return false;
+            }
+        }
+    }
+
+    // if closed handle
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __handle_select_closed_write( this, node );
+        return true;
+    }
+
+    // handle blocked consumer case via handoff (buffer is implicitly empty)
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( chan, elem );
+        __set_avail_then_unlock( node, mutex_lock );
+        return true;
+    }
+
+    // insert node in list if buffer is full, work will be completed by someone else
+    if ( count == size ) {
+        #ifdef CHAN_STATS
+        p_blocks++;
+        #endif
+
+        insert_last( prods, node );
+        unlock( mutex_lock );
+        return false;
+    } // if
+
+    // otherwise carry out write either via normal insert
+    __buf_insert( chan, elem );
+    __set_avail_then_unlock( node, mutex_lock );
+    return true;
+}
+static inline bool unregister_select( chan_write(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
+
+static inline bool on_selected( chan_write(T) & this, select_node & node ) with(this) { 
+    if ( node.extra == 0p ) // check if woken up due to closed channel
+        __closed_insert( chan, elem );
+
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+
 } // forall( T )
+
+

libcfa/src/concurrency/future.hfa

@@ -19 +19 @@
 #include "monitor.hfa"
 #include "select.hfa"
+#include "locks.hfa"
 
 //----------------------------------------------------------------------------
@@ -26 +27 @@
 //  future_t is lockfree and uses atomics which aren't needed given we use locks here
 forall( T ) {
-    // enum(int) { FUTURE_EMPTY = 0, FUTURE_FULFILLED = 1 }; // Enums seem to be broken so feel free to add this back afterwards
+    // enum { FUTURE_EMPTY = 0, FUTURE_FULFILLED = 1 }; // Enums seem to be broken so feel free to add this back afterwards
 
     // temporary enum replacement
@@ -44 +45 @@
     };
 
-    // C_TODO: perhaps allow exceptions to be inserted like uC++?
-
         static inline {
 
@@ -53 +52 @@
         }
 
-        void ?{}(future(T) & this) {
+        void ?{}( future(T) & this ) {
                         this.waiters{};
             this.state = FUTURE_EMPTY;
@@ -60 +59 @@
 
                 // Reset future back to original state
-                void reset(future(T) & this) with(this)
+                void reset( future(T) & this ) with(this)
         {
             lock( lock );
@@ -82 +81 @@
         void _internal_flush( future(T) & this ) with(this) {
             while( ! waiters`isEmpty ) {
+                if ( !__handle_waituntil_OR( waiters ) ) // handle special waituntil OR case
+                    break; // if handle_OR returns false then waiters is empty so break
                 select_node &s = try_pop_front( waiters );
 
-                if ( s.race_flag == 0p )
-                    // poke in result so that woken threads do not need to reacquire any locks
-                    // *(((future_node(T) &)s).my_result) = result;
+                if ( s.clause_status == 0p ) // poke in result so that woken threads do not need to reacquire any locks
                     copy_T( result, *(((future_node(T) &)s).my_result) );
-                else if ( !install_select_winner( s, &this ) ) continue;
                 
-                // only unpark if future is not selected
-                // or if it is selected we only unpark if we win the race
-                unpark( s.blocked_thread );
+                wake_one( waiters, s );
             }
         }
 
                 // Fulfil the future, returns whether or not someone was unblocked
-                bool fulfil( future(T) & this, T & val ) with(this) {
+                bool fulfil( future(T) & this, T val ) with(this) {
             lock( lock );
             if( state != FUTURE_EMPTY )
@@ -153 +149 @@
         }
 
-        void * register_select( future(T) & this, select_node & s ) with(this) {
-            lock( lock );
-
-            // future not ready -> insert select node and return 0p
+        bool register_select( future(T) & this, select_node & s ) with(this) {
+            lock( lock );
+
+            // check if we can complete operation. If so race to establish winner in special OR case
+            if ( !s.park_counter && state != FUTURE_EMPTY ) { 
+                if ( !__make_select_node_available( s ) ) { // we didn't win the race so give up on registering
+                    unlock( lock );
+                    return false;
+                }
+            }
+
+            // future not ready -> insert select node and return
             if( state == FUTURE_EMPTY ) {
                 insert_last( waiters, s );
                 unlock( lock );
-                return 0p;
-            }
-
-            // future ready and we won race to install it as the select winner return 1p
-            if ( install_select_winner( s, &this ) ) {
-                unlock( lock );
-                return 1p;
-            }
-
-            unlock( lock );
-            // future ready and we lost race to install it as the select winner
-            return 2p;
-        }
-
-        void unregister_select( future(T) & this, select_node & s ) with(this) {
+                return false;
+            }
+
+            __make_select_node_available( s );
+            unlock( lock );
+            return true;
+        }
+
+        bool unregister_select( future(T) & this, select_node & s ) with(this) {
+            if ( ! s`isListed ) return false;
             lock( lock );
             if ( s`isListed ) remove( s );
             unlock( lock );
+            return false;
         }
                 
+        bool on_selected( future(T) & this, select_node & node ) { return true; }
         }
 }
@@ -186 +187 @@
 // These futures below do not support select statements so they may not be as useful as 'future'
 //  however the 'single_future' is cheap and cheerful and is most likely more performant than 'future'
-//  since it uses raw atomics and no locks afaik
+//  since it uses raw atomics and no locks
 //
 // As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future'

libcfa/src/concurrency/invoke.h

@@ -217 +217 @@
                 struct __thread_user_link cltr_link;
 
-                // used to point to this thd's current clh node
-                volatile bool * clh_node;
-
                 struct processor * last_proc;
+
+        // ptr used during handover between blocking lists to allow for stack allocation of intrusive nodes
+        // main use case is wait-morphing to allow a different node to be used to block on condvar vs lock
+        void * link_node;
 
                 PRNG_STATE_T random_state;                                              // fast random numbers

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.
 //
@@ -79 +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 );
 }
 
@@ -117 +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;
+    }
 }
 
@@ -134 +138 @@
         recursion_count--;
         if ( recursion_count == 0 ) {
-                pop_and_set_new_owner( this );
+                pop_node( this );
         }
         unlock( lock );
@@ -147 +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
@@ -156 +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 );
@@ -167 +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;
 }
@@ -175 +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;
+}
+
+bool on_selected( blocking_lock & this, select_node & node ) { return true; }
 
 //-----------------------------------------------------------------------------
@@ -311 +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 );
         }
 
@@ -340 +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
@@ -357 +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 );
         }
 
@@ -417 +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);
         }
@@ -459 +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 );
         }
 
@@ -493 +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 );
         }
 
@@ -584 +645 @@
         return thrd != 0p;
 }
+

libcfa/src/concurrency/locks.hfa

@@ -30 +30 @@
 #include "time.hfa"
 
+#include "select.hfa"
+
 #include <fstream.hfa>
 
@@ -37 +39 @@
 #include <unistd.h>
 
-// C_TODO: cleanup this and locks.cfa
-// - appropriate separation of interface and impl
-// - clean up unused/unneeded locks
-// - 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();
+}
+
+// macros for default routine impls for is_blocking_lock trait that do not wait-morph
+
+#define DEFAULT_ON_NOTIFY( lock_type ) \
+    static inline void on_notify( lock_type & this, thread$ * t ){ unpark(t); }
+
+#define DEFAULT_ON_WAIT( lock_type ) \
+    static inline size_t on_wait( lock_type & this, __cfa_pre_park pp_fn, void * pp_datum ) { \
+        unlock( this ); \
+        pre_park_then_park( pp_fn, pp_datum ); \
+        return 0; \
+    }
+
+// on_wakeup impl if lock should be reacquired after waking up
+#define DEFAULT_ON_WAKEUP_REACQ( lock_type ) \
+    static inline void on_wakeup( lock_type & this, size_t recursion ) { lock( this ); }
+
+// on_wakeup impl if lock will not be reacquired after waking up
+#define DEFAULT_ON_WAKEUP_NO_REACQ( lock_type ) \
+    static inline void on_wakeup( lock_type & this, size_t recursion ) {}
+
+
 
 //-----------------------------------------------------------------------------
@@ -67 +109 @@
 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 ); }
+static inline bool   register_select( single_acquisition_lock & this, select_node & node ) { return register_select( (blocking_lock &)this, node ); }
+static inline bool   unregister_select( single_acquisition_lock & this, select_node & node ) { return unregister_select( (blocking_lock &)this, node ); }
+static inline bool   on_selected( single_acquisition_lock & this, select_node & node ) { return on_selected( (blocking_lock &)this, node ); }
 
 //----------
@@ -81 +126 @@
 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 ); }
+static inline bool   register_select( owner_lock & this, select_node & node ) { return register_select( (blocking_lock &)this, node ); }
+static inline bool   unregister_select( owner_lock & this, select_node & node ) { return unregister_select( (blocking_lock &)this, node ); }
+static inline bool   on_selected( owner_lock & this, select_node & node ) { return on_selected( (blocking_lock &)this, node ); }
 
 //-----------------------------------------------------------------------------
@@ -128 +176 @@
 static inline void ?{}(mcs_spin_node & this) { this.next = 0p; this.locked = true; }
 
-static inline mcs_spin_node * volatile & ?`next ( mcs_spin_node * node ) {
-        return node->next;
-}
-
 struct mcs_spin_lock {
         mcs_spin_queue queue;
@@ -137 +181 @@
 
 static inline void lock(mcs_spin_lock & l, mcs_spin_node & n) {
+    n.locked = true;
         mcs_spin_node * prev = __atomic_exchange_n(&l.queue.tail, &n, __ATOMIC_SEQ_CST);
-        n.locked = true;
-        if(prev == 0p) return;
+        if( prev == 0p ) return;
         prev->next = &n;
-        while(__atomic_load_n(&n.locked, __ATOMIC_RELAXED)) Pause();
+        while( __atomic_load_n(&n.locked, __ATOMIC_RELAXED) ) Pause();
 }
 
@@ -147 +191 @@
         mcs_spin_node * n_ptr = &n;
         if (__atomic_compare_exchange_n(&l.queue.tail, &n_ptr, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) return;
-        while (__atomic_load_n(&n.next, __ATOMIC_RELAXED) == 0p) {}
+        while (__atomic_load_n(&n.next, __ATOMIC_RELAXED) == 0p) Pause();
         n.next->locked = false;
 }
@@ -156 +200 @@
 // - Kernel thd blocking alternative to the spinlock
 // - No ownership (will deadlock on reacq)
+// - no reacq on wakeup
 struct futex_mutex {
         // lock state any state other than UNLOCKED is locked
@@ -169 +214 @@
 }
 
-static inline void  ?{}( futex_mutex & this ) with(this) { val = 0; }
-
-static inline bool internal_try_lock(futex_mutex & this, int & compare_val) with(this) {
+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) {
+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) {
+static inline void lock( futex_mutex & this ) with(this) {
         int state;
 
@@ -190 +235 @@
                 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
@@ -212 +254 @@
 }
 
-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 ) {}
+DEFAULT_ON_NOTIFY( futex_mutex )
+DEFAULT_ON_WAIT( futex_mutex )
+DEFAULT_ON_WAKEUP_NO_REACQ( futex_mutex )
 
 //-----------------------------------------------------------------------------
@@ -232 +272 @@
         int val; 
 };
-
 static inline void  ?{}( go_mutex & this ) with(this) { val = 0; }
+// static inline void ?{}( go_mutex & this, go_mutex this2 ) = void; // these don't compile correctly at the moment so they should be omitted
+// static inline void ?=?( go_mutex & this, go_mutex this2 ) = void;
 
 static inline bool internal_try_lock(go_mutex & this, int & compare_val, int new_val ) with(this) {
@@ -244 +285 @@
 
 // if this is called recursively IT WILL DEADLOCK!!!!!
-static inline void lock(go_mutex & this) with(this) {
+static inline void lock( go_mutex & this ) with( this ) {
         int state, init_state;
 
@@ -255 +296 @@
             while( !val ) { // lock unlocked
                 state = 0;
-                if (internal_try_lock(this, state, init_state)) return;
+                if ( internal_try_lock( this, state, init_state ) ) return;
             }
             for (int i = 0; i < 30; i++) Pause();
@@ -262 +303 @@
         while( !val ) { // lock unlocked
             state = 0;
-            if (internal_try_lock(this, state, init_state)) return;
+            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);
+        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
+        futex( (int*)&val, FUTEX_WAIT, 2 ); // if val is not 2 this returns with EWOULDBLOCK
     }
 }
@@ -276 +317 @@
 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;
+    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 ) {}
-
-//-----------------------------------------------------------------------------
-// CLH Spinlock
-// - No recursive acquisition
-// - Needs to be released by owner
-
-struct clh_lock {
-        volatile bool * volatile tail;
-    volatile bool * volatile head;
-};
-
-static inline void  ?{}( clh_lock & this ) { this.tail = malloc(); *this.tail = true; }
-static inline void ^?{}( clh_lock & this ) { free(this.tail); }
-
-static inline void lock(clh_lock & l) {
-        thread$ * curr_thd = active_thread();
-        *(curr_thd->clh_node) = false;
-        volatile bool * prev = __atomic_exchange_n((bool **)(&l.tail), (bool *)(curr_thd->clh_node), __ATOMIC_SEQ_CST);
-        while(!__atomic_load_n(prev, __ATOMIC_SEQ_CST)) Pause();
-    __atomic_store_n((bool **)(&l.head), (bool *)curr_thd->clh_node, __ATOMIC_SEQ_CST);
-    curr_thd->clh_node = prev;
-}
-
-static inline void unlock(clh_lock & l) {
-        __atomic_store_n((bool *)(l.head), true, __ATOMIC_SEQ_CST);
-}
-
-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 ) { lock(this); }
+        futex( (int *)&val, FUTEX_WAKE, 1 );
+}
+
+DEFAULT_ON_NOTIFY( go_mutex )
+DEFAULT_ON_WAIT( go_mutex )
+DEFAULT_ON_WAKEUP_NO_REACQ( go_mutex )
 
 //-----------------------------------------------------------------------------
@@ -334 +345 @@
         this.lock_value = 0;
 }
+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) {
+static inline bool internal_try_lock( exp_backoff_then_block_lock & this, size_t & compare_val ) with(this) {
         return __atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
 }
 
-static inline bool try_lock(exp_backoff_then_block_lock & this) { size_t compare_val = 0; return internal_try_lock(this, compare_val); }
-
-static inline bool try_lock_contention(exp_backoff_then_block_lock & this) with(this) {
-        return !__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE);
-}
-
-static inline bool block(exp_backoff_then_block_lock & this) with(this) {
+static inline bool try_lock( exp_backoff_then_block_lock & this ) { size_t compare_val = 0; return internal_try_lock( this, compare_val ); }
+
+static inline bool try_lock_contention( exp_backoff_then_block_lock & this ) with(this) {
+        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 );
     if (__atomic_load_n( &lock_value, __ATOMIC_SEQ_CST) != 2) {
@@ -359 +372 @@
 }
 
-static inline void lock(exp_backoff_then_block_lock & this) with(this) {
+static inline void lock( exp_backoff_then_block_lock & this ) with(this) {
         size_t compare_val = 0;
         int spin = 4;
@@ -378 +391 @@
 }
 
-static inline void unlock(exp_backoff_then_block_lock & this) with(this) {
+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 );
@@ -386 +399 @@
 }
 
-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); return 0; }
-static inline void on_wakeup(exp_backoff_then_block_lock & this, size_t recursion ) { lock(this); }
+DEFAULT_ON_NOTIFY( exp_backoff_then_block_lock )
+DEFAULT_ON_WAIT( exp_backoff_then_block_lock )
+DEFAULT_ON_WAKEUP_REACQ( exp_backoff_then_block_lock )
 
 //-----------------------------------------------------------------------------
@@ -418 +431 @@
 
 // if this is called recursively IT WILL DEADLOCK!!!!!
-static inline void lock(fast_block_lock & this) with(this) {
+static inline void lock( fast_block_lock & this ) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
         if ( held ) {
@@ -430 +443 @@
 }
 
-static inline void unlock(fast_block_lock & this) with(this) {
+static inline void unlock( fast_block_lock & this ) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
         /* paranoid */ verifyf( held != false, "Attempt to release lock %p that isn't held", &this );
@@ -439 +452 @@
 }
 
-static inline void on_notify(fast_block_lock & this, struct thread$ * t ) with(this) {
+static inline void on_notify( fast_block_lock & this, struct thread$ * t ) with(this) {
     lock( lock __cfaabi_dbg_ctx2 );
     insert_last( blocked_threads, *t );
     unlock( lock );
 }
-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 ) { }
+DEFAULT_ON_WAIT( fast_block_lock )
+DEFAULT_ON_WAKEUP_NO_REACQ( fast_block_lock )
 
 //-----------------------------------------------------------------------------
@@ -456 +469 @@
 struct simple_owner_lock {
         // List of blocked threads
-        dlist( thread$ ) blocked_threads;
+        dlist( select_node ) blocked_threads;
 
         // Spin lock used for mutual exclusion
@@ -477 +490 @@
 static inline void ?=?( simple_owner_lock & this, simple_owner_lock this2 ) = void;
 
-static inline void lock(simple_owner_lock & this) with(this) {
-        if (owner == active_thread()) {
+static inline void lock( simple_owner_lock & this ) with(this) {
+        if ( owner == active_thread() ) {
                 recursion_count++;
                 return;
@@ -484 +497 @@
         lock( lock __cfaabi_dbg_ctx2 );
 
-        if (owner != 0p) {
-                insert_last( blocked_threads, *active_thread() );
+        if ( owner != 0p ) {
+        select_node node;
+                insert_last( blocked_threads, node );
                 unlock( lock );
                 park( );
@@ -495 +509 @@
 }
 
-// TODO: fix duplicate def issue and bring this back
-// void pop_and_set_new_owner( simple_owner_lock & this ) with( this ) {
-        // thread$ * t = &try_pop_front( blocked_threads );
-        // owner = t;
-        // recursion_count = ( t ? 1 : 0 );
-        // unpark( t );
-// }
-
-static inline void unlock(simple_owner_lock & this) with(this) {
+static inline void pop_node( simple_owner_lock & this ) with(this) {
+    __handle_waituntil_OR( blocked_threads );
+    select_node * node = &try_pop_front( blocked_threads );
+    if ( node ) {
+        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;
+    }
+}
+
+static inline void unlock( simple_owner_lock & this ) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
         /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
@@ -510 +530 @@
         recursion_count--;
         if ( recursion_count == 0 ) {
-                // pop_and_set_new_owner( this );
-                thread$ * t = &try_pop_front( blocked_threads );
-                owner = t;
-                recursion_count = ( t ? 1 : 0 );
-                unpark( t );
+                pop_node( this );
         }
         unlock( lock );
 }
 
-static inline void on_notify(simple_owner_lock & this, struct thread$ * t ) with(this) {
+static inline void on_notify( simple_owner_lock & this, thread$ * t ) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
         // lock held
         if ( owner != 0p ) {
-                insert_last( blocked_threads, *t );
+                insert_last( blocked_threads, *(select_node *)t->link_node );
         }
         // lock not held
@@ -534 +550 @@
 }
 
-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 );
@@ -541 +557 @@
         size_t ret = recursion_count;
 
-        // pop_and_set_new_owner( this );
-
-        thread$ * t = &try_pop_front( blocked_threads );
-        owner = t;
-        recursion_count = ( t ? 1 : 0 );
-        unpark( t );
-
+        pop_node( this );
+
+    select_node node;
+    active_thread()->link_node = (void *)&node;
         unlock( lock );
+
+    pre_park_then_park( pp_fn, pp_datum );
+
         return ret;
 }
 
-static inline void on_wakeup(simple_owner_lock & this, size_t recursion ) with(this) { recursion_count = recursion; }
+static inline void on_wakeup( simple_owner_lock & this, size_t recursion ) with(this) { recursion_count = recursion; }
+
+// waituntil() support
+static inline bool register_select( simple_owner_lock & this, select_node & node ) with(this) {
+    lock( lock __cfaabi_dbg_ctx2 );
+
+    // check if we can complete operation. If so race to establish winner in special OR case
+    if ( !node.park_counter && ( owner == active_thread() || owner == 0p ) ) {
+        if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+           unlock( lock );
+           return false;
+        }
+    }
+
+    if ( owner == active_thread() ) {
+                recursion_count++;
+        if ( node.park_counter ) __make_select_node_available( node );
+        unlock( lock );
+                return true;
+        }
+
+    if ( owner != 0p ) {
+                insert_last( blocked_threads, node );
+                unlock( lock );
+                return false;
+        }
+    
+        owner = active_thread();
+        recursion_count = 1;
+
+    if ( node.park_counter ) __make_select_node_available( node );
+    unlock( lock );
+    return true;
+}
+
+static inline bool unregister_select( simple_owner_lock & this, select_node & node ) with(this) {
+    lock( lock __cfaabi_dbg_ctx2 );
+    if ( node`isListed ) {
+        remove( node );
+        unlock( lock );
+        return false;
+    }
+
+    if ( owner == active_thread() ) {
+        recursion_count--;
+        if ( recursion_count == 0 ) {
+            pop_node( this );
+        }
+    }
+    unlock( lock );
+    return false;
+}
+
+static inline bool on_selected( simple_owner_lock & this, select_node & node ) { return true; }
+
 
 //-----------------------------------------------------------------------------
@@ -578 +648 @@
 
 // if this is called recursively IT WILL DEADLOCK!
-static inline void lock(spin_queue_lock & this) with(this) {
+static inline void lock( spin_queue_lock & this ) with(this) {
         mcs_spin_node node;
         lock( lock, node );
@@ -586 +656 @@
 }
 
-static inline void unlock(spin_queue_lock & this) with(this) {
+static inline void unlock( spin_queue_lock & this ) with(this) {
         __atomic_store_n(&held, false, __ATOMIC_RELEASE);
 }
 
-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 ) { lock(this); }
-
+DEFAULT_ON_NOTIFY( spin_queue_lock )
+DEFAULT_ON_WAIT( spin_queue_lock )
+DEFAULT_ON_WAKEUP_REACQ( spin_queue_lock )
 
 //-----------------------------------------------------------------------------
@@ -621 +688 @@
 
 // if this is called recursively IT WILL DEADLOCK!!!!!
-static inline void lock(mcs_block_spin_lock & this) with(this) {
+static inline void lock( mcs_block_spin_lock & this ) with(this) {
         mcs_node node;
         lock( lock, node );
@@ -633 +700 @@
 }
 
-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 ) {lock(this); }
+DEFAULT_ON_NOTIFY( mcs_block_spin_lock )
+DEFAULT_ON_WAIT( mcs_block_spin_lock )
+DEFAULT_ON_WAKEUP_REACQ( mcs_block_spin_lock )
 
 //-----------------------------------------------------------------------------
@@ -661 +728 @@
 
 // if this is called recursively IT WILL DEADLOCK!!!!!
-static inline void lock(block_spin_lock & this) with(this) {
+static inline void lock( block_spin_lock & this ) with(this) {
         lock( lock );
         while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause();
@@ -668 +735 @@
 }
 
-static inline void unlock(block_spin_lock & this) with(this) {
+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) {
+static inline void on_notify( block_spin_lock & this, struct thread$ * t ) with(this.lock) {
         // first we acquire internal fast_block_lock
         lock( lock __cfaabi_dbg_ctx2 );
@@ -686 +753 @@
         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 ) with(this) {
+DEFAULT_ON_WAIT( block_spin_lock )
+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
         while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause();
@@ -693 +760 @@
         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 );
-};
 
 //-----------------------------------------------------------------------------
@@ -714 +767 @@
 forall(L & | is_blocking_lock(L)) {
         struct info_thread;
-
-        // // for use by sequence
-        // info_thread(L) *& Back( info_thread(L) * this );
-        // info_thread(L) *& Next( info_thread(L) * this );
 }
 

libcfa/src/concurrency/mutex_stmt.hfa

@@ -15 +15 @@
 };
 
-
 struct __mutex_stmt_lock_guard {
     void ** lockarr;
@@ -30 +29 @@
 
 forall(L & | is_lock(L)) {
-
-    struct scoped_lock {
-        L * internal_lock;
-    };
-
-    static inline void ?{}( scoped_lock(L) & this, L & internal_lock ) {
-        this.internal_lock = &internal_lock;
-        lock(internal_lock);
-    }
-    
-    static inline void ^?{}( scoped_lock(L) & this ) with(this) {
-        unlock(*internal_lock);
-    }
-
-    static inline void * __get_mutexstmt_lock_ptr( L & this ) {
-        return &this;
-    }
-
-    static inline L __get_mutexstmt_lock_type( L & this );
-
-    static inline L __get_mutexstmt_lock_type( L * this );
+    static inline void * __get_mutexstmt_lock_ptr( L & this ) { return &this; }
+    static inline L __get_mutexstmt_lock_type( L & this ) {}
+    static inline L __get_mutexstmt_lock_type( L * this ) {}
 }

libcfa/src/concurrency/preemption.cfa

@@ -117 +117 @@
                 __cfadbg_print_buffer_decl( preemption, " KERNEL: preemption tick %lu\n", currtime.tn);
                 Duration period = node->period;
-                if( period == 0) {
+                if( period == 0 ) {
                         node->set = false;                  // Node is one-shot, just mark it as not pending
                 }

libcfa/src/concurrency/select.hfa

@@ +1 @@
+//
+// Cforall Version 1.0.0 Copyright (C) 2021 University of Waterloo
+//
+// The contents of this file are covered under the licence agreement in the
+// file "LICENCE" distributed with Cforall.
+//
+// channel.hfa -- LIBCFATHREAD
+// Runtime locks that used with the runtime thread system.
+//
+// Author           : Colby Alexander Parsons
+// Created On       : Thu Jan 21 19:46:50 2023
+// Last Modified By :
+// Last Modified On :
+// Update Count     :
+//
+
 #pragma once
 
 #include "containers/list.hfa"
-#include <stdint.h>
-#include <kernel.hfa>
-#include <locks.hfa>
+#include "alarm.hfa"
+#include "kernel.hfa"
+#include "time.hfa"
 
+struct select_node;
+
+// node status
+static const unsigned long int __SELECT_UNSAT = 0;
+static const unsigned long int __SELECT_PENDING = 1; // used only by special OR case
+static const unsigned long int __SELECT_SAT = 2;
+static const unsigned long int __SELECT_RUN = 3;
+
+// these are used inside the compiler to aid in code generation
+static inline bool __CFA_has_clause_run( unsigned long int status ) { return status == __SELECT_RUN; }
+static inline void __CFA_maybe_park( int * park_counter ) {
+    if ( __atomic_sub_fetch( park_counter, 1, __ATOMIC_SEQ_CST) < 0 )
+        park();
+}
+
+// node used for coordinating waituntil synchronization
 struct select_node {
+    int * park_counter;                 // If this is 0p then the node is in a special OR case waituntil
+    unsigned long int * clause_status;  // needs to point at ptr sized location, if this is 0p then node is not part of a waituntil
+
+    void * extra;                       // used to store arbitrary data needed by some primitives
+
     thread$ * blocked_thread;
-    void ** race_flag;
     inline dlink(select_node);
 };
 P9_EMBEDDED( select_node, dlink(select_node) )
 
-void ?{}( select_node & this ) {
-    this.blocked_thread = 0p;
-    this.race_flag = 0p;
+static inline void ?{}( select_node & this ) {
+    this.blocked_thread = active_thread();
+    this.clause_status = 0p;
+    this.park_counter = 0p;
+    this.extra = 0p;
 }
 
-void ?{}( select_node & this, thread$ * blocked_thread ) {
+static inline void ?{}( select_node & this, thread$ * blocked_thread ) {
     this.blocked_thread = blocked_thread;
-    this.race_flag = 0p;
+    this.clause_status = 0p;
+    this.park_counter = 0p;
+    this.extra = 0p;
 }
 
-void ?{}( select_node & this, thread$ * blocked_thread, void ** race_flag ) {
+static inline void ?{}( select_node & this, thread$ * blocked_thread, void * extra ) {
     this.blocked_thread = blocked_thread;
-    this.race_flag = race_flag;
+    this.clause_status = 0p;
+    this.park_counter = 0p;
+    this.extra = extra;
 }
+static inline void ^?{}( select_node & this ) {}
 
-void ^?{}( select_node & this ) {}
+// this is used inside the compiler to aid in code generation
+static inline unsigned long int * __get_clause_status( select_node & s ) { return s.clause_status; }
 
+// this is used inside the compiler to attempt to establish an else clause as a winner in the OR special case race
+static inline bool __select_node_else_race( select_node & this ) with( this ) {
+    unsigned long int cmp_status = __SELECT_UNSAT;
+    return *clause_status == 0 
+            && __atomic_compare_exchange_n( clause_status, &cmp_status, __SELECT_SAT, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST );
+}
 
 //-----------------------------------------------------------------------------
 // is_selectable
-trait is_selectable(T & | sized(T)) {
-    // For registering a select on a selectable concurrency primitive
-    // return 0p if primitive not accessible yet
-    // return 1p if primitive gets acquired
-    // return 2p if primitive is accessible but some other primitive won the race
-    // C_TODO: add enum for return values
-    void * register_select( T &, select_node & );
+forall(T & | sized(T))
+trait is_selectable {
+    // For registering a select stmt on a selectable concurrency primitive
+    // Returns bool that indicates if operation is already SAT
+    bool register_select( T &, select_node & );
 
-    void unregister_select( T &, select_node &  );
+    // For unregistering a select stmt on a selectable concurrency primitive
+    // If true is returned then the corresponding code block is run (only in non-special OR case and only if node status is not RUN)
+    bool unregister_select( T &, select_node &  );
+
+    // This routine is run on the selecting thread prior to executing the statement corresponding to the select_node
+    //    passed as an arg to this routine
+    // If on_selected returns false, the statement is not run, if it returns true it is run.
+    bool on_selected( T &, select_node & );
 };
 
-static inline bool install_select_winner( select_node & this, void * primitive_ptr ) with(this) {
-    // temporary needed for atomic instruction
-    void * cmp_flag = 0p;
-    
-    // if we dont win the selector race we need to potentially 
-    //   ignore this node and move to the next one so we return accordingly
-    if ( *race_flag != 0p || 
-        !__atomic_compare_exchange_n(
-            race_flag, 
-            &cmp_flag, 
-            primitive_ptr, 
-            false,
-            __ATOMIC_SEQ_CST,
-            __ATOMIC_SEQ_CST
-        )
-    ) return false; // lost race and some other node triggered select
-    return true; // won race so this node is what the select proceeds with
+//=============================================================================================
+// Waituntil Helpers
+//=============================================================================================
+
+// used for the 2-stage avail needed by the special OR case
+static inline bool __mark_select_node( select_node & this, unsigned long int val ) with( this ) {
+    /* paranoid */ verify( park_counter == 0p );
+    /* paranoid */ verify( clause_status != 0p );
+
+    unsigned long int cmp_status = __SELECT_UNSAT;
+    while( !__atomic_compare_exchange_n( clause_status, &cmp_status, val, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) {
+        if ( cmp_status != __SELECT_PENDING ) return false;
+        cmp_status = __SELECT_UNSAT;
+    }
+    return true;
 }
+
+static inline void __make_select_node_unsat( select_node & this ) with( this ) {
+    __atomic_store_n( clause_status, __SELECT_UNSAT, __ATOMIC_SEQ_CST );
+}
+static inline void __make_select_node_sat( select_node & this ) with( this ) {
+    __atomic_store_n( clause_status, __SELECT_SAT, __ATOMIC_SEQ_CST );
+}
+
+static inline bool __make_select_node_pending( select_node & this ) with( this ) {
+    return __mark_select_node( this, __SELECT_PENDING );
+}
+
+// when a primitive becomes available it calls the following routine on it's node to update the select state:
+// return true if we want to unpark the thd
+static inline bool __make_select_node_available( select_node & this ) with( this ) {
+    /* paranoid */ verify( clause_status != 0p );
+    if( !park_counter )
+        return __mark_select_node( this, (unsigned long int)&this );
+
+    unsigned long int cmp_status = __SELECT_UNSAT;
+
+    return *clause_status == 0 // C_TODO might not need a cmp_xchg in non special OR case
+        && __atomic_compare_exchange_n( clause_status, &cmp_status, __SELECT_SAT, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) // can maybe just use atomic write
+        && !__atomic_add_fetch( park_counter, 1, __ATOMIC_SEQ_CST);
+}
+
+// Handles the special OR case of the waituntil statement
+// Since only one select node can win in the OR case, we need to race to set the node available BEFORE
+//    performing the operation since if we lose the race the operation should not be performed as it will be lost
+// Returns true if execution can continue normally and false if the queue has now been drained
+static inline bool __handle_waituntil_OR( dlist( select_node ) & queue ) {
+    if ( queue`isEmpty ) return false;
+    if ( queue`first.clause_status && !queue`first.park_counter ) {
+        while ( !queue`isEmpty ) {
+            // if node not a special OR case or if we win the special OR case race break
+            if ( !queue`first.clause_status || queue`first.park_counter || __make_select_node_available( queue`first ) )
+                return true;
+            // otherwise we lost the special OR race so discard node
+            try_pop_front( queue );
+        }
+        return false;
+    }
+    return true;
+}
+
+// wake one thread from the list
+static inline void wake_one( dlist( select_node ) & queue, select_node & popped ) {
+    if ( !popped.clause_status                              // normal case, node is not a select node
+        || ( popped.clause_status && !popped.park_counter ) // If popped link is special case OR selecting unpark but don't call __make_select_node_available
+        || __make_select_node_available( popped ) )         // check if popped link belongs to a selecting thread
+        unpark( popped.blocked_thread );
+}
+
+static inline void wake_one( dlist( select_node ) & queue ) { wake_one( queue, try_pop_front( queue ) ); }
+
+static inline void setup_clause( select_node & this, unsigned long int * clause_status, int * park_counter ) {
+    this.blocked_thread = active_thread();
+    this.clause_status = clause_status;
+    this.park_counter = park_counter;
+}
+
+// waituntil ( timeout( ... ) ) support
+struct select_timeout_node {
+    alarm_node_t a_node;
+    select_node * s_node;
+};
+void ?{}( select_timeout_node & this, Duration duration, Alarm_Callback callback );
+void ^?{}( select_timeout_node & this );
+void timeout_handler_select_cast( alarm_node_t & node );
+
+// Selectable trait routines
+bool register_select( select_timeout_node & this, select_node & node );
+bool unregister_select( select_timeout_node & this, select_node & node );
+bool on_selected( select_timeout_node & this, select_node & node );
+
+// Gateway routines to waituntil on duration
+select_timeout_node timeout( Duration duration );
+select_timeout_node sleep( Duration duration );

libcfa/src/concurrency/thread.cfa

@@ -53 +53 @@
         preferred = ready_queue_new_preferred();
         last_proc = 0p;
+    link_node = 0p;
         PRNG_SET_SEED( random_state, __global_random_mask ? __global_random_prime : __global_random_prime ^ rdtscl() );
         #if defined( __CFA_WITH_VERIFY__ )
@@ -59 +60 @@
         #endif
 
-        clh_node = malloc( );
-        *clh_node = false;
-
         doregister(curr_cluster, this);
         monitors{ &self_mon_p, 1, (fptr_t)0 };
@@ -70 +68 @@
                 canary = 0xDEADDEADDEADDEADp;
         #endif
-        free(clh_node);
         unregister(curr_cluster, this);
         ^self_cor{};