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

ADT ast-experimental
Last change on this file since f76dd1a was 339e30a, checked in by caparsons <caparson@…>, 3 years ago

added new future and added rudimentary select statement support for it
  • Property mode set to 100644
File size: 8.7 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 }
59
60 // Reset future back to original state
61 void reset(future(T) & this) with(this)
62 {
63 lock( lock );
64 if( ! waiters`isEmpty )
65 abort("Attempting to reset a future with blocked waiters");
66 state = FUTURE_EMPTY;
67 unlock( lock );
68 }
69
70 // check if the future is available
71 // currently no mutual exclusion because I can't see when you need this call to be synchronous or protected
72 bool available( future(T) & this ) { return this.state; }
73
74
75 // memcpy wrapper to help copy values
76 void copy_T( T & from, T & to ) {
77 memcpy((void *)&to, (void *)&from, sizeof(T));
78 }
79
80 // internal helper to signal waiters off of the future
81 void _internal_flush( future(T) & this ) with(this) {
82 while( ! waiters`isEmpty ) {
83 select_node &s = try_pop_front( waiters );
84
85 if ( s.race_flag == 0p )
86 // poke in result so that woken threads do not need to reacquire any locks
87 // *(((future_node(T) &)s).my_result) = result;
88 copy_T( result, *(((future_node(T) &)s).my_result) );
89 else if ( !install_select_winner( s, &this ) ) continue;
90
91 // only unpark if future is not selected
92 // or if it is selected we only unpark if we win the race
93 unpark( s.blocked_thread );
94 }
95 }
96
97 // Fulfil the future, returns whether or not someone was unblocked
98 bool fulfil( future(T) & this, T & val ) with(this) {
99 lock( lock );
100 if( state != FUTURE_EMPTY )
101 abort("Attempting to fulfil a future that has already been fulfilled");
102
103 copy_T( val, result );
104
105 bool ret_val = ! waiters`isEmpty;
106 state = FUTURE_FULFILLED;
107 _internal_flush( this );
108 unlock( lock );
109 return ret_val;
110 }
111
112 // Wait for the future to be fulfilled
113 // Also return whether the thread had to block or not
114 [T, bool] get( future(T) & this ) with( this ) {
115 lock( lock );
116 T ret_val;
117 if( state == FUTURE_FULFILLED ) {
118 copy_T( result, ret_val );
119 unlock( lock );
120 return [ret_val, false];
121 }
122
123 future_node(T) node = { active_thread(), &ret_val };
124 insert_last( waiters, ((select_node &)node) );
125 unlock( lock );
126 park( );
127
128 return [ret_val, true];
129 }
130
131 // Wait for the future to be fulfilled
132 T get( future(T) & this ) {
133 [T, bool] tt;
134 tt = get(this);
135 return tt.0;
136 }
137
138 // Gets value if it is available and returns [ val, true ]
139 // otherwise returns [ default_val, false]
140 // will not block
141 [T, bool] try_get( future(T) & this ) with(this) {
142 lock( lock );
143 T ret_val;
144 if( state == FUTURE_FULFILLED ) {
145 copy_T( result, ret_val );
146 unlock( lock );
147 return [ret_val, true];
148 }
149 unlock( lock );
150 // cast to (T *) needed to trick the resolver to let me return *0p
151 return [ret_val, false];
152 }
153
154 void * register_select( future(T) & this, select_node & s ) with(this) {
155 lock( lock );
156
157 // future not ready -> insert select node and return 0p
158 if( state == FUTURE_EMPTY ) {
159 insert_last( waiters, s );
160 unlock( lock );
161 return 0p;
162 }
163
164 // future ready and we won race to install it as the select winner return 1p
165 if ( install_select_winner( s, &this ) ) {
166 unlock( lock );
167 return 1p;
168 }
169
170 unlock( lock );
171 // future ready and we lost race to install it as the select winner
172 return 2p;
173 }
174
175 void unregister_select( future(T) & this, select_node & s ) with(this) {
176 lock( lock );
177 if ( s`isListed ) remove( s );
178 unlock( lock );
179 }
180
181 }
182}
183
184//--------------------------------------------------------------------------------------------------------
185// These futures below do not support select statements so they may not be as useful as 'future'
186// however the 'single_future' is cheap and cheerful and is most likely more performant than 'future'
187// since it uses raw atomics and no locks afaik
188//
189// As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future'
190// since it is monitor based and also is not compatible with select statements
191//--------------------------------------------------------------------------------------------------------
192
193forall( T ) {
194 struct single_future {
195 inline future_t;
196 T result;
197 };
198
199 static inline {
200 // Reset future back to original state
201 void reset(single_future(T) & this) { reset( (future_t&)this ); }
202
203 // check if the future is available
204 bool available( single_future(T) & this ) { return available( (future_t&)this ); }
205
206 // Mark the future as abandoned, meaning it will be deleted by the server
207 // This doesn't work beause of the potential need for a destructor
208 void abandon( single_future(T) & this );
209
210 // Fulfil the future, returns whether or not someone was unblocked
211 thread$ * fulfil( single_future(T) & this, T result ) {
212 this.result = result;
213 return fulfil( (future_t&)this );
214 }
215
216 // Wait for the future to be fulfilled
217 // Also return whether the thread had to block or not
218 [T, bool] wait( single_future(T) & this ) {
219 bool r = wait( (future_t&)this );
220 return [this.result, r];
221 }
222
223 // Wait for the future to be fulfilled
224 T wait( single_future(T) & this ) {
225 [T, bool] tt;
226 tt = wait(this);
227 return tt.0;
228 }
229 }
230}
231
232forall( T ) {
233 monitor multi_future {
234 inline future_t;
235 condition blocked;
236 bool has_first;
237 T result;
238 };
239
240 static inline {
241 void ?{}(multi_future(T) & this) {
242 this.has_first = false;
243 }
244
245 bool $first( multi_future(T) & mutex this ) {
246 if (this.has_first) {
247 wait( this.blocked );
248 return false;
249 }
250
251 this.has_first = true;
252 return true;
253 }
254
255 void $first_done( multi_future(T) & mutex this ) {
256 this.has_first = false;
257 signal_all( this.blocked );
258 }
259
260 // Reset future back to original state
261 void reset(multi_future(T) & mutex this) {
262 if( this.has_first != false) abort("Attempting to reset a multi_future with at least one blocked threads");
263 if( !is_empty(this.blocked) ) abort("Attempting to reset a multi_future with multiple blocked threads");
264 reset( (future_t&)this );
265 }
266
267 // Fulfil the future, returns whether or not someone was unblocked
268 bool fulfil( multi_future(T) & this, T result ) {
269 this.result = result;
270 return fulfil( (future_t&)this ) != 0p;
271 }
272
273 // Wait for the future to be fulfilled
274 // Also return whether the thread had to block or not
275 [T, bool] wait( multi_future(T) & this ) {
276 bool sw = $first( this );
277 bool w = !sw;
278 if ( sw ) {
279 w = wait( (future_t&)this );
280 $first_done( this );
281 }
282
283 return [this.result, w];
284 }
285
286 // Wait for the future to be fulfilled
287 T wait( multi_future(T) & this ) {
288 return wait(this).0;
289 }
290 }
291}
Note: See TracBrowser for help on using the repository browser.