source: libcfa/src/concurrency/channel.hfa @ 5908fb4

ADTast-experimental
Last change on this file since 5908fb4 was 5908fb4, checked in by caparsons <caparson@…>, 9 months 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.