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