source: libcfa/src/concurrency/channel.hfa@ b51b2a6

ast-experimental
Last change on this file since b51b2a6 was 88b49bb, checked in by caparsons <caparson@…>, 2 years ago

added a small waituntil optimization and added some improvements to statistic collection
  • Property mode set to 100644
File size: 18.0 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 p_blocks, p_ops, c_blocks, c_ops; // 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 p_blocks = 0;
78 p_ops = 0;
79 c_blocks = 0;
80 c_ops = 0;
81 #endif
82}
83
84static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
85static inline void ^?{}( channel(T) &c ) with(c) {
86 #ifdef CHAN_STATS
87 printf("Channel %p Blocks: %lu,\t\tOperations: %lu,\t%.2f%% of ops blocked\n", &c, p_blocks + c_blocks, p_ops + c_ops, ((double)p_blocks + c_blocks)/(p_ops + c_ops) * 100);
88 printf("Channel %p Consumer Blocks: %lu,\tConsumer Ops: %lu,\t%.2f%% of Consumer ops blocked\n", &c, p_blocks, p_ops, ((double)p_blocks)/p_ops * 100);
89 printf("Channel %p Producer Blocks: %lu,\tProducer Ops: %lu,\t%.2f%% of Producer ops blocked\n", &c, c_blocks, c_ops, ((double)c_blocks)/c_ops * 100);
90 #endif
91 verifyf( __handle_waituntil_OR( cons ) || __handle_waituntil_OR( prods ) || cons`isEmpty && prods`isEmpty,
92 "Attempted to delete channel with waiting threads (Deadlock).\n" );
93 if ( size != 0 ) delete( buffer );
94}
95static inline size_t get_count( channel(T) & chan ) with(chan) { return __atomic_load_n( &count, __ATOMIC_RELAXED ); }
96static inline size_t get_size( channel(T) & chan ) with(chan) { return __atomic_load_n( &size, __ATOMIC_RELAXED ); }
97static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
98static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
99static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
100
101// closes the channel and notifies all blocked threads
102static inline void close( channel(T) & chan ) with(chan) {
103 lock( mutex_lock );
104 closed = true;
105
106 // flush waiting consumers and producers
107 while ( has_waiting_consumers( chan ) ) {
108 if( !__handle_waituntil_OR( cons ) ) // ensure we only signal special OR case threads when they win the race
109 break; // if __handle_waituntil_OR returns false cons is empty so break
110 cons`first.extra = 0p;
111 wake_one( cons );
112 }
113 while ( has_waiting_producers( chan ) ) {
114 if( !__handle_waituntil_OR( prods ) ) // ensure we only signal special OR case threads when they win the race
115 break; // if __handle_waituntil_OR returns false prods is empty so break
116 prods`first.extra = 0p;
117 wake_one( prods );
118 }
119 unlock(mutex_lock);
120}
121
122static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
123
124// used to hand an element to a blocked consumer and signal it
125static inline void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
126 memcpy( cons`first.extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
127 wake_one( cons );
128}
129
130// used to hand an element to a blocked producer and signal it
131static inline void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
132 memcpy( (void *)&retval, prods`first.extra, sizeof(T) );
133 wake_one( prods );
134}
135
136static inline void flush( channel(T) & chan, T elem ) with(chan) {
137 lock( mutex_lock );
138 while ( count == 0 && !cons`isEmpty ) {
139 __cons_handoff( chan, elem );
140 }
141 unlock( mutex_lock );
142}
143
144// handles buffer insert
145static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
146 memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
147 count += 1;
148 back++;
149 if ( back == size ) back = 0;
150}
151
152// needed to avoid an extra copy in closed case
153static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
154 lock( mutex_lock );
155 #ifdef CHAN_STATS
156 p_ops++;
157 #endif
158
159 ConsEmpty: if ( !cons`isEmpty ) {
160 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
161 __cons_handoff( chan, elem );
162 unlock( mutex_lock );
163 return true;
164 }
165
166 if ( count == size ) { unlock( mutex_lock ); return false; }
167
168 __buf_insert( chan, elem );
169 unlock( mutex_lock );
170 return true;
171}
172
173// attempts a nonblocking insert
174// returns true if insert was successful, false otherwise
175static inline bool try_insert( channel(T) & chan, T elem ) { return __internal_try_insert( chan, elem ); }
176
177// handles closed case of insert routine
178static inline void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
179 channel_closed except{ &channel_closed_vt, &elem, &chan };
180 throwResume except; // throw closed resumption
181 if ( !__internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
182}
183
184static inline void insert( channel(T) & chan, T elem ) with(chan) {
185 // check for close before acquire mx
186 if ( unlikely(closed) ) {
187 __closed_insert( chan, elem );
188 return;
189 }
190
191 lock( mutex_lock );
192
193 #ifdef CHAN_STATS
194 if ( !closed ) p_ops++;
195 #endif
196
197 // if closed handle
198 if ( unlikely(closed) ) {
199 unlock( mutex_lock );
200 __closed_insert( chan, elem );
201 return;
202 }
203
204 // buffer count must be zero if cons are blocked (also handles zero-size case)
205 ConsEmpty: if ( !cons`isEmpty ) {
206 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
207 __cons_handoff( chan, elem );
208 unlock( mutex_lock );
209 return;
210 }
211
212 // wait if buffer is full, work will be completed by someone else
213 if ( count == size ) {
214 #ifdef CHAN_STATS
215 p_blocks++;
216 #endif
217
218 // check for if woken due to close
219 if ( unlikely( block( prods, &elem, mutex_lock ) ) )
220 __closed_insert( chan, elem );
221 return;
222 } // if
223
224 __buf_insert( chan, elem );
225 unlock( mutex_lock );
226}
227
228// does the buffer remove and potentially does waiting producer work
229static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
230 memcpy( (void *)&retval, (void *)&buffer[front], sizeof(T) );
231 count -= 1;
232 front = (front + 1) % size;
233 if (count == size - 1 && !prods`isEmpty ) {
234 if ( !__handle_waituntil_OR( prods ) ) return;
235 __buf_insert( chan, *(T *)prods`first.extra ); // do waiting producer work
236 wake_one( prods );
237 }
238}
239
240// needed to avoid an extra copy in closed case and single return val case
241static inline bool __internal_try_remove( channel(T) & chan, T & retval ) with(chan) {
242 lock( mutex_lock );
243 #ifdef CHAN_STATS
244 c_ops++;
245 #endif
246
247 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
248 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
249 __prods_handoff( chan, retval );
250 unlock( mutex_lock );
251 return true;
252 }
253
254 if ( count == 0 ) { unlock( mutex_lock ); return false; }
255
256 __do_remove( chan, retval );
257 unlock( mutex_lock );
258 return true;
259}
260
261// attempts a nonblocking remove
262// returns [T, true] if insert was successful
263// returns [T, false] if insert was successful (T uninit)
264static inline [T, bool] try_remove( channel(T) & chan ) {
265 T retval;
266 bool success = __internal_try_remove( chan, retval );
267 return [ retval, success ];
268}
269
270static inline T try_remove( channel(T) & chan ) {
271 T retval;
272 __internal_try_remove( chan, retval );
273 return retval;
274}
275
276// handles closed case of insert routine
277static inline void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
278 channel_closed except{ &channel_closed_vt, 0p, &chan };
279 throwResume except; // throw resumption
280 if ( !__internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
281}
282
283static inline T remove( channel(T) & chan ) with(chan) {
284 T retval;
285 if ( unlikely(closed) ) {
286 __closed_remove( chan, retval );
287 return retval;
288 }
289 lock( mutex_lock );
290
291 #ifdef CHAN_STATS
292 if ( !closed ) c_ops++;
293 #endif
294
295 if ( unlikely(closed) ) {
296 unlock( mutex_lock );
297 __closed_remove( chan, retval );
298 return retval;
299 }
300
301 // have to check for the zero size channel case
302 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
303 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
304 __prods_handoff( chan, retval );
305 unlock( mutex_lock );
306 return retval;
307 }
308
309 // wait if buffer is empty, work will be completed by someone else
310 if ( count == 0 ) {
311 #ifdef CHAN_STATS
312 c_blocks++;
313 #endif
314 // check for if woken due to close
315 if ( unlikely( block( cons, &retval, mutex_lock ) ) )
316 __closed_remove( chan, retval );
317 return retval;
318 }
319
320 // Remove from buffer
321 __do_remove( chan, retval );
322 unlock( mutex_lock );
323 return retval;
324}
325
326///////////////////////////////////////////////////////////////////////////////////////////
327// The following is support for waituntil (select) statements
328///////////////////////////////////////////////////////////////////////////////////////////
329static inline bool unregister_chan( channel(T) & chan, select_node & node ) with(chan) {
330 if ( !node`isListed && !node.park_counter ) return false; // handle special OR case
331 lock( mutex_lock );
332 if ( node`isListed ) { // op wasn't performed
333 remove( node );
334 unlock( mutex_lock );
335 return false;
336 }
337 unlock( mutex_lock );
338
339 // only return true when not special OR case, not exceptional calse and status is SAT
340 return ( node.extra == 0p || !node.park_counter ) ? false : *node.clause_status == __SELECT_SAT;
341}
342
343// type used by select statement to capture a chan read as the selected operation
344struct chan_read {
345 T & ret;
346 channel(T) & chan;
347};
348
349static inline void ?{}( chan_read(T) & cr, channel(T) & chan, T & ret ) {
350 &cr.chan = &chan;
351 &cr.ret = &ret;
352}
353static inline chan_read(T) ?<<?( T & ret, channel(T) & chan ) { chan_read(T) cr{ chan, ret }; return cr; }
354
355static inline void __handle_select_closed_read( chan_read(T) & this, select_node & node ) with(this.chan, this) {
356 __closed_remove( chan, ret );
357 // if we get here then the insert succeeded
358 __make_select_node_available( node );
359}
360
361static inline bool register_select( chan_read(T) & this, select_node & node ) with(this.chan, this) {
362 lock( mutex_lock );
363 node.extra = &ret; // set .extra so that if it == 0p later in on_selected it is due to channel close
364
365 #ifdef CHAN_STATS
366 if ( !closed ) c_ops++;
367 #endif
368
369 if ( !node.park_counter ) {
370 // are we special case OR and front of cons is also special case OR
371 if ( !unlikely(closed) && !prods`isEmpty && prods`first.clause_status && !prods`first.park_counter ) {
372 if ( !__make_select_node_pending( node ) ) {
373 unlock( mutex_lock );
374 return false;
375 }
376
377 if ( __handle_waituntil_OR( prods ) ) {
378 __prods_handoff( chan, ret );
379 __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
380 unlock( mutex_lock );
381 return true;
382 }
383 __make_select_node_unsat( node );
384 }
385 // check if we can complete operation. If so race to establish winner in special OR case
386 if ( count != 0 || !prods`isEmpty || unlikely(closed) ) {
387 if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
388 unlock( mutex_lock );
389 return false;
390 }
391 }
392 }
393
394 if ( unlikely(closed) ) {
395 unlock( mutex_lock );
396 __handle_select_closed_read( this, node );
397 return true;
398 }
399
400 // have to check for the zero size channel case
401 ZeroSize: if ( size == 0 && !prods`isEmpty ) {
402 if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
403 __prods_handoff( chan, ret );
404 __set_avail_then_unlock( node, mutex_lock );
405 return true;
406 }
407
408 // wait if buffer is empty, work will be completed by someone else
409 if ( count == 0 ) {
410 #ifdef CHAN_STATS
411 c_blocks++;
412 #endif
413
414 insert_last( cons, node );
415 unlock( mutex_lock );
416 return false;
417 }
418
419 // Remove from buffer
420 __do_remove( chan, ret );
421 __set_avail_then_unlock( node, mutex_lock );
422 return true;
423}
424static inline bool unregister_select( chan_read(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
425static inline bool on_selected( chan_read(T) & this, select_node & node ) with(this) {
426 if ( node.extra == 0p ) // check if woken up due to closed channel
427 __closed_remove( chan, ret );
428 // This is only reachable if not closed or closed exception was handled
429 return true;
430}
431
432// type used by select statement to capture a chan write as the selected operation
433struct chan_write {
434 T elem;
435 channel(T) & chan;
436};
437
438static inline void ?{}( chan_write(T) & cw, channel(T) & chan, T elem ) {
439 &cw.chan = &chan;
440 memcpy( (void *)&cw.elem, (void *)&elem, sizeof(T) );
441}
442static inline chan_write(T) ?>>?( T elem, channel(T) & chan ) { chan_write(T) cw{ chan, elem }; return cw; }
443
444static inline void __handle_select_closed_write( chan_write(T) & this, select_node & node ) with(this.chan, this) {
445 __closed_insert( chan, elem );
446 // if we get here then the insert succeeded
447 __make_select_node_available( node );
448}
449
450static inline bool register_select( chan_write(T) & this, select_node & node ) with(this.chan, this) {
451 lock( mutex_lock );
452 node.extra = &elem; // set .extra so that if it == 0p later in on_selected it is due to channel close
453
454 #ifdef CHAN_STATS
455 if ( !closed ) p_ops++;
456 #endif
457
458 // special OR case handling
459 if ( !node.park_counter ) {
460 // are we special case OR and front of cons is also special case OR
461 if ( !unlikely(closed) && !cons`isEmpty && cons`first.clause_status && !cons`first.park_counter ) {
462 if ( !__make_select_node_pending( node ) ) {
463 unlock( mutex_lock );
464 return false;
465 }
466
467 if ( __handle_waituntil_OR( cons ) ) {
468 __cons_handoff( chan, elem );
469 __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
470 unlock( mutex_lock );
471 return true;
472 }
473 __make_select_node_unsat( node );
474 }
475 // check if we can complete operation. If so race to establish winner in special OR case
476 if ( count != size || !cons`isEmpty || unlikely(closed) ) {
477 if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
478 unlock( mutex_lock );
479 return false;
480 }
481 }
482 }
483
484 // if closed handle
485 if ( unlikely(closed) ) {
486 unlock( mutex_lock );
487 __handle_select_closed_write( this, node );
488 return true;
489 }
490
491 // handle blocked consumer case via handoff (buffer is implicitly empty)
492 ConsEmpty: if ( !cons`isEmpty ) {
493 if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
494 __cons_handoff( chan, elem );
495 __set_avail_then_unlock( node, mutex_lock );
496 return true;
497 }
498
499 // insert node in list if buffer is full, work will be completed by someone else
500 if ( count == size ) {
501 #ifdef CHAN_STATS
502 p_blocks++;
503 #endif
504
505 insert_last( prods, node );
506 unlock( mutex_lock );
507 return false;
508 } // if
509
510 // otherwise carry out write either via normal insert
511 __buf_insert( chan, elem );
512 __set_avail_then_unlock( node, mutex_lock );
513 return true;
514}
515static inline bool unregister_select( chan_write(T) & this, select_node & node ) { return unregister_chan( this.chan, node ); }
516
517static inline bool on_selected( chan_write(T) & this, select_node & node ) with(this) {
518 if ( node.extra == 0p ) // check if woken up due to closed channel
519 __closed_insert( chan, elem );
520
521 // This is only reachable if not closed or closed exception was handled
522 return true;
523}
524
525} // forall( T )
526
527
Note: See TracBrowser for help on using the repository browser.