source: libcfa/src/concurrency/select.hfa @ 1d66a91

Last change on this file since 1d66a91 was bdbb448, checked in by caparsons <caparson@…>, 16 months ago

updated documentation related to waituntil changes

  • Property mode set to 100644
File size: 9.2 KB
RevLine 
[e23b3ce]1//
2// Cforall Version 1.0.0 Copyright (C) 2021 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// channel.hfa -- LIBCFATHREAD
8// Runtime locks that used with the runtime thread system.
9//
10// Author           : Colby Alexander Parsons
11// Created On       : Thu Jan 21 19:46:50 2023
12// Last Modified By :
13// Last Modified On :
14// Update Count     :
15//
16
[5e4a830]17#pragma once
18
[339e30a]19#include "containers/list.hfa"
[e23b3ce]20#include "alarm.hfa"
[beeff61e]21#include "kernel.hfa"
[c4f411e]22#include "time.hfa"
[339e30a]23
[beeff61e]24struct select_node;
25
26// node status
27static const unsigned long int __SELECT_UNSAT = 0;
[c4f411e]28static const unsigned long int __SELECT_PENDING = 1; // used only by special OR case
29static const unsigned long int __SELECT_SAT = 2;
30static const unsigned long int __SELECT_RUN = 3;
[beeff61e]31
[c4f411e]32// these are used inside the compiler to aid in code generation
[beeff61e]33static inline bool __CFA_has_clause_run( unsigned long int status ) { return status == __SELECT_RUN; }
34static inline void __CFA_maybe_park( int * park_counter ) {
35    if ( __atomic_sub_fetch( park_counter, 1, __ATOMIC_SEQ_CST) < 0 )
36        park();
37}
38
39// node used for coordinating waituntil synchronization
[339e30a]40struct select_node {
[beeff61e]41    int * park_counter;                 // If this is 0p then the node is in a special OR case waituntil
42    unsigned long int * clause_status;  // needs to point at ptr sized location, if this is 0p then node is not part of a waituntil
43
44    void * extra;                       // used to store arbitrary data needed by some primitives
45
[339e30a]46    thread$ * blocked_thread;
47    inline dlink(select_node);
48};
49P9_EMBEDDED( select_node, dlink(select_node) )
50
[beeff61e]51static inline void ?{}( select_node & this ) {
52    this.blocked_thread = active_thread();
53    this.clause_status = 0p;
54    this.park_counter = 0p;
55    this.extra = 0p;
[339e30a]56}
57
[beeff61e]58static inline void ?{}( select_node & this, thread$ * blocked_thread ) {
[339e30a]59    this.blocked_thread = blocked_thread;
[beeff61e]60    this.clause_status = 0p;
61    this.park_counter = 0p;
62    this.extra = 0p;
[339e30a]63}
64
[beeff61e]65static inline void ?{}( select_node & this, thread$ * blocked_thread, void * extra ) {
[339e30a]66    this.blocked_thread = blocked_thread;
[beeff61e]67    this.clause_status = 0p;
68    this.park_counter = 0p;
69    this.extra = extra;
[339e30a]70}
[beeff61e]71static inline void ^?{}( select_node & this ) {}
[339e30a]72
[c4f411e]73// this is used inside the compiler to aid in code generation
[beeff61e]74static inline unsigned long int * __get_clause_status( select_node & s ) { return s.clause_status; }
[339e30a]75
[c4f411e]76// this is used inside the compiler to attempt to establish an else clause as a winner in the OR special case race
77static inline bool __select_node_else_race( select_node & this ) with( this ) {
78    unsigned long int cmp_status = __SELECT_UNSAT;
79    return *clause_status == 0
80            && __atomic_compare_exchange_n( clause_status, &cmp_status, __SELECT_SAT, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST );
81}
82
[339e30a]83//-----------------------------------------------------------------------------
84// is_selectable
[beeff61e]85forall(T & | sized(T))
86trait is_selectable {
87    // For registering a select stmt on a selectable concurrency primitive
88    // Returns bool that indicates if operation is already SAT
89    bool register_select( T &, select_node & );
90
91    // For unregistering a select stmt on a selectable concurrency primitive
92    // 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)
[70a4ed5]93    bool unregister_select( T &, select_node & );
[beeff61e]94
95    // This routine is run on the selecting thread prior to executing the statement corresponding to the select_node
96    //    passed as an arg to this routine
[bdbb448]97    // If on_selected returns false, the statement is not run if the on_selected call occured as
98    //    part of a statement block that only ran due to an unregister_select that returned true
99    //    In all other executions the return value is ignored.
[b93bf85]100    bool on_selected( T &, select_node & );
[339e30a]101};
102
[c4f411e]103//=============================================================================================
104// Waituntil Helpers
105//=============================================================================================
106
[6f774be]107static inline void __make_select_node_unsat( select_node & this ) with( this ) {
108    __atomic_store_n( clause_status, __SELECT_UNSAT, __ATOMIC_SEQ_CST );
109}
110static inline void __make_select_node_sat( select_node & this ) with( this ) {
111    __atomic_store_n( clause_status, __SELECT_SAT, __ATOMIC_SEQ_CST );
112}
113
[c4f411e]114// used for the 2-stage avail needed by the special OR case
115static inline bool __mark_select_node( select_node & this, unsigned long int val ) with( this ) {
116    /* paranoid */ verify( park_counter == 0p );
117    /* paranoid */ verify( clause_status != 0p );
118
[beeff61e]119    unsigned long int cmp_status = __SELECT_UNSAT;
[c4f411e]120    while( !__atomic_compare_exchange_n( clause_status, &cmp_status, val, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) {
121        if ( cmp_status != __SELECT_PENDING ) return false;
122        cmp_status = __SELECT_UNSAT;
123    }
124    return true;
125}
126
[6f774be]127// used for the 2-stage avail by the thread who owns a pending node
128static inline bool __pending_set_other( select_node & other, select_node & mine, unsigned long int val ) with( other ) {
129    /* paranoid */ verify( park_counter == 0p );
130    /* paranoid */ verify( clause_status != 0p );
131
132    unsigned long int cmp_status = __SELECT_UNSAT;
133    while( !__atomic_compare_exchange_n( clause_status, &cmp_status, val, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) {
134        if ( cmp_status != __SELECT_PENDING )
135            return false;
136
137        // toggle current status flag to avoid starvation/deadlock
138        __make_select_node_unsat( mine );
139        cmp_status = __SELECT_UNSAT;
140        if ( !__atomic_compare_exchange_n( mine.clause_status, &cmp_status, __SELECT_PENDING, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) )
141            return false;
142        cmp_status = __SELECT_UNSAT;
143    }
144    return true;
[e23b3ce]145}
[c4f411e]146
147static inline bool __make_select_node_pending( select_node & this ) with( this ) {
148    return __mark_select_node( this, __SELECT_PENDING );
[beeff61e]149}
150
151// when a primitive becomes available it calls the following routine on it's node to update the select state:
152// return true if we want to unpark the thd
153static inline bool __make_select_node_available( select_node & this ) with( this ) {
[c4f411e]154    /* paranoid */ verify( clause_status != 0p );
155    if( !park_counter )
156        return __mark_select_node( this, (unsigned long int)&this );
[beeff61e]157
[c4f411e]158    unsigned long int cmp_status = __SELECT_UNSAT;
[beeff61e]159
[c4f411e]160    return *clause_status == 0 // C_TODO might not need a cmp_xchg in non special OR case
[beeff61e]161        && __atomic_compare_exchange_n( clause_status, &cmp_status, __SELECT_SAT, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) // can maybe just use atomic write
162        && !__atomic_add_fetch( park_counter, 1, __ATOMIC_SEQ_CST);
163}
164
165// Handles the special OR case of the waituntil statement
166// Since only one select node can win in the OR case, we need to race to set the node available BEFORE
167//    performing the operation since if we lose the race the operation should not be performed as it will be lost
168// Returns true if execution can continue normally and false if the queue has now been drained
169static inline bool __handle_waituntil_OR( dlist( select_node ) & queue ) {
170    if ( queue`isEmpty ) return false;
171    if ( queue`first.clause_status && !queue`first.park_counter ) {
172        while ( !queue`isEmpty ) {
173            // if node not a special OR case or if we win the special OR case race break
[e23b3ce]174            if ( !queue`first.clause_status || queue`first.park_counter || __make_select_node_available( queue`first ) )
175                return true;
[beeff61e]176            // otherwise we lost the special OR race so discard node
177            try_pop_front( queue );
178        }
179        return false;
180    }
181    return true;
182}
183
184// wake one thread from the list
185static inline void wake_one( dlist( select_node ) & queue, select_node & popped ) {
186    if ( !popped.clause_status                              // normal case, node is not a select node
187        || ( popped.clause_status && !popped.park_counter ) // If popped link is special case OR selecting unpark but don't call __make_select_node_available
188        || __make_select_node_available( popped ) )         // check if popped link belongs to a selecting thread
189        unpark( popped.blocked_thread );
[339e30a]190}
[beeff61e]191
192static inline void wake_one( dlist( select_node ) & queue ) { wake_one( queue, try_pop_front( queue ) ); }
193
194static inline void setup_clause( select_node & this, unsigned long int * clause_status, int * park_counter ) {
195    this.blocked_thread = active_thread();
196    this.clause_status = clause_status;
197    this.park_counter = park_counter;
198}
199
[e23b3ce]200// waituntil ( timeout( ... ) ) support
201struct select_timeout_node {
202    alarm_node_t a_node;
203    select_node * s_node;
204};
205void ?{}( select_timeout_node & this, Duration duration, Alarm_Callback callback );
206void ^?{}( select_timeout_node & this );
207void timeout_handler_select_cast( alarm_node_t & node );
208
209// Selectable trait routines
210bool register_select( select_timeout_node & this, select_node & node );
211bool unregister_select( select_timeout_node & this, select_node & node );
[b93bf85]212bool on_selected( select_timeout_node & this, select_node & node );
[e23b3ce]213
214// Gateway routines to waituntil on duration
215select_timeout_node timeout( Duration duration );
216select_timeout_node sleep( Duration duration );
Note: See TracBrowser for help on using the repository browser.