Changeset 24d6572 for libcfa/src/concurrency/future.hfa

Timestamp:

Jun 12, 2023, 2:45:32 PM (2 years ago)

Author:

Fangren Yu <f37yu@…>

Branches:

ast-experimental, master

Children:

62d62db

Parents:

34b4268 (diff), 251ce80 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' into ast-experimental

File:

• libcfa/src/concurrency/future.hfa (modified) (5 diffs)

libcfa/src/concurrency/future.hfa

@@ -5 +5 @@
 // file "LICENCE" distributed with Cforall.
 //
-// io/types.hfa --
-//
-// Author           : Thierry Delisle & Peiran Hong
+// concurrency/future.hfa --
+//
+// Author           : Thierry Delisle & Peiran Hong & Colby Parsons
 // Created On       : Wed Jan 06 17:33:18 2021
 // Last Modified By :
@@ -18 +18 @@
 #include "bits/locks.hfa"
 #include "monitor.hfa"
-
+#include "select.hfa"
+#include "locks.hfa"
+
+//----------------------------------------------------------------------------
+// future
+// I don't use future_t here since 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 }; // Enums seem to be broken so feel free to add this back afterwards
+
+    // temporary enum replacement
+    const int FUTURE_EMPTY = 0;
+    const int FUTURE_FULFILLED = 1;
+
         struct future {
+                int state;
+                T result;
+                dlist( select_node ) waiters;
+        futex_mutex lock;
+        };
+
+    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;
+        }
+                
+        void on_selected( future(T) & this, select_node & node ) {}
+        }
+}
+
+//--------------------------------------------------------------------------------------------------------
+// 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;
@@ -27 +201 @@
         static inline {
                 // Reset future back to original state
-                void reset(future(T) & this) { reset( (future_t&)this ); }
+                void reset(single_future(T) & this) { reset( (future_t&)this ); }
 
                 // check if the future is available
-                bool available( future(T) & this ) { return available( (future_t&)this ); }
+                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( future(T) & this );
+                void abandon( single_future(T) & this );
 
                 // Fulfil the future, returns whether or not someone was unblocked
-                thread$ * fulfil( future(T) & this, T result ) {
+                thread$ * fulfil( single_future(T) & this, T result ) {
                         this.result = result;
                         return fulfil( (future_t&)this );
@@ -44 +218 @@
                 // Wait for the future to be fulfilled
                 // Also return whether the thread had to block or not
-                [T, bool] wait( future(T) & this ) {
+                [T, bool] wait( single_future(T) & this ) {
                         bool r = wait( (future_t&)this );
                         return [this.result, r];
@@ -50 +224 @@
 
                 // Wait for the future to be fulfilled
-                T wait( future(T) & this ) {
+                T wait( single_future(T) & this ) {
                         [T, bool] tt;
                         tt = wait(this);