source: libcfa/src/concurrency/future.hfa @ 1afd9ccb

ADTast-experimental
Last change on this file since 1afd9ccb was 5e4a830, checked in by Peter A. Buhr <pabuhr@…>, 20 months ago

add #pragma once to .h and .hfa files

  • Property mode set to 100644
File size: 8.6 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2020 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// concurrency/future.hfa --
8//
9// Author           : Thierry Delisle & Peiran Hong & Colby Parsons
10// Created On       : Wed Jan 06 17:33:18 2021
11// Last Modified By :
12// Last Modified On :
13// Update Count     :
14//
15
16#pragma once
17
18#include "bits/locks.hfa"
19#include "monitor.hfa"
20#include "select.hfa"
21
22//----------------------------------------------------------------------------
23// future
24// I don't use future_t here since I need to use a lock for this future
25//  since it supports multiple consumers
26//  future_t is lockfree and uses atomics which aren't needed given we use locks here
27forall( T ) {
28    // enum(int) { FUTURE_EMPTY = 0, FUTURE_FULFILLED = 1 }; // Enums seem to be broken so feel free to add this back afterwards
29
30    // temporary enum replacement
31    const int FUTURE_EMPTY = 0;
32    const int FUTURE_FULFILLED = 1;
33
34        struct future {
35                int state;
36                T result;
37                dlist( select_node ) waiters;
38        futex_mutex lock;
39        };
40
41    struct future_node {
42        inline select_node;
43        T * my_result;
44    };
45
46    // C_TODO: perhaps allow exceptions to be inserted like uC++?
47
48        static inline {
49
50        void ?{}( future_node(T) & this, thread$ * blocked_thread, T * my_result ) {
51            ((select_node &)this){ blocked_thread };
52            this.my_result = my_result;
53        }
54
55        void ?{}(future(T) & this) {
56                        this.waiters{};
57            this.state = FUTURE_EMPTY;
58            this.lock{};
59                }
60
61                // Reset future back to original state
62                void reset(future(T) & this) with(this)
63        {
64            lock( lock );
65            if( ! waiters`isEmpty )
66                abort("Attempting to reset a future with blocked waiters");
67            state = FUTURE_EMPTY;
68            unlock( lock );
69        }
70
71                // check if the future is available
72        // currently no mutual exclusion because I can't see when you need this call to be synchronous or protected
73                bool available( future(T) & this ) { return this.state; }
74
75
76        // memcpy wrapper to help copy values
77        void copy_T( T & from, T & to ) {
78            memcpy((void *)&to, (void *)&from, sizeof(T));
79        }
80
81        // internal helper to signal waiters off of the future
82        void _internal_flush( future(T) & this ) with(this) {
83            while( ! waiters`isEmpty ) {
84                select_node &s = try_pop_front( waiters );
85
86                if ( s.race_flag == 0p )
87                    // poke in result so that woken threads do not need to reacquire any locks
88                    // *(((future_node(T) &)s).my_result) = result;
89                    copy_T( result, *(((future_node(T) &)s).my_result) );
90                else if ( !install_select_winner( s, &this ) ) continue;
91               
92                // only unpark if future is not selected
93                // or if it is selected we only unpark if we win the race
94                unpark( s.blocked_thread );
95            }
96        }
97
98                // Fulfil the future, returns whether or not someone was unblocked
99                bool fulfil( future(T) & this, T & val ) with(this) {
100            lock( lock );
101            if( state != FUTURE_EMPTY )
102                abort("Attempting to fulfil a future that has already been fulfilled");
103
104            copy_T( val, result );
105
106            bool ret_val = ! waiters`isEmpty;
107            state = FUTURE_FULFILLED;
108                        _internal_flush( this );
109            unlock( lock );
110            return ret_val;
111                }
112
113                // Wait for the future to be fulfilled
114                // Also return whether the thread had to block or not
115                [T, bool] get( future(T) & this ) with( this ) {
116            lock( lock );
117            T ret_val;
118            if( state == FUTURE_FULFILLED ) {
119                copy_T( result, ret_val );
120                unlock( lock );
121                return [ret_val, false];
122            }
123
124            future_node(T) node = { active_thread(), &ret_val };
125            insert_last( waiters, ((select_node &)node) );
126            unlock( lock );
127            park( );
128
129                        return [ret_val, true];
130                }
131
132                // Wait for the future to be fulfilled
133                T get( future(T) & this ) {
134                        [T, bool] tt;
135                        tt = get(this);
136                        return tt.0;
137                }
138
139        // Gets value if it is available and returns [ val, true ]
140        // otherwise returns [ default_val, false]
141        // will not block
142        [T, bool] try_get( future(T) & this ) with(this) {
143            lock( lock );
144            T ret_val;
145            if( state == FUTURE_FULFILLED ) {
146                copy_T( result, ret_val );
147                unlock( lock );
148                return [ret_val, true];
149            }
150            unlock( lock );
151           
152            return [ret_val, false];
153        }
154
155        void * register_select( future(T) & this, select_node & s ) with(this) {
156            lock( lock );
157
158            // future not ready -> insert select node and return 0p
159            if( state == FUTURE_EMPTY ) {
160                insert_last( waiters, s );
161                unlock( lock );
162                return 0p;
163            }
164
165            // future ready and we won race to install it as the select winner return 1p
166            if ( install_select_winner( s, &this ) ) {
167                unlock( lock );
168                return 1p;
169            }
170
171            unlock( lock );
172            // future ready and we lost race to install it as the select winner
173            return 2p;
174        }
175
176        void unregister_select( future(T) & this, select_node & s ) with(this) {
177            lock( lock );
178            if ( s`isListed ) remove( s );
179            unlock( lock );
180        }
181               
182        }
183}
184
185//--------------------------------------------------------------------------------------------------------
186// These futures below do not support select statements so they may not be as useful as 'future'
187//  however the 'single_future' is cheap and cheerful and is most likely more performant than 'future'
188//  since it uses raw atomics and no locks afaik
189//
190// As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future'
191//  since it is monitor based and also is not compatible with select statements
192//--------------------------------------------------------------------------------------------------------
193
194forall( T ) {
195        struct single_future {
196                inline future_t;
197                T result;
198        };
199
200        static inline {
201                // Reset future back to original state
202                void reset(single_future(T) & this) { reset( (future_t&)this ); }
203
204                // check if the future is available
205                bool available( single_future(T) & this ) { return available( (future_t&)this ); }
206
207                // Mark the future as abandoned, meaning it will be deleted by the server
208                // This doesn't work beause of the potential need for a destructor
209                void abandon( single_future(T) & this );
210
211                // Fulfil the future, returns whether or not someone was unblocked
212                thread$ * fulfil( single_future(T) & this, T result ) {
213                        this.result = result;
214                        return fulfil( (future_t&)this );
215                }
216
217                // Wait for the future to be fulfilled
218                // Also return whether the thread had to block or not
219                [T, bool] wait( single_future(T) & this ) {
220                        bool r = wait( (future_t&)this );
221                        return [this.result, r];
222                }
223
224                // Wait for the future to be fulfilled
225                T wait( single_future(T) & this ) {
226                        [T, bool] tt;
227                        tt = wait(this);
228                        return tt.0;
229                }
230        }
231}
232
233forall( T ) {
234        monitor multi_future {
235                inline future_t;
236                condition blocked;
237                bool has_first;
238                T result;
239        };
240
241        static inline {
242                void ?{}(multi_future(T) & this) {
243                        this.has_first = false;
244                }
245
246                bool $first( multi_future(T) & mutex this ) {
247                        if (this.has_first) {
248                                wait( this.blocked );
249                                return false;
250                        }
251
252                        this.has_first = true;
253                        return true;
254                }
255
256                void $first_done( multi_future(T) & mutex this ) {
257                        this.has_first = false;
258                        signal_all( this.blocked );
259                }
260
261                // Reset future back to original state
262                void reset(multi_future(T) & mutex this) {
263                        if( this.has_first != false) abort("Attempting to reset a multi_future with at least one blocked threads");
264                        if( !is_empty(this.blocked) ) abort("Attempting to reset a multi_future with multiple blocked threads");
265                        reset( (future_t&)this );
266                }
267
268                // Fulfil the future, returns whether or not someone was unblocked
269                bool fulfil( multi_future(T) & this, T result ) {
270                        this.result = result;
271                        return fulfil( (future_t&)this ) != 0p;
272                }
273
274                // Wait for the future to be fulfilled
275                // Also return whether the thread had to block or not
276                [T, bool] wait( multi_future(T) & this ) {
277                        bool sw = $first( this );
278                        bool w = !sw;
279                        if ( sw ) {
280                                w = wait( (future_t&)this );
281                                $first_done( this );
282                        }
283
284                        return [this.result, w];
285                }
286
287                // Wait for the future to be fulfilled
288                T wait( multi_future(T) & this ) {
289                        return wait(this).0;
290                }
291        }
292}
Note: See TracBrowser for help on using the repository browser.