Changeset 6b33e89 for libcfa/src/concurrency/select.hfa

Timestamp:

Apr 25, 2025, 7:39:09 AM (5 months ago)

Author:

Peter A. Buhr <pabuhr@…>

Branches:

master

Children:

65bd3c2

Parents:

b195498

Message:

change backquote call to regular call

File:

• libcfa/src/concurrency/select.hfa (modified) (6 diffs)

libcfa/src/concurrency/select.hfa

@@ -10 +10 @@
 // Author           : Colby Alexander Parsons
 // Created On       : Thu Jan 21 19:46:50 2023
-// Last Modified By : Kyoung Seo
-// Last Modified On : Wed Mar 19 12:00:00 2025
-// Update Count     : 1
+// Last Modified By : Peter A. Buhr
+// Last Modified On : Fri Apr 25 07:31:26 2025
+// Update Count     : 5
 //
 
@@ -33 +33 @@
 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();
+        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;
-    inline dlink(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;
+        inline dlink(select_node);
 };
 P9_EMBEDDED( select_node, dlink(select_node) )
 
 static inline void ?{}( select_node & this ) {
-    this.blocked_thread = active_thread();
-    this.clause_status = 0p;
-    this.park_counter = 0p;
-    this.extra = 0p;
+        this.blocked_thread = active_thread();
+        this.clause_status = 0p;
+        this.park_counter = 0p;
+        this.extra = 0p;
 }
 
 static inline void ?{}( select_node & this, thread$ * blocked_thread ) {
-    this.blocked_thread = blocked_thread;
-    this.clause_status = 0p;
-    this.park_counter = 0p;
-    this.extra = 0p;
+        this.blocked_thread = blocked_thread;
+        this.clause_status = 0p;
+        this.park_counter = 0p;
+        this.extra = 0p;
 }
 
 static inline void ?{}( select_node & this, thread$ * blocked_thread, void * extra ) {
-    this.blocked_thread = blocked_thread;
-    this.clause_status = 0p;
-    this.park_counter = 0p;
-    this.extra = extra;
+        this.blocked_thread = blocked_thread;
+        this.clause_status = 0p;
+        this.park_counter = 0p;
+        this.extra = extra;
 }
 static inline void ^?{}( select_node & this ) {}
@@ -76 +76 @@
 // 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 );
+        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 );
 }
 
@@ -85 +85 @@
 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 & );
-
-    // 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 true is returned proceed as normal, if false is returned the statement is skipped
-    bool on_selected( T &, select_node & );
+        // 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 & );
+
+        // 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 true is returned proceed as normal, if false is returned the statement is skipped
+        bool on_selected( T &, select_node & );
 };
 // Used inside the compiler to allow for overloading on return type for operations such as '?<<?' for channels
@@ -107 +107 @@
 
 static inline void __make_select_node_unsat( select_node & this ) with( this ) {
-    __atomic_store_n( clause_status, __SELECT_UNSAT, __ATOMIC_SEQ_CST );
+        __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 );
+        __atomic_store_n( clause_status, __SELECT_SAT, __ATOMIC_SEQ_CST );
 }
 
 // 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;
+        /* 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;
 }
 
 // used for the 2-stage avail by the thread who owns a pending node
 static inline bool __pending_set_other( select_node & other, select_node & mine, unsigned long int val ) with( other ) {
-    /* 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;
-
-        // toggle current status flag to avoid starvation/deadlock
-        __make_select_node_unsat( mine );
-        cmp_status = __SELECT_UNSAT;
-        if ( !__atomic_compare_exchange_n( mine.clause_status, &cmp_status, __SELECT_PENDING, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) )
-            return false;
-        cmp_status = __SELECT_UNSAT;
-    }
-    return true;
+        /* 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;
+
+                // toggle current status flag to avoid starvation/deadlock
+                __make_select_node_unsat( mine );
+                cmp_status = __SELECT_UNSAT;
+                if ( ! __atomic_compare_exchange_n( mine.clause_status, &cmp_status, __SELECT_PENDING, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) )
+                        return false;
+                cmp_status = __SELECT_UNSAT;
+        }
+        return true;
 }
 
 static inline bool __make_select_node_pending( select_node & this ) with( this ) {
-    return __mark_select_node( this, __SELECT_PENDING );
+        return __mark_select_node( this, __SELECT_PENDING );
 }
 
@@ -153 +153 @@
 // 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);
+        /* 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
+// 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;
+        if ( isEmpty( queue ) ) return false;
+        if ( first( queue ).clause_status && ! first( queue ).park_counter ) {
+                while ( ! isEmpty( queue ) ) {
+                        // if node not a special OR case or if we win the special OR case race break
+                        if ( ! first( queue ).clause_status || first( queue ).park_counter || __make_select_node_available( first( queue ) ) )
+                                return true;
+                        // otherwise we lost the special OR race so discard node
+                        remove_first( 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 ) ); }
+        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, remove_first( 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;
+        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;
+        alarm_node_t a_node;
+        select_node * s_node;
 };
 void ?{}( select_timeout_node & this, Duration duration, Alarm_Callback callback );