source: libcfa/src/concurrency/future.hfa@ fbaea970

Last change on this file since fbaea970 was fbaea970, checked in by Peter A. Buhr <pabuhr@…>, 5 days ago

rename private waituntil routines register_select, unregister_select, and on_selected with tailing $
  • Property mode set to 100644
File size: 13.0 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 & Peter Buhr
10// Created On : Wed Jan 06 17:33:18 2021
11// Last Modified By : Peter A. Buhr
12// Last Modified On : Sun Nov 23 22:48:08 2025
13// Update Count : 208
14//
15
16#pragma once
17
18#include "bits/locks.hfa"
19#include "monitor.hfa"
20#include "select.hfa"
21#include "locks.hfa"
22
23//--------------------------------------------------------------------------------------------------------
24// future does not use future_t as it needs a lock to support multiple consumers. future_t is lockfree
25// and uses atomics which are not needed.
26//--------------------------------------------------------------------------------------------------------
27
28forall( T ) {
29 // PRIVATE
30
31 struct future_node$ {
32 inline select_node;
33 T * my_result;
34 };
35
36 static inline {
37 // memcpy wrapper to help copy values
38 void copy_T$( T & to, T & from ) { memcpy( (void *)&to, (void *)&from, sizeof(T) ); }
39 } // distribution
40
41 enum { FUTURE_EMPTY$ = 0, FUTURE_FULFILLED$ = 1 };
42
43 // PUBLIC
44
45 struct future {
46 int state;
47 T result;
48 exception_t * except;
49 futex_mutex lock;
50 dlist( select_node ) waiters;
51 };
52 __CFA_SELECT_GET_TYPE( future(T) ); // magic
53
54 static inline {
55 // PRIVATE
56
57 bool register_select$( future(T) & fut, select_node & s ) with( fut ) { // for waituntil statement
58 lock( lock );
59
60 // check if we can complete operation. If so race to establish winner in special OR case
61 if ( !s.park_counter && state != FUTURE_EMPTY$ ) {
62 if ( !__make_select_node_available( s ) ) { // we didn't win the race so give up on registering
63 unlock( lock );
64 return false;
65 }
66 }
67
68 // future not ready -> insert select node and return
69 if ( state == FUTURE_EMPTY$ ) {
70 insert_last( waiters, s );
71 unlock( lock );
72 return false;
73 }
74
75 __make_select_node_available( s );
76 unlock( lock );
77 return true;
78 }
79
80 bool unregister_select$( future(T) & fut, select_node & s ) with( fut ) { // for waituntil statement
81 if ( ! isListed( s ) ) return false;
82 lock( lock );
83 if ( isListed( s ) ) remove( s );
84 unlock( lock );
85 return false;
86 }
87
88 bool on_selected$( future(T) &, select_node & ) { return true; } // for waituntil statement
89
90 // PUBLIC
91
92 // General
93
94 void ?{}( future_node$(T) & fut, thread$ * blocked_thread, T * my_result ) {
95 ((select_node &)fut){ blocked_thread };
96 fut.my_result = my_result;
97 }
98
99 void ?{}( future(T) & fut ) with( fut ) {
100 except = 0p;
101 state = FUTURE_EMPTY$;
102 lock{};
103 }
104
105 void ^?{}( future(T) & fut ) with( fut ) {
106 free( except );
107 }
108
109 // Used by Client
110
111 // PRIVATE
112
113 // Return a value/exception from the future.
114 T get$( future(T) & fut ) with( fut ) { // helper
115 void exceptCheck() { // helper
116 if ( except ) {
117 exception_t * ex = ( exception_t * ) alloca( except->virtual_table->size );
118 except->virtual_table->copy( ex, except );
119 unlock( lock );
120 throwResume * ex;
121 }
122 }
123 T ret_val;
124
125 // LOCK ACQUIRED IN PUBLIC get
126 if ( state == FUTURE_FULFILLED$ ) {
127 exceptCheck();
128 copy_T$( ret_val, result );
129 unlock( lock );
130 return ret_val;
131 }
132
133 future_node$(T) node = { active_thread(), &ret_val };
134 insert_last( waiters, ((select_node &)node) );
135 unlock( lock );
136 park( );
137 exceptCheck();
138 return ret_val;
139 }
140
141 // PUBLIC
142
143 bool available( future( T ) & fut ) { return __atomic_load_n( &fut.state, __ATOMIC_RELAXED ); } // future result available ?
144
145 // Return a value/exception from the future.
146 [T, bool] get( future(T) & fut ) with( fut ) {
147 lock( lock );
148 bool ret = state == FUTURE_EMPTY$;
149 return [ get$( fut ), ret ];
150 }
151
152 T get( future(T) & fut ) with( fut ) {
153 lock( lock );
154 return get$( fut );
155 }
156 T ?()( future(T) & fut ) { return get( fut ); } // alternate interface
157
158 // Non-blocking get: true => return defined value, false => value return undefined.
159 [T, bool] try_get( future(T) & fut ) with( fut ) {
160 lock( lock );
161 T ret_val;
162 if ( state == FUTURE_FULFILLED$ ) {
163 copy_T$( ret_val, result );
164 unlock( lock );
165 return [ret_val, true];
166 }
167 unlock( lock );
168 return [ret_val, false];
169 }
170
171 // Used by Server
172
173 // PRIVATE
174
175 bool fulfil$( future(T) & fut ) with( fut ) { // helper
176 bool ret_val = ! isEmpty( waiters );
177 state = FUTURE_FULFILLED$;
178 while ( ! isEmpty( waiters ) ) {
179 if ( !__handle_waituntil_OR( waiters ) ) // handle special waituntil OR case
180 break; // if handle_OR returns false then waiters is empty so break
181 select_node &s = remove_first( waiters );
182
183 if ( s.clause_status == 0p ) // poke in result so that woken threads do not need to reacquire any locks
184 copy_T$( *(((future_node$(T) &)s).my_result), result );
185
186 wake_one( waiters, s );
187 }
188 unlock( lock );
189 return ret_val;
190 }
191
192 // PUBLIC
193
194 // Load a value/exception into the future, returns whether or not waiting threads.
195 bool fulfil( future(T) & fut, T val ) with( fut ) {
196 lock( lock );
197 if ( state != FUTURE_EMPTY$ ) abort("Attempting to fulfil a future that has already been fulfilled");
198 copy_T$( result, val );
199 return fulfil$( fut );
200 }
201 bool ?()( future(T) & fut, T val ) { return fulfil( fut, val ); } // alternate interface
202
203 bool fulfil( future(T) & fut, exception_t * ex ) with( fut ) {
204 lock( lock );
205 if ( state != FUTURE_EMPTY$ ) abort( "Attempting to fulfil a future that has already been fulfilled" );
206 except = ( exception_t * ) malloc( ex->virtual_table->size );
207 ex->virtual_table->copy( except, ex );
208 return fulfil$( fut );
209 }
210 bool ?()( future(T) & fut, exception_t * ex ) { return fulfil( fut, ex ); } // alternate interface
211
212 void reset( future(T) & fut ) with( fut ) { // mark future as empty (for reuse)
213 lock( lock );
214 if ( ! isEmpty( waiters ) ) abort( "Attempting to reset a future with blocked waiters" );
215 state = FUTURE_EMPTY$;
216 free( except );
217 except = 0p;
218 unlock( lock );
219 }
220 } // static inline
221} // forall( T )
222
223//--------------------------------------------------------------------------------------------------------
224// future_rc uses reference counting to eliminate explicit storage-management and support the waituntil
225// statement.
226//--------------------------------------------------------------------------------------------------------
227
228forall( T ) {
229 // PRIVATE
230
231 struct future_rc_impl$ {
232 futex_mutex lock; // concurrent protection
233 size_t refCnt; // number of references to future
234 future(T) fut; // underlying future
235 }; // future_rc_impl$
236
237 static inline {
238 size_t incRef$( future_rc_impl$( T ) & impl ) with( impl ) {
239 return __atomic_fetch_add( &refCnt, 1, __ATOMIC_SEQ_CST );
240 } // incRef$
241
242 size_t decRef$( future_rc_impl$( T ) & impl ) with( impl ) {
243 return __atomic_fetch_add( &refCnt, -1, __ATOMIC_SEQ_CST );
244 } // decRef$
245
246 void ?{}( future_rc_impl$( T ) & frc ) with( frc ) {
247 refCnt = 1; // count initial object
248 } // ?{}
249 } // static inline
250
251 // PUBLIC
252
253 struct future_rc {
254 future_rc_impl$(T) * impl;
255 }; // future_rc
256 __CFA_SELECT_GET_TYPE( future_rc(T) ); // magic
257
258 static inline {
259 // PRIVATE
260
261 bool register_select$( future_rc(T) & frc, select_node & s ) with( frc ) { // for waituntil statement
262 return register_select$( frc.impl->fut, s );
263 }
264
265 bool unregister_select$( future_rc(T) & frc, select_node & s ) with( frc ) { // for waituntil statement
266 return unregister_select$( frc.impl->fut, s );
267 }
268
269 bool on_selected$( future_rc(T) &, select_node & ) { return true; } // for waituntil statement
270
271 // PUBLIC
272
273 // General
274
275 void ?{}( future_rc( T ) & frc ) with( frc ) { // default constructor
276 impl = new();
277 } // ?{}
278
279 void ?{}( future_rc( T ) & to, future_rc( T ) & from ) with( to ) { // copy constructor
280 impl = from.impl; // point at new impl
281 incRef$( *impl );
282 } // ?{}
283
284 void ^?{}( future_rc( T ) & frc ) with( frc ) {
285 if ( decRef$( *impl ) == 1 ) { delete( impl ); impl = 0p; }
286 } // ^?{}
287
288 future_rc( T ) & ?=?( future_rc( T ) & lhs, future_rc( T ) & rhs ) with( lhs ) {
289 if ( impl == rhs.impl ) return lhs; // self assignment ?
290 if ( decRef$( *impl ) == 1 ) { delete( impl ); impl = 0p; } // no references ? => delete current impl
291 impl = rhs.impl; // point at new impl
292 incRef$( *impl ); // and increment reference count
293 return lhs;
294 } // ?+?
295
296 // Used by Client
297
298 bool available( future_rc( T ) & frc ) { return available( frc.impl->fut ); } // future result available ?
299
300 // Return a value/exception from the future.
301 [T, bool] get( future_rc(T) & frc ) with( frc ) { return get( impl->fut ); } // return future value
302 T get( future_rc(T) & frc ) with( frc ) { return get( impl->fut ); } // return future value
303 T ?()( future_rc(T) & frc ) with( frc ) { return get( frc ); } // alternate interface
304 [T, bool] try_get( future_rc(T) & frc ) with( frc ) { return try_get( impl->fut ); }
305
306 int ?==?( future_rc( T ) & lhs, future_rc( T ) & rhs ) { return lhs.impl == rhs.impl; } // referential equality
307
308 // Used by Server
309
310 // Load a value/exception into the future, returns whether or not waiting threads.
311 bool fulfil( future_rc(T) & frc, T val ) with( frc ) { return fulfil( impl->fut, val ); } // copy-in future value
312 bool ?()( future_rc(T) & frc, T val ) { return fulfil( frc, val ); } // alternate interface
313
314 bool fulfil( future_rc(T) & frc, exception_t * ex ) with( frc ) { return fulfil( impl->fut, ex ); } // insert future exception
315 bool ?()( future_rc(T) & frc, exception_t * ex ) { return fulfil( frc, ex ); } // alternate interface
316
317 void reset( future_rc(T) & frc ) with( frc ) { reset( impl->fut ); } // mark future as empty (for reuse)
318 } // static inline
319} // forall( T )
320
321//--------------------------------------------------------------------------------------------------------
322// These futures below do not support waituntil statements so they may not have as many features as 'future'
323// however the 'single_future' is cheap and cheerful and is most likely more performant than 'future'
324// since it uses raw atomics and no locks
325//
326// As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future'
327// since it is monitor based and also is not compatible with waituntil statement.
328//--------------------------------------------------------------------------------------------------------
329
330forall( T ) {
331 struct single_future {
332 inline future_t;
333 T result;
334 };
335
336 static inline {
337 // Reset future back to original state
338 void reset(single_future(T) & this) { reset( (future_t&)this ); }
339
340 // check if the future is available
341 bool available( single_future(T) & this ) { return available( (future_t&)this ); }
342
343 // Mark the future as abandoned, meaning it will be deleted by the server
344 // This doesn't work beause of the potential need for a destructor
345 // void abandon( single_future(T) & this );
346
347 // Fulfil the future, returns whether or not someone was unblocked
348 thread$ * fulfil( single_future(T) & this, T result ) {
349 this.result = result;
350 return fulfil( (future_t&)this );
351 }
352
353 // Wait for the future to be fulfilled
354 // Also return whether the thread had to block or not
355 [T, bool] wait( single_future(T) & this ) {
356 bool r = wait( (future_t&)this );
357 return [this.result, r];
358 }
359
360 // Wait for the future to be fulfilled
361 T wait( single_future(T) & this ) {
362 [T, bool] tt;
363 tt = wait( this );
364 return tt.0;
365 }
366 }
367}
368
369forall( T ) {
370 monitor multi_future {
371 inline future_t;
372 condition blocked;
373 bool has_first;
374 T result;
375 };
376
377 static inline {
378 void ?{}(multi_future(T) & this) {
379 this.has_first = false;
380 }
381
382 bool $first( multi_future(T) & mutex this ) {
383 if ( this.has_first ) {
384 wait( this.blocked );
385 return false;
386 }
387
388 this.has_first = true;
389 return true;
390 }
391
392 void $first_done( multi_future(T) & mutex this ) {
393 this.has_first = false;
394 signal_all( this.blocked );
395 }
396
397 // Reset future back to original state
398 void reset(multi_future(T) & mutex this) {
399 if ( this.has_first != false ) abort("Attempting to reset a multi_future with at least one blocked threads");
400 if ( ! empty( this.blocked ) ) abort("Attempting to reset a multi_future with multiple blocked threads");
401 reset( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) );
402 }
403
404 // Fulfil the future, returns whether or not someone was unblocked
405 bool fulfil( multi_future(T) & this, T result ) {
406 this.result = result;
407 return fulfil( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) ) != 0p;
408 }
409
410 // Wait for the future to be fulfilled
411 // Also return whether the thread had to block or not
412 [T, bool] wait( multi_future(T) & this ) {
413 bool sw = $first( this );
414 bool w = !sw;
415 if ( sw ) {
416 w = wait( (future_t&)*(future_t*)((uintptr_t)&this + sizeof(monitor$)) );
417 $first_done( this );
418 }
419
420 return [this.result, w];
421 }
422
423 // Wait for the future to be fulfilled
424 T wait( multi_future(T) & this ) {
425 return wait( this ).0;
426 }
427 }
428}
Note: See TracBrowser for help on using the repository browser.