[70f8bcd2] | 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 | // |
---|
[339e30a] | 7 | // concurrency/future.hfa -- |
---|
[70f8bcd2] | 8 | // |
---|
[339e30a] | 9 | // Author : Thierry Delisle & Peiran Hong & Colby Parsons |
---|
[70f8bcd2] | 10 | // Created On : Wed Jan 06 17:33:18 2021 |
---|
| 11 | // Last Modified By : |
---|
| 12 | // Last Modified On : |
---|
| 13 | // Update Count : |
---|
| 14 | // |
---|
| 15 | |
---|
[5e4a830] | 16 | #pragma once |
---|
[70f8bcd2] | 17 | |
---|
| 18 | #include "bits/locks.hfa" |
---|
| 19 | #include "monitor.hfa" |
---|
[339e30a] | 20 | #include "select.hfa" |
---|
[beeff61e] | 21 | #include "locks.hfa" |
---|
[70f8bcd2] | 22 | |
---|
[339e30a] | 23 | //---------------------------------------------------------------------------- |
---|
| 24 | // future |
---|
| 25 | // I don't use future_t here since I need to use a lock for this future |
---|
| 26 | // since it supports multiple consumers |
---|
| 27 | // future_t is lockfree and uses atomics which aren't needed given we use locks here |
---|
[fd54fef] | 28 | forall( T ) { |
---|
[beeff61e] | 29 | // enum { FUTURE_EMPTY = 0, FUTURE_FULFILLED = 1 }; // Enums seem to be broken so feel free to add this back afterwards |
---|
[339e30a] | 30 | |
---|
| 31 | // temporary enum replacement |
---|
| 32 | const int FUTURE_EMPTY = 0; |
---|
| 33 | const int FUTURE_FULFILLED = 1; |
---|
| 34 | |
---|
[70f8bcd2] | 35 | struct future { |
---|
[339e30a] | 36 | int state; |
---|
| 37 | T result; |
---|
| 38 | dlist( select_node ) waiters; |
---|
| 39 | futex_mutex lock; |
---|
| 40 | }; |
---|
| 41 | |
---|
| 42 | struct future_node { |
---|
| 43 | inline select_node; |
---|
| 44 | T * my_result; |
---|
| 45 | }; |
---|
| 46 | |
---|
| 47 | static inline { |
---|
| 48 | |
---|
| 49 | void ?{}( future_node(T) & this, thread$ * blocked_thread, T * my_result ) { |
---|
| 50 | ((select_node &)this){ blocked_thread }; |
---|
| 51 | this.my_result = my_result; |
---|
| 52 | } |
---|
| 53 | |
---|
| 54 | void ?{}(future(T) & this) { |
---|
| 55 | this.waiters{}; |
---|
| 56 | this.state = FUTURE_EMPTY; |
---|
[5e180c2] | 57 | this.lock{}; |
---|
[339e30a] | 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 ) { |
---|
[beeff61e] | 83 | if ( !__handle_waituntil_OR( waiters ) ) // handle special waituntil OR case |
---|
| 84 | break; // if handle_OR returns false then waiters is empty so break |
---|
[339e30a] | 85 | select_node &s = try_pop_front( waiters ); |
---|
| 86 | |
---|
[beeff61e] | 87 | if ( s.clause_status == 0p ) |
---|
[339e30a] | 88 | // poke in result so that woken threads do not need to reacquire any locks |
---|
| 89 | copy_T( result, *(((future_node(T) &)s).my_result) ); |
---|
[beeff61e] | 90 | else if ( !__make_select_node_available( s ) ) continue; |
---|
[339e30a] | 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 |
---|
[beeff61e] | 99 | bool fulfil( future(T) & this, T val ) with(this) { |
---|
[339e30a] | 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 ); |
---|
[5e180c2] | 151 | |
---|
[339e30a] | 152 | return [ret_val, false]; |
---|
| 153 | } |
---|
| 154 | |
---|
[beeff61e] | 155 | bool register_select( future(T) & this, select_node & s ) with(this) { |
---|
[339e30a] | 156 | lock( lock ); |
---|
| 157 | |
---|
[beeff61e] | 158 | // check if we can complete operation. If so race to establish winner in special OR case |
---|
| 159 | if ( !s.park_counter && state != FUTURE_EMPTY ) { |
---|
| 160 | if ( !__make_select_node_available( s ) ) { // we didn't win the race so give up on registering |
---|
| 161 | unlock( lock ); |
---|
| 162 | return false; |
---|
| 163 | } |
---|
[339e30a] | 164 | } |
---|
| 165 | |
---|
[beeff61e] | 166 | // future not ready -> insert select node and return |
---|
| 167 | if( state == FUTURE_EMPTY ) { |
---|
| 168 | insert_last( waiters, s ); |
---|
[339e30a] | 169 | unlock( lock ); |
---|
[beeff61e] | 170 | return false; |
---|
[339e30a] | 171 | } |
---|
| 172 | |
---|
[beeff61e] | 173 | __make_select_node_available( s ); |
---|
[339e30a] | 174 | unlock( lock ); |
---|
[beeff61e] | 175 | return true; |
---|
[339e30a] | 176 | } |
---|
| 177 | |
---|
[beeff61e] | 178 | bool unregister_select( future(T) & this, select_node & s ) with(this) { |
---|
| 179 | if ( ! s`isListed ) return false; |
---|
[339e30a] | 180 | lock( lock ); |
---|
| 181 | if ( s`isListed ) remove( s ); |
---|
| 182 | unlock( lock ); |
---|
[beeff61e] | 183 | return false; |
---|
[339e30a] | 184 | } |
---|
| 185 | |
---|
[beeff61e] | 186 | bool on_selected( future(T) & this, select_node & node ) { return true; } |
---|
[339e30a] | 187 | } |
---|
| 188 | } |
---|
| 189 | |
---|
| 190 | //-------------------------------------------------------------------------------------------------------- |
---|
| 191 | // These futures below do not support select statements so they may not be as useful as 'future' |
---|
| 192 | // however the 'single_future' is cheap and cheerful and is most likely more performant than 'future' |
---|
[beeff61e] | 193 | // since it uses raw atomics and no locks |
---|
[339e30a] | 194 | // |
---|
| 195 | // As far as 'multi_future' goes I can't see many use cases as it will be less performant than 'future' |
---|
| 196 | // since it is monitor based and also is not compatible with select statements |
---|
| 197 | //-------------------------------------------------------------------------------------------------------- |
---|
| 198 | |
---|
| 199 | forall( T ) { |
---|
| 200 | struct single_future { |
---|
[70f8bcd2] | 201 | inline future_t; |
---|
| 202 | T result; |
---|
| 203 | }; |
---|
| 204 | |
---|
| 205 | static inline { |
---|
| 206 | // Reset future back to original state |
---|
[339e30a] | 207 | void reset(single_future(T) & this) { reset( (future_t&)this ); } |
---|
[70f8bcd2] | 208 | |
---|
| 209 | // check if the future is available |
---|
[339e30a] | 210 | bool available( single_future(T) & this ) { return available( (future_t&)this ); } |
---|
[70f8bcd2] | 211 | |
---|
| 212 | // Mark the future as abandoned, meaning it will be deleted by the server |
---|
| 213 | // This doesn't work beause of the potential need for a destructor |
---|
[339e30a] | 214 | void abandon( single_future(T) & this ); |
---|
[70f8bcd2] | 215 | |
---|
| 216 | // Fulfil the future, returns whether or not someone was unblocked |
---|
[339e30a] | 217 | thread$ * fulfil( single_future(T) & this, T result ) { |
---|
[70f8bcd2] | 218 | this.result = result; |
---|
| 219 | return fulfil( (future_t&)this ); |
---|
| 220 | } |
---|
| 221 | |
---|
| 222 | // Wait for the future to be fulfilled |
---|
| 223 | // Also return whether the thread had to block or not |
---|
[339e30a] | 224 | [T, bool] wait( single_future(T) & this ) { |
---|
[70f8bcd2] | 225 | bool r = wait( (future_t&)this ); |
---|
| 226 | return [this.result, r]; |
---|
| 227 | } |
---|
| 228 | |
---|
| 229 | // Wait for the future to be fulfilled |
---|
[339e30a] | 230 | T wait( single_future(T) & this ) { |
---|
[70f8bcd2] | 231 | [T, bool] tt; |
---|
| 232 | tt = wait(this); |
---|
| 233 | return tt.0; |
---|
| 234 | } |
---|
| 235 | } |
---|
| 236 | } |
---|
| 237 | |
---|
[fd54fef] | 238 | forall( T ) { |
---|
[70f8bcd2] | 239 | monitor multi_future { |
---|
| 240 | inline future_t; |
---|
| 241 | condition blocked; |
---|
| 242 | bool has_first; |
---|
| 243 | T result; |
---|
| 244 | }; |
---|
| 245 | |
---|
| 246 | static inline { |
---|
| 247 | void ?{}(multi_future(T) & this) { |
---|
| 248 | this.has_first = false; |
---|
| 249 | } |
---|
| 250 | |
---|
| 251 | bool $first( multi_future(T) & mutex this ) { |
---|
| 252 | if (this.has_first) { |
---|
| 253 | wait( this.blocked ); |
---|
| 254 | return false; |
---|
| 255 | } |
---|
| 256 | |
---|
| 257 | this.has_first = true; |
---|
| 258 | return true; |
---|
| 259 | } |
---|
| 260 | |
---|
| 261 | void $first_done( multi_future(T) & mutex this ) { |
---|
| 262 | this.has_first = false; |
---|
| 263 | signal_all( this.blocked ); |
---|
| 264 | } |
---|
| 265 | |
---|
| 266 | // Reset future back to original state |
---|
| 267 | void reset(multi_future(T) & mutex this) { |
---|
| 268 | if( this.has_first != false) abort("Attempting to reset a multi_future with at least one blocked threads"); |
---|
| 269 | if( !is_empty(this.blocked) ) abort("Attempting to reset a multi_future with multiple blocked threads"); |
---|
| 270 | reset( (future_t&)this ); |
---|
| 271 | } |
---|
| 272 | |
---|
| 273 | // Fulfil the future, returns whether or not someone was unblocked |
---|
| 274 | bool fulfil( multi_future(T) & this, T result ) { |
---|
| 275 | this.result = result; |
---|
[c323837] | 276 | return fulfil( (future_t&)this ) != 0p; |
---|
[70f8bcd2] | 277 | } |
---|
| 278 | |
---|
| 279 | // Wait for the future to be fulfilled |
---|
| 280 | // Also return whether the thread had to block or not |
---|
| 281 | [T, bool] wait( multi_future(T) & this ) { |
---|
| 282 | bool sw = $first( this ); |
---|
| 283 | bool w = !sw; |
---|
| 284 | if ( sw ) { |
---|
| 285 | w = wait( (future_t&)this ); |
---|
| 286 | $first_done( this ); |
---|
| 287 | } |
---|
| 288 | |
---|
| 289 | return [this.result, w]; |
---|
| 290 | } |
---|
| 291 | |
---|
| 292 | // Wait for the future to be fulfilled |
---|
| 293 | T wait( multi_future(T) & this ) { |
---|
| 294 | return wait(this).0; |
---|
| 295 | } |
---|
| 296 | } |
---|
[beeff61e] | 297 | } |
---|