//
// 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 24 16:08:52 2025
// Update Count     : 222
//

#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 ) {
	// PRIVATE

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

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

	enum { FUTURE_EMPTY$ = 0, FUTURE_FULFILLED$ = 1 };

	// PUBLIC

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

	static inline {
		// PRIVATE

		bool register_select$( future(T) & fut, select_node & s ) with( fut ) { // for waituntil statement
			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) & fut, select_node & s ) with( fut ) { // for waituntil statement
		  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; } // for waituntil statement

		// PUBLIC

		// General

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

 		void ?{}( future(T) & fut ) with( fut ) {
 			except = 0p;
 			state = FUTURE_EMPTY$;
 		}

		void ^?{}( future(T) & fut ) with( fut ) {
			free( except );
		}

		// Used by Client

		// PRIVATE

		// Return a value/exception from the future.
		T get$( future(T) & fut ) with( fut ) {			// helper
			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;
				}
			}
			T ret_val;

			// LOCK ACQUIRED IN PUBLIC get
			if ( state == FUTURE_FULFILLED$ ) {
				exceptCheck();
				copy_T$( ret_val, result );
				unlock( lock );
				return ret_val;
			}

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

		// PUBLIC

		bool available( future( T ) & fut ) { return __atomic_load_n( &fut.state, __ATOMIC_RELAXED ); } // future result available ?

		// Return a value/exception from the future.
		[T, bool] get( future(T) & fut ) with( fut ) {
			lock( lock );
			bool ret = state == FUTURE_EMPTY$;
			return [ get$( fut ), ret ];
		}

		T get( future(T) & fut ) with( fut ) {
			lock( lock );
			return get$( fut );
		}
		T ?()( future(T) & fut ) { return get( fut ); }	// alternate interface

		// Non-blocking get: true => return defined value, false => value return undefined.
		[T, bool] try_get( future(T) & fut ) with( fut ) {
			lock( lock );
			T ret_val;
			if ( state == FUTURE_FULFILLED$ ) {
				copy_T$( ret_val, result );
				unlock( lock );
				return [ret_val, true];
			}
			unlock( lock );
			return [ret_val, false];
		}

		// Used by Server

		// PRIVATE

		bool fulfil$( future(T) & fut ) with( fut ) {	// 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$( *(((future_node$(T) &)s).my_result), result );

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

		// PUBLIC

		// Load a value/exception into the future, returns whether or not waiting threads.
		bool fulfil( future(T) & fut, T val ) with( fut ) {
			lock( lock );
		  if ( state != FUTURE_EMPTY$ ) abort("Attempting to fulfil a future that has already been fulfilled");
			copy_T$( result, val );
			return fulfil$( fut );
		}
		bool ?()( future(T) & fut, T val ) { return fulfil( fut, val ); } // alternate interface

		bool fulfil( future(T) & fut, exception_t * ex ) with( fut ) {
			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$( fut );
		}
		bool ?()( future(T) & fut, exception_t * ex ) { return fulfil( fut, ex ); } // alternate interface

		void reset( future(T) & fut ) with( fut ) {		// mark future as empty (for reuse)
			lock( lock );
		  if ( ! isEmpty( waiters ) ) abort( "Attempting to reset a future with blocked waiters" );
			state = FUTURE_EMPTY$;
			free( except );
			except = 0p;
			unlock( lock );
		}
	} // static inline
} // forall( T )

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

forall( T ) {
	// PRIVATE

	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 {
		size_t incRef$( future_rc_impl$( T ) & impl ) with( impl ) {
			return __atomic_fetch_add( &refCnt, 1, __ATOMIC_SEQ_CST );
		} // incRef$

		size_t decRef$( future_rc_impl$( T ) & impl ) with( impl ) {
			return __atomic_fetch_add( &refCnt, -1, __ATOMIC_SEQ_CST );
		} // decRef$

		void ?{}( future_rc_impl$( T ) & frc ) with( frc ) {
			refCnt = 1;									// count initial object
		} // ?{}
	} // static inline
	
	// PUBLIC

	struct future_rc {
		future_rc_impl$(T) * impl;		
	}; // future_rc
	__CFA_SELECT_GET_TYPE( future_rc(T) );				// magic
		
	static inline {
		// PRIVATE

		bool register_select$( future_rc(T) & frc, select_node & s ) with( frc ) { // for waituntil statement
			return register_select$( frc.impl->fut, s );
		}

		bool unregister_select$( future_rc(T) & frc, select_node & s ) with( frc ) { // for waituntil statement
			return unregister_select$( frc.impl->fut, s );
		}

		bool on_selected$( future_rc(T) &, select_node & ) { return true; } // for waituntil statement

		// PUBLIC

		// General

		void ?{}( future_rc( T ) & frc ) with( frc ) {	// default constructor
			impl = new();
		} // ?{}

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

		void ^?{}( future_rc( T ) & frc ) with( frc ) {
			if ( decRef$( *impl ) == 1 ) { 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 ) == 1 ) { delete( impl ); impl = 0p; } // no references ? => delete current impl
			impl = rhs.impl;							// point at new impl
			incRef$( *impl );							//   and increment reference count
			return lhs;
		} // ?+?

		// Used by Client

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

		// Return a value/exception from the future.
		[T, bool] get( future_rc(T) & frc ) with( frc ) { return get( impl->fut ); } // return future value
		T get( future_rc(T) & frc ) with( frc ) { return get( impl->fut ); } // return future value
		T ?()( future_rc(T) & frc ) with( frc ) { return get( frc ); } // alternate interface
		[T, bool] try_get( future_rc(T) & frc ) with( frc ) { return try_get( impl->fut ); }

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

		// Used by Server

		// Load a value/exception into the future, returns whether or not waiting threads.
		bool fulfil( future_rc(T) & frc, T val ) with( frc ) { return fulfil( impl->fut, val ); } // copy-in future value
		bool ?()( future_rc(T) & frc, T val ) { return fulfil( frc, val ); } // alternate interface

		bool fulfil( future_rc(T) & frc, exception_t * ex ) with( frc ) { return fulfil( impl->fut, ex ); } // insert future exception
		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 )

//--------------------------------------------------------------------------------------------------------
// This future does not support waituntil statements so it does not have as many features as 'future'.
// However, it is cheap and cheerful and is more performant than 'future' since it uses raw atomics
// and no locks
//--------------------------------------------------------------------------------------------------------

forall( T ) {
	// PUBLIC

	struct single_future {
		inline future_t;
		T result;
	};

	static inline {
		// PUBLIC

		bool available( single_future(T) & fut ) { return available( (future_t &)fut ); } // future result available ?

		// Return a value/exception from the future.
		[T, bool] get( single_future(T) & fut ) { return [fut.result, wait( fut )]; }
		T get( single_future(T) & fut ) { wait( fut ); return fut.result; }
		T ?()( single_future(T) & fut ) { return get( fut ); }	// alternate interface

		// Load a value into the future, returns whether or not waiting threads.
		bool fulfil( single_future(T) & fut, T result ) {
			fut.result = result;
			return fulfil( (future_t &)fut ) != 0p;
		}
		bool ?()( single_future(T) & fut, T val ) { return fulfil( fut, val ); } // alternate interface

		void reset( single_future(T) & fut ) { reset( (future_t &)fut ); } // mark future as empty (for reuse)
	} // static inline
} // forall( T )