source: libcfa/src/concurrency/select.hfa@ 5ba1356

//
// 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 : Peter A. Buhr
// Last Modified On : Fri Apr 25 07:31:26 2025
// Update Count     : 5
//

#pragma once

#include "collections/list.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;
        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;
}

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;
}

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;
}
static inline 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
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 & );
};
// Used inside the compiler to allow for overloading on return type for operations such as '?<<?' for channels
// YOU MUST USE THIS MACRO OR INCLUDE AN EQUIVALENT DECL FOR YOUR TYPE TO SUPPORT WAITUNTIL
#define __CFA_SELECT_GET_TYPE( typename ) typename __CFA_select_get_type( typename __CFA_t )


//=============================================================================================
// Waituntil Helpers
//=============================================================================================

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 );
}

// 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;
}

// 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;
}

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 ( 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, 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;
}

// 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 );
select_timeout_node __CFA_select_get_type( select_timeout_node this );

// Timer macro for waituntil
#define _timeout(D) (select_timeout_node) { D, timeout_handler_select_cast }