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

ADT ast-experimental
Last change on this file since ff443e5 was 5e4a830, checked in by Peter A. Buhr <pabuhr@…>, 3 years 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.