//
// 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 & Peter Buhr
// Created On       : Wed Jan 06 17:33:18 2021
// Last Modified By : Peter A. Buhr
// Last Modified On : Mon Nov 17 08:58:38 2025
// Update Count     : 164
//

#pragma once

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

//--------------------------------------------------------------------------------------------------------
// future does not use future_t as it needs a lock to support multiple consumers.  future_t is lockfree
// and uses atomics which are not needed.
//--------------------------------------------------------------------------------------------------------

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 ) with( this ) {
 			waiters{};
 			except = 0p;
 			state = FUTURE_EMPTY;
 			lock{};
 		}

		void ^?{}( future(T) & this ) with( this ) {
			free( 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; }
	}
}

//--------------------------------------------------------------------------------------------------------
// future_rc uses reference counting to eliminate explicit storage-management and support the waituntil
// statement.
//--------------------------------------------------------------------------------------------------------

forall( T ) {
	struct future_rc_impl$ {
		futex_mutex lock;								// concurrent protection
		size_t refCnt;									// number of references to future
		future(T) fut;									// underlying future
	}; // future_rc_impl$

	static inline {
		void incRef$( future_rc_impl$( T ) & impl ) with( impl ) {
			__atomic_fetch_add( &refCnt, 1, __ATOMIC_RELAXED );
//			lock( lock );
//			refCnt += 1;
//			unlock( lock );
		} // incRef$

		bool decRef$( future_rc_impl$( T ) & impl ) with( impl ) {
			return __atomic_fetch_add( &refCnt, -1, __ATOMIC_RELAXED ) == 1;
			// lock( lock );
			// refCnt -= 1;
			// bool ret = refCnt == 0;
			// unlock( lock );
			// return ret;
		} // decRef$

		void ?{}( future_rc_impl$( T ) & frc ) with( frc ) {
			lock{};										// intialization
			refCnt = 1;
		} // ?{}

		void ^?{}( future_rc_impl$( T ) & frc ) with( frc ) {
			decRef$( frc );
		} // ^?{}
	} // static inline
	
	struct future_rc {
		future_rc_impl$(T) * impl;		
	}; // future_rc
	__CFA_SELECT_GET_TYPE( future_rc(T) );
		
	static inline {
		void ?{}( future_rc( T ) & frc ) with( frc ) {
			impl = new();
		} // ?{}

		void ?{}( future_rc( T ) & to, future_rc( T ) & from ) with( to ) {
			impl = from.impl;							// point at new impl
			incRef$( *impl );
		} // ?{}

		void ^?{}( future_rc( T ) & frc ) with( frc ) {
			if ( decRef$( *impl ) ) { delete( impl ); impl = 0p; }
		} // ^?{}

		future_rc( T ) & ?=?( future_rc( T ) & lhs, future_rc( T ) & rhs ) with( lhs ) {
		  if ( impl == rhs.impl ) return lhs;			// self assignment ?
			if ( decRef$( *impl ) ) { delete( impl ); impl = 0p; } // no references => delete current impl
			impl = rhs.impl;							// point at new impl
			incRef$( *impl );							//   and increment reference count
			return lhs;
		} // ?+?

		bool register_select( future_rc(T) & this, select_node & s ) with( this ) {
			return register_select( this.impl->fut, s );
		}

		bool unregister_select( future_rc(T) & this, select_node & s ) with( this ) {
			return unregister_select( this.impl->fut, s );
		}

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

		// USED BY CLIENT

		bool available( future_rc( T ) & frc ) { return available( frc.impl->fut ); } // future result available ?

		bool fulfil( future_rc(T) & frc, T val ) with( frc ) { return fulfil( impl->fut, val ); }
		bool ?()( future_rc(T) & frc, T val ) { return fulfil( frc, val ); } // alternate interface

		int ?==?( future_rc( T ) & lhs, future_rc( T ) & rhs ) { return lhs.impl == rhs.impl; } // referential equality

		// USED BY SERVER

		T get( future_rc(T) & frc ) with( frc ) { return get( impl->fut ); }
		T ?()( future_rc(T) & frc ) with( frc ) { return get( frc ); } // alternate interface

		bool fulfil( future_rc(T) & frc, exception_t * ex ) with( frc ) { return fulfil( impl->fut, ex ); }
		bool ?()( future_rc(T) & frc, exception_t * ex ) { return fulfil( frc, ex ); } // alternate interface

		void reset( future_rc(T) & frc ) with( frc ) { reset( impl->fut ); } // mark future as empty (for reuse)
	} // static inline
} // forall( T )

//--------------------------------------------------------------------------------------------------------
// These futures below do not support waituntil 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 waituntil statement.
//--------------------------------------------------------------------------------------------------------

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