source: libcfa/src/concurrency/future.hfa @ 509ec82

//
// 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 : Fri Nov 29 16:07:34 2024
// Update Count     : 30
//

#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;
                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.state = FUTURE_EMPTY;
            this.lock{};
                }

                // Reset future back to original state
                void reset( future(T) & this ) with(this)
        {
            lock( lock );
            if( ! waiters`isEmpty )
                abort("Attempting to reset a future with blocked waiters");
            state = FUTURE_EMPTY;
            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));
        }

        // internal helper to signal waiters off of the future
        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.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 );
            }
        }

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

            bool ret_val = ! waiters`isEmpty;
            state = FUTURE_FULFILLED;
                        _internal_flush( this );
            unlock( lock );
            return ret_val;
                }

                // 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 ) {
            lock( lock );
            T ret_val;
            if( state == FUTURE_FULFILLED ) {
                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( );

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

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

//--------------------------------------------------------------------------------------------------------
// 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&)this );
                }

                // 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&)this ) != 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&)this );
                                $first_done( this );
                        }

                        return [this.result, w];
                }

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