source: libcfa/src/concurrency/channel.hfa@ 76e77a4

ADT ast-experimental
Last change on this file since 76e77a4 was 5908fb4, checked in by caparsons <caparson@…>, 2 years ago

changed channel fetch routines to use relaxed atomics
  • Property mode set to 100644
File size: 17.7 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2021 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// channel.hfa -- LIBCFATHREAD
8// Runtime locks that used with the runtime thread system.
9//
10// Author : Colby Alexander Parsons
11// Created On : Thu Jan 21 19:46:50 2022
12// Last Modified By :
13// Last Modified On :
14// Update Count :
15//
16
17#pragma once
18
19#include <locks.hfa>
20#include <list.hfa>
21#include "select.hfa"
22
23// returns true if woken due to shutdown
24// blocks thread on list and releases passed lock
25static inline bool block( dlist( select_node ) & queue, void * elem_ptr, go_mutex & lock ) {
26 select_node sn{ active_thread(), elem_ptr };
27 insert_last( queue, sn );
28 unlock( lock );
29 park();
30 return sn.extra == 0p;
31}
32
33// Waituntil support (un)register_select helper routine
34// Sets select node avail if not special OR case and then unlocks
35static inline void __set_avail_then_unlock( select_node & node, go_mutex & mutex_lock ) {
36 if ( node.park_counter ) __make_select_node_available( node );
37 unlock( mutex_lock );
38}
39
40// void * used for some fields since exceptions don't work with parametric polymorphism currently
41exception channel_closed {
42 // on failed insert elem is a ptr to the element attempting to be inserted
43 // on failed remove elem ptr is 0p
44 // on resumption of a failed insert this elem will be inserted
45 // so a user may modify it in the resumption handler
46 void * elem;
47
48 // pointer to chan that is closed
49 void * closed_chan;
50};
51vtable(channel_closed) channel_closed_vt;
52
53// #define CHAN_STATS // define this to get channel stats printed in dtor
54
55forall( T ) {
56
57struct __attribute__((aligned(128))) channel {
58 size_t size, front, back, count;
59 T * buffer;
60 dlist( select_node ) prods, cons; // lists of blocked threads
61 go_mutex mutex_lock; // MX lock
62 bool closed; // indicates channel close/open
63 #ifdef CHAN_STATS
64 size_t blocks, operations; // counts total ops and ops resulting in a blocked thd
65 #endif
66};
67
68static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
69 size = _size;
70 front = back = count = 0;
71 if ( size != 0 ) buffer = aalloc( size );
72 prods{};
73 cons{};
74 mutex_lock{};
75 closed = false;
76 #ifdef CHAN_STATS
77 blocks = 0;
78 operations = 0;
79 #endif
80}
81
82static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
83static inline void ^?{}( channel(T) &c ) with(c) {
84 #ifdef CHAN_STATS
85 printf("Channel %p Blocks: %lu, Operations: %lu, %.2f%% of ops blocked\n", &c, blocks, operations, ((double)blocks)/operations * 100);
86 #endif
87 verifyf( cons`isEmpty && prods`isEmpty, "Attempted to delete channel with waiting threads (Deadlock).\n" );
88 if ( size != 0 ) delete( buffer );
89}
90static inline size_t get_count( channel(T) & chan ) with(chan) { return __atomic_load_n( &count, __ATOMIC_RELAXED ); }
91static inline size_t get_size( channel(T) & chan ) with(chan) { return __atomic_load_n( &size, __ATOMIC_RELAXED ); }
92static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
93static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
94static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
95
96// closes the channel and notifies all blocked threads
97static inline void close( channel(T) & chan ) with(chan) {
98 lock( mutex_lock );
99 closed = true;
100
101 // flush waiting consumers and producers
102 while ( has_waiting_consumers( chan ) ) {
103 if( !__handle_waituntil_OR( cons ) ) // ensure we only signal special OR case threads when they win the race
104 break; // if __handle_waituntil_OR returns false cons is empty so break
105 cons`first.extra = 0p;
106 wake_one( cons );
107 }
108 while ( has_waiting_producers( chan ) ) {
109 if( !__handle_waituntil_OR( prods ) ) // ensure we only signal special OR case threads when they win the race
110 break; // if __handle_waituntil_OR returns false prods is empty so break
111 prods`first.extra = 0p;
112 wake_one( prods );
113 }
114 unlock(mutex_lock);
115}
116
117static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
118
119// used to hand an element to a blocked consumer and signal it
120static inline void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
121 memcpy( cons`first.extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
122 wake_one( cons );
123}
124
125// used to hand an element to a blocked producer and signal it
126static inline void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
127 memcpy( (void *)&retval, prods`first.extra, sizeof(T) );
128 wake_one( prods );
129}
130
131static inline void flush( channel(T) & chan, T elem ) with(chan) {
132 lock( mutex_lock );
133 while ( count == 0 && !cons`isEmpty ) {
134 __cons_handoff( chan, elem );
135 }
136 unlock( mutex_lock );
137}
138
139// handles buffer insert
140static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
141 memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
142 count += 1;
143 back++;
144 if ( back == size ) back = 0;
145}
146
147// needed to avoid an extra copy in closed case
148static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
149 lock( mutex_lock );
150 #ifdef CHAN_STATS
151 operations++;
152 #endif
153
154 ConsEmpty: if ( !cons`isEmpty ) {
155 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
156 __cons_handoff( chan, elem );
157 unlock( mutex_lock );
158 return true;
159 }
160
161 if ( count == size ) { unlock( mutex_lock ); return false; }
162
163 __buf_insert( chan, elem );
164 unlock( mutex_lock );
165 return true;
166}
167
168// attempts a nonblocking insert
169// returns true if insert was successful, false otherwise
170static inline bool try_insert( channel(T) & chan, T elem ) { return __internal_try_insert( chan, elem ); }
171
172// handles closed case of insert routine
173static inline void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
174 channel_closed except{ &channel_closed_vt, &elem, &chan };
175 throwResume except; // throw closed resumption
176 if ( !__internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
177}
178
179static inline void insert( channel(T) & chan, T elem ) with(chan) {
180 // check for close before acquire mx
181 if ( unlikely(closed) ) {
182 __closed_insert( chan, elem );
183 return;
184 }
185
186 lock( mutex_lock );
187
188 #ifdef CHAN_STATS
189 if ( !closed ) operations++;
190 #endif
191
192 // if closed handle
193 if ( unlikely(closed) ) {
194 unlock( mutex_lock );
195 __closed_insert( chan, elem );
196 return;
197 }
198
199 // buffer count must be zero if cons are blocked (also handles zero-size case)
200 ConsEmpty: if ( !cons`isEmpty ) {
201 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
202 __cons_handoff( chan, elem );
203 unlock( mutex_lock );
204 return;
205 }
206
207 // wait if buffer is full, work will be completed by someone else
208 if ( count == size ) {
209 #ifdef CHAN_STATS
210 blocks++;
211 #endif
212
213 // check for if woken due to close
214 if ( unlikely( block( prods, &elem, mutex_lock ) ) )
215 __closed_insert( chan, elem );
216 return;
217 } // if
218
219 __buf_insert( chan, elem );
220 unlock( mutex_lock );
221}
222
223// does the buffer remove and potentially does waiting producer work
224static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
225 memcpy( (void *)&retval, (void *)&buffer[front], sizeof(T) );
226 count -= 1;
227 front = (front + 1) % size;
228 if (count == size - 1 && !prods`isEmpty ) {
229 if ( !__handle_waituntil_OR( prods ) ) return;
230 __buf_insert( chan, *(T *)prods`first.extra ); // do waiting producer work
231 wake_one( prods );
232 }
233}
234
235// needed to avoid an extra copy in closed case and single return val case
236static inline bool __internal_try_remove( channel(T) & chan, T & retval ) with(chan) {
237 lock( mutex_lock );
238 #ifdef CHAN_STATS
239 operations++;
240 #endif
241
242 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
243 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
244 __prods_handoff( chan, retval );
245 unlock( mutex_lock );
246 return true;
247 }
248
249 if ( count == 0 ) { unlock( mutex_lock ); return false; }
250
251 __do_remove( chan, retval );
252 unlock( mutex_lock );
253 return true;
254}
255
256// attempts a nonblocking remove
257// returns [T, true] if insert was successful
258// returns [T, false] if insert was successful (T uninit)
259static inline [T, bool] try_remove( channel(T) & chan ) {
260 T retval;
261 bool success = __internal_try_remove( chan, retval );
262 return [ retval, success ];
263}
264
265static inline T try_remove( channel(T) & chan ) {
266 T retval;
267 __internal_try_remove( chan, retval );
268 return retval;
269}
270
271// handles closed case of insert routine
272static inline void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
273 channel_closed except{ &channel_closed_vt, 0p, &chan };
274 throwResume except; // throw resumption
275 if ( !__internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
276}
277
278static inline T remove( channel(T) & chan ) with(chan) {
279 T retval;
280 if ( unlikely(closed) ) {
281 __closed_remove( chan, retval );
282 return retval;
283 }
284 lock( mutex_lock );
285
286 #ifdef CHAN_STATS
287 if ( !closed ) operations++;
288 #endif
289
290 if ( unlikely(closed) ) {
291 unlock( mutex_lock );
292 __closed_remove( chan, retval );
293 return retval;
294 }
295
296 // have to check for the zero size channel case
297 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
298 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
299 __prods_handoff( chan, retval );
300 unlock( mutex_lock );
301 return retval;
302 }
303
304 // wait if buffer is empty, work will be completed by someone else
305 if ( count == 0 ) {
306 #ifdef CHAN_STATS
307 blocks++;
308 #endif
309 // check for if woken due to close
310 if ( unlikely( block( cons, &retval, mutex_lock ) ) )
311 __closed_remove( chan, retval );
312 return retval;
313 }
314
315 // Remove from buffer
316 __do_remove( chan, retval );
317 unlock( mutex_lock );
318 return retval;
319}
320
321///////////////////////////////////////////////////////////////////////////////////////////
322// The following is support for waituntil (select) statements
323///////////////////////////////////////////////////////////////////////////////////////////
324static inline bool unregister_chan( channel(T) & chan, select_node & node ) with(chan) {
325 // if ( !node`isListed && !node.park_counter ) return false; // handle special OR case C_TODO: try adding this back
326 lock( mutex_lock );
327 if ( node`isListed ) { // op wasn't performed
328 #ifdef CHAN_STATS
329 operations--;
330 #endif
331 remove( node );
332 unlock( mutex_lock );
333 return false;
334 }
335 unlock( mutex_lock );
336
337 // only return true when not special OR case, not exceptional calse and status is SAT
338 return ( node.extra == 0p || !node.park_counter ) ? false : *node.clause_status == __SELECT_SAT;
339}
340
341// type used by select statement to capture a chan read as the selected operation
342struct chan_read {
343 T & ret;
344 channel(T) & chan;
345};
346
347static inline void ?{}( chan_read(T) & cr, channel(T) & chan, T & ret ) {
348 &cr.chan = &chan;
349 &cr.ret = &ret;
350}
351static inline chan_read(T) ?<<?( T & ret, channel(T) & chan ) { chan_read(T) cr{ chan, ret }; return cr; }
352
353static inline void __handle_select_closed_read( chan_read(T) & this, select_node & node ) with(this.chan, this) {
354 __closed_remove( chan, ret );
355 // if we get here then the insert succeeded
356 __make_select_node_available( node );
357}
358
359static inline bool register_select( chan_read(T) & this, select_node & node ) with(this.chan, this) {
360 lock( mutex_lock );
361 node.extra = &ret; // set .extra so that if it == 0p later in on_selected it is due to channel close
362
363 #ifdef CHAN_STATS
364 if ( !closed ) operations++;
365 #endif
366
367 if ( !node.park_counter ) {
368 // are we special case OR and front of cons is also special case OR
369 if ( !unlikely(closed) && !prods`isEmpty && prods`first.clause_status && !prods`first.park_counter ) {
370 if ( !__make_select_node_pending( node ) ) {
371 unlock( mutex_lock );
372 return false;
373 }
374
375 if ( __handle_waituntil_OR( prods ) ) {
376 __prods_handoff( chan, ret );
377 __make_select_node_sat( node ); // need to to mark SAT now that we know operation is done or else threads could get stuck in __mark_select_node
378 unlock( mutex_lock );
379 return true;
380 }
381 __make_select_node_unsat( node );
382 }
383 // check if we can complete operation. If so race to establish winner in special OR case
384 if ( count != 0 || !prods`isEmpty || unlikely(closed) ) {
385 if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
386 unlock( mutex_lock );
387 return false;
388 }
389 }
390 }
391
392 if ( unlikely(closed) ) {
393 unlock( mutex_lock );
394 __handle_select_closed_read( this, node );
395 return true;
396 }
397
398 // have to check for the zero size channel case
399 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
400 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
401 __prods_handoff( chan, ret );
402 __set_avail_then_unlock( node, mutex_lock );
403 return true;
404 }
405
406 // wait if buffer is empty, work will be completed by someone else
407 if ( count == 0 ) {
408 #ifdef CHAN_STATS
409 blocks++;
410 #endif
411
412 insert_last( cons, node );
413 unlock( mutex_lock );
414 return false;
415 }
416
417 // Remove from buffer
418 __do_remove( chan, ret );
419 __set_avail_then_unlock( node, mutex_lock );
420 return true;
421}
422static inline bool unregister_select( chan_read(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
423static inline bool on_selected( chan_read(T) & this, select_node & node ) with(this) {
424 if ( node.extra == 0p ) // check if woken up due to closed channel
425 __closed_remove( chan, ret );
426 // This is only reachable if not closed or closed exception was handled
427 return true;
428}
429
430// type used by select statement to capture a chan write as the selected operation
431struct chan_write {
432 T elem;
433 channel(T) & chan;
434};
435
436static inline void ?{}( chan_write(T) & cw, channel(T) & chan, T elem ) {
437 &cw.chan = &chan;
438 memcpy( (void *)&cw.elem, (void *)&elem, sizeof(T) );
439}
440static inline chan_write(T) ?>>?( T elem, channel(T) & chan ) { chan_write(T) cw{ chan, elem }; return cw; }
441
442static inline void __handle_select_closed_write( chan_write(T) & this, select_node & node ) with(this.chan, this) {
443 __closed_insert( chan, elem );
444 // if we get here then the insert succeeded
445 __make_select_node_available( node );
446}
447
448static inline bool register_select( chan_write(T) & this, select_node & node ) with(this.chan, this) {
449 lock( mutex_lock );
450 node.extra = &elem; // set .extra so that if it == 0p later in on_selected it is due to channel close
451
452 #ifdef CHAN_STATS
453 if ( !closed ) operations++;
454 #endif
455
456 // special OR case handling
457 if ( !node.park_counter ) {
458 // are we special case OR and front of cons is also special case OR
459 if ( !unlikely(closed) && !cons`isEmpty && cons`first.clause_status && !cons`first.park_counter ) {
460 if ( !__make_select_node_pending( node ) ) {
461 unlock( mutex_lock );
462 return false;
463 }
464
465 if ( __handle_waituntil_OR( cons ) ) {
466 __cons_handoff( chan, elem );
467 __make_select_node_sat( node ); // need to to mark SAT now that we know operation is done or else threads could get stuck in __mark_select_node
468 unlock( mutex_lock );
469 return true;
470 }
471 __make_select_node_unsat( node );
472 }
473 // check if we can complete operation. If so race to establish winner in special OR case
474 if ( count != size || !cons`isEmpty || unlikely(closed) ) {
475 if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
476 unlock( mutex_lock );
477 return false;
478 }
479 }
480 }
481
482 // if closed handle
483 if ( unlikely(closed) ) {
484 unlock( mutex_lock );
485 __handle_select_closed_write( this, node );
486 return true;
487 }
488
489 // handle blocked consumer case via handoff (buffer is implicitly empty)
490 ConsEmpty: if ( !cons`isEmpty ) {
491 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
492 __cons_handoff( chan, elem );
493 __set_avail_then_unlock( node, mutex_lock );
494 return true;
495 }
496
497 // insert node in list if buffer is full, work will be completed by someone else
498 if ( count == size ) {
499 #ifdef CHAN_STATS
500 blocks++;
501 #endif
502
503 insert_last( prods, node );
504 unlock( mutex_lock );
505 return false;
506 } // if
507
508 // otherwise carry out write either via normal insert
509 __buf_insert( chan, elem );
510 __set_avail_then_unlock( node, mutex_lock );
511 return true;
512}
513static inline bool unregister_select( chan_write(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
514
515static inline bool on_selected( chan_write(T) & this, select_node & node ) with(this) {
516 if ( node.extra == 0p ) // check if woken up due to closed channel
517 __closed_insert( chan, elem );
518
519 // This is only reachable if not closed or closed exception was handled
520 return true;
521}
522
523} // forall( T )
524
525
Note: See TracBrowser for help on using the repository browser.