source: libcfa/src/concurrency/future.hfa@ 0d6fd21

//
// Cforall Version 1.0.0 Copyright (C) 2020 University of Waterloo
//
// The contents of this file are covered under the licence agreement in the
// file "LICENCE" distributed with Cforall.
//
// concurrency/future.hfa --
//
// Author           : Thierry Delisle & Peiran Hong & Colby Parsons
// Created On       : Wed Jan 06 17:33:18 2021
// Last Modified By : Peter A. Buhr
// Last Modified On : Wed Apr 23 22:41:10 2025
// Update Count     : 22
//

#pragma once

#include "bits/locks.hfa"
#include "monitor.hfa"
#include "select.hfa"
#include "locks.hfa"

//----------------------------------------------------------------------------
// future
// I don't use future_t here as I need to use a lock for this future since it supports multiple consumers.
// future_t is lockfree and uses atomics which aren't needed given we use locks here
forall( T ) {
        enum { FUTURE_EMPTY = 0, FUTURE_FULFILLED = 1 };

        struct future {
                int state;
                T result;
                exception_t * except;
                dlist( select_node ) waiters;
                futex_mutex lock;
        };
        __CFA_SELECT_GET_TYPE( future(T) );

        struct future_node {
                inline select_node;
                T * my_result;
        };

        static inline {

                void ?{}( future_node(T) & this, thread$ * blocked_thread, T * my_result ) {
                        ((select_node &)this){ blocked_thread };
                        this.my_result = my_result;
                }

                void ?{}( future(T) & this ) {
                        this.waiters{};
                        this.except = 0p;
                        this.state = FUTURE_EMPTY;
                        this.lock{};
                }

                void ^?{}( future(T) & this ) {
                        free( this.except );
                }

                // Reset future back to original state
                void reset( future(T) & this ) with(this) {
                        lock( lock );
                        if ( ! isEmpty( waiters ) )
                                abort("Attempting to reset a future with blocked waiters");
                        state = FUTURE_EMPTY;
                        free( except );
                        except = 0p;
                        unlock( lock );
                }

                // check if the future is available
                // currently no mutual exclusion because I can't see when you need this call to be synchronous or protected
                bool available( future(T) & this ) { return __atomic_load_n( &this.state, __ATOMIC_RELAXED ); }


                // memcpy wrapper to help copy values
                void copy_T( T & from, T & to ) {
                        memcpy((void *)&to, (void *)&from, sizeof(T));
                }

                bool fulfil$( future(T) & this ) with(this) {   // helper
                        bool ret_val = ! isEmpty( waiters );
                        state = FUTURE_FULFILLED;
                        while ( ! isEmpty( waiters ) ) {
                                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 = remove_first( waiters );

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

                                wake_one( waiters, s );
                        }
                        unlock( lock );
                        return ret_val;
                }

                // Fulfil the future, returns whether or not someone was unblocked
                bool fulfil( future(T) & this, T val ) with(this) {
                        lock( lock );
                        if ( state != FUTURE_EMPTY )
                                abort("Attempting to fulfil a future that has already been fulfilled");

                        copy_T( val, result );
                        return fulfil$( this );
                }

                bool ?()( future(T) & this, T val ) {                   // alternate interface
                        return fulfil( this, val );
                }

                // Load an exception to the future, returns whether or not someone was unblocked
                bool fulfil( future(T) & this, exception_t * ex ) with(this) {
                        lock( lock );
                        if ( state != FUTURE_EMPTY )
                                abort("Attempting to fulfil a future that has already been fulfilled");

                        except = ( exception_t * ) malloc( ex->virtual_table->size );
                        ex->virtual_table->copy( except, ex );
                        return fulfil$( this );
                }

                bool ?()( future(T) & this, exception_t * ex ) { // alternate interface
                        return fulfil( this, ex );
                }

                // Wait for the future to be fulfilled
                // Also return whether the thread had to block or not
                [T, bool] get( future(T) & this ) with( this ) {
                        void exceptCheck() {                                            // helper
                                if ( except ) {
                                        exception_t * ex = ( exception_t * ) alloca( except->virtual_table->size );
                                        except->virtual_table->copy( ex, except );
                                        unlock( lock );
                                        throwResume * ex;
                                }
                        }

                        lock( lock );
                        T ret_val;
                        if ( state == FUTURE_FULFILLED ) {
                                exceptCheck();
                                copy_T( result, ret_val );
                                unlock( lock );
                                return [ret_val, false];
                        }

                        future_node(T) node = { active_thread(), &ret_val };
                        insert_last( waiters, ((select_node &)node) );
                        unlock( lock );
                        park( );
                        exceptCheck();

                        return [ret_val, true];
                }

                // Wait for the future to be fulfilled
                T get( future(T) & this ) {
                        [T, bool] tt;
                        tt = get(this);
                        return tt.0;
                }

                T ?()( future(T) & this ) {                                             // alternate interface
                        return get( this );
                }

                // Gets value if it is available and returns [ val, true ]
                // otherwise returns [ default_val, false]
                // will not block
                [T, bool] try_get( future(T) & this ) with(this) {
                        lock( lock );
                        T ret_val;
                        if ( state == FUTURE_FULFILLED ) {
                                copy_T( result, ret_val );
                                unlock( lock );
                                return [ret_val, true];
                        }
                        unlock( lock );

                        return [ret_val, false];
                }

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

                        __make_select_node_available( s );
                        unlock( lock );
                        return true;
                }

                bool unregister_select( future(T) & this, select_node & s ) with(this) {
                        if ( ! isListed( s ) ) return false;
                        lock( lock );
                        if ( isListed( s ) ) remove( s );
                        unlock( lock );
                        return false;
                }

                bool on_selected( future(T) &, select_node & ) { return true; }
        }
}

//--------------------------------------------------------------------------------------------------------
// These futures below do not support select statements so they may not have as many features 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
//
// As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future'
//  since it is monitor based and also is not compatible with select statements
//--------------------------------------------------------------------------------------------------------

forall( T ) {
        struct single_future {
                inline future_t;
                T result;
        };

        static inline {
                // Reset future back to original state
                void reset(single_future(T) & this) { reset( (future_t&)this ); }

                // check if the future is available
                bool available( single_future(T) & this ) { return available( (future_t&)this ); }

                // Mark the future as abandoned, meaning it will be deleted by the server
                // This doesn't work beause of the potential need for a destructor
                // void abandon( single_future(T) & this );

                // Fulfil the future, returns whether or not someone was unblocked
                thread$ * fulfil( single_future(T) & this, T result ) {
                        this.result = result;
                        return fulfil( (future_t&)this );
                }

                // Wait for the future to be fulfilled
                // Also return whether the thread had to block or not
                [T, bool] wait( single_future(T) & this ) {
                        bool r = wait( (future_t&)this );
                        return [this.result, r];
                }

                // Wait for the future to be fulfilled
                T wait( single_future(T) & this ) {
                        [T, bool] tt;
                        tt = wait(this);
                        return tt.0;
                }
        }
}

forall( T ) {
        monitor multi_future {
                inline future_t;
                condition blocked;
                bool has_first;
                T result;
        };

        static inline {
                void ?{}(multi_future(T) & this) {
                        this.has_first = false;
                }

                bool $first( multi_future(T) & mutex this ) {
                        if ( this.has_first ) {
                                wait( this.blocked );
                                return false;
                        }

                        this.has_first = true;
                        return true;
                }

                void $first_done( multi_future(T) & mutex this ) {
                        this.has_first = false;
                        signal_all( this.blocked );
                }

                // Reset future back to original state
                void reset(multi_future(T) & mutex this) {
                        if ( this.has_first != false ) abort("Attempting to reset a multi_future with at least one blocked threads");
                        if ( !is_empty(this.blocked) ) abort("Attempting to reset a multi_future with multiple blocked threads");
                        reset( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) );
                }

                // Fulfil the future, returns whether or not someone was unblocked
                bool fulfil( multi_future(T) & this, T result ) {
                        this.result = result;
                        return fulfil( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) ) != 0p;
                }

                // Wait for the future to be fulfilled
                // Also return whether the thread had to block or not
                [T, bool] wait( multi_future(T) & this ) {
                        bool sw = $first( this );
                        bool w = !sw;
                        if ( sw ) {
                                w = wait( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) );
                                $first_done( this );
                        }

                        return [this.result, w];
                }

                // Wait for the future to be fulfilled
                T wait( multi_future(T) & this ) {
                        return wait(this).0;
                }
        }
}