Context Navigation

-              rb195498
+              r6b33e89
 forall( T ) {
+struct __attribute__((aligned(128))) channel {
+    size_t size, front, back, count;
+    T * buffer;
+    dlist( select_node ) prods, cons; // lists of blocked threads
+    go_mutex mutex_lock;              // MX lock
+    bool closed;                      // indicates channel close/open
+    #ifdef CHAN_STATS
+    size_t p_blocks, p_ops, c_blocks, c_ops;      // counts total ops and ops resulting in a blocked thd
+    #endif
+};
+static inline void ?{}( channel(T) & this, channel(T) this2 ) = void;
+static inline void ?=?( channel(T) & this, channel(T) this2 ) = void;
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    if ( size != 0 ) buffer = aalloc( size );
+    prods{};
+    cons{};
+    mutex_lock{};
+    closed = false;
+    #ifdef CHAN_STATS
+    p_blocks = 0;
+    p_ops = 0;
+    c_blocks = 0;
+    c_ops = 0;
+    #endif
+}
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) {
+    #ifdef CHAN_STATS
+    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);
+    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);
+    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);
+    #endif
+    verifyf( __handle_waituntil_OR( cons ) || __handle_waituntil_OR( prods ) || cons`isEmpty && prods`isEmpty,
+        "Attempted to delete channel with waiting threads (Deadlock).\n" );
+    if ( size != 0 ) delete( buffer );
+}
+static inline size_t get_count( channel(T) & chan ) with(chan) { return __atomic_load_n( &count, __ATOMIC_RELAXED ); }
+static inline size_t get_size( channel(T) & chan ) with(chan) { return __atomic_load_n( &size, __ATOMIC_RELAXED ); }
+static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
+static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
+static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
+// closes the channel and notifies all blocked threads
+static inline void close( channel(T) & chan ) with(chan) {
+    lock( mutex_lock );
+    closed = true;
+    // flush waiting consumers and producers
+    while ( has_waiting_consumers( chan ) ) {
+        if( !__handle_waituntil_OR( cons ) ) // ensure we only signal special OR case threads when they win the race
+            break;  // if __handle_waituntil_OR returns false cons is empty so break
+        cons`first.extra = 0p;
+        wake_one( cons );
+    }
+    while ( has_waiting_producers( chan ) ) {
+        if( !__handle_waituntil_OR( prods ) ) // ensure we only signal special OR case threads when they win the race
+            break;  // if __handle_waituntil_OR returns false prods is empty so break
+        prods`first.extra = 0p;
+        wake_one( prods );
+    }
+    unlock(mutex_lock);
+}
+static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
+// used to hand an element to a blocked consumer and signal it
+static inline void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
+    memcpy( cons`first.extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
+    wake_one( cons );
+}
+// used to hand an element to a blocked producer and signal it
+static inline void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
+    memcpy( (void *)&retval, prods`first.extra, sizeof(T) );
+    wake_one( prods );
+}
+static inline void flush( channel(T) & chan, T elem ) with(chan) {
+    lock( mutex_lock );
+    while ( count == 0 && !cons`isEmpty ) {
+        __cons_handoff( chan, elem );
+    }
+    unlock( mutex_lock );
+}
+// handles buffer insert
+static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
+    memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+// needed to avoid an extra copy in closed case
+static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
+    lock( mutex_lock );
+    #ifdef CHAN_STATS
+    p_ops++;
+    #endif
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( chan, elem );
+        unlock( mutex_lock );
+        return true;
+    }
+    if ( count == size ) { unlock( mutex_lock ); return false; }
+    __buf_insert( chan, elem );
+    unlock( mutex_lock );
+    return true;
+}
+// attempts a nonblocking insert
+// returns true if insert was successful, false otherwise
+static inline bool try_insert( channel(T) & chan, T elem ) { return __internal_try_insert( chan, elem ); }
+// handles closed case of insert routine
+static inline void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
+    channel_closed except{ &channel_closed_vt, &elem, &chan };
+    throwResume except; // throw closed resumption
+    if ( !__internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
+}
+static inline void insert( channel(T) & chan, T elem ) with(chan) {
+    // check for close before acquire mx
+    if ( unlikely(closed) ) {
+        __closed_insert( chan, elem );
+        return;
+    }
+    lock( mutex_lock );
+    #ifdef CHAN_STATS
+    if ( !closed ) p_ops++;
+    #endif
+    // if closed handle
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __closed_insert( chan, elem );
+        return;
+    }
+    // buffer count must be zero if cons are blocked (also handles zero-size case)
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( chan, elem );
+        unlock( mutex_lock );
+        return;
+    }
+    // wait if buffer is full, work will be completed by someone else
+    if ( count == size ) {
+        #ifdef CHAN_STATS
+        p_blocks++;
+        #endif
+        // check for if woken due to close
+        if ( unlikely( block( prods, &elem, mutex_lock ) ) )
+            __closed_insert( chan, elem );
+        return;
+    } // if
+    __buf_insert( chan, elem );
+    unlock( mutex_lock );
+}
+// does the buffer remove and potentially does waiting producer work
+static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
+    memcpy( (void *)&retval, (void *)&buffer[front], sizeof(T) );
+    count -= 1;
+    front = (front + 1) % size;
+    if (count == size - 1 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) return;
+        __buf_insert( chan, *(T *)prods`first.extra );  // do waiting producer work
+        wake_one( prods );
+    }
+}
+// needed to avoid an extra copy in closed case and single return val case
+static inline bool __internal_try_remove( channel(T) & chan, T & retval ) with(chan) {
+    lock( mutex_lock );
+    #ifdef CHAN_STATS
+    c_ops++;
+    #endif
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, retval );
+        unlock( mutex_lock );
+        return true;
+    }
+    if ( count == 0 ) { unlock( mutex_lock ); return false; }
+    __do_remove( chan, retval );
+    unlock( mutex_lock );
+    return true;
+}
+// attempts a nonblocking remove
+// returns [T, true] if insert was successful
+// returns [T, false] if insert was successful (T uninit)
+static inline [T, bool] try_remove( channel(T) & chan ) {
+    T retval;
+    bool success = __internal_try_remove( chan, retval );
+    return [ retval, success ];
+}
+static inline T try_remove( channel(T) & chan ) {
+    T retval;
+    __internal_try_remove( chan, retval );
+    return retval;
+}
+// handles closed case of insert routine
+static inline void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
+    channel_closed except{ &channel_closed_vt, 0p, &chan };
+    throwResume except; // throw resumption
+    if ( !__internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
+}
+static inline T remove( channel(T) & chan ) with(chan) {
+    T retval;
+    if ( unlikely(closed) ) {
+        __closed_remove( chan, retval );
+        return retval;
+    }
+    lock( mutex_lock );
+    #ifdef CHAN_STATS
+    if ( !closed ) c_ops++;
+    #endif
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __closed_remove( chan, retval );
+        return retval;
+    }
+    // have to check for the zero size channel case
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, retval );
+        unlock( mutex_lock );
+        return retval;
+    }
+    // wait if buffer is empty, work will be completed by someone else
+    if ( count == 0 ) {
+        #ifdef CHAN_STATS
+        c_blocks++;
+        #endif
+        // check for if woken due to close
+        if ( unlikely( block( cons, &retval, mutex_lock ) ) )
+            __closed_remove( chan, retval );
+        return retval;
+    }
+    // Remove from buffer
+    __do_remove( chan, retval );
+    unlock( mutex_lock );
+    return retval;
+}
+static inline void remove( channel(T) & chan ) { T elem = (T)remove( chan ); }
+///////////////////////////////////////////////////////////////////////////////////////////
+// The following is Go-style operator support for channels
+///////////////////////////////////////////////////////////////////////////////////////////
+static inline void ?<<?( channel(T) & chan, T elem ) { insert( chan, elem ); }
+static inline void ?<<?( T & ret, channel(T) & chan ) { ret = remove( chan ); }
+///////////////////////////////////////////////////////////////////////////////////////////
+// The following is support for waituntil (select) statements
+///////////////////////////////////////////////////////////////////////////////////////////
+static inline bool unregister_chan( channel(T) & chan, select_node & node ) with(chan) {
+    if ( !node`isListed && !node.park_counter ) return false; // handle special OR case
+    lock( mutex_lock );
+    if ( node`isListed ) { // op wasn't performed
+        remove( node );
+        unlock( mutex_lock );
+        return false;
+    }
+    unlock( mutex_lock );
+    // only return true when not special OR case and status is SAT
+    return !node.park_counter ? false : *node.clause_status == __SELECT_SAT;
+}
+// special case of __handle_waituntil_OR, that does some work to avoid starvation/deadlock case
+static inline bool __handle_pending( dlist( select_node ) & queue, select_node & mine ) {
+    while ( !queue`isEmpty ) {
+        // if node not a special OR case or if we win the special OR case race break
+        if ( !queue`first.clause_status || queue`first.park_counter || __pending_set_other( queue`first, mine, ((unsigned long int)(&(queue`first))) ) )
+            return true;
+         struct __attribute__((aligned(128))) channel {
+                size_t size, front, back, count;
+                T * buffer;
+                dlist( select_node ) prods, cons;                               // lists of blocked threads
+                go_mutex mutex_lock;                                                    // MX lock
+                bool closed;                                                                    // indicates channel close/open
+                #ifdef CHAN_STATS
+                size_t p_blocks, p_ops, c_blocks, c_ops;                // counts total ops and ops resulting in a blocked thd
+            #endif
+        };
+         // type used by select statement to capture a chan read as the selected operation
+         struct chan_read {
+                 T * ret;
+                 channel(T) * chan;
+         };
+         __CFA_SELECT_GET_TYPE( chan_read(T) );
+         // type used by select statement to capture a chan read as the selected operation that doesn't have a param to read to
+         struct chan_read_no_ret {
+                 T retval;
+                 chan_read( T ) c_read;
+         };
+         __CFA_SELECT_GET_TYPE( chan_read_no_ret(T) );
+         // type used by select statement to capture a chan write as the selected operation
+         struct chan_write {
+                 T elem;
+                 channel(T) * chan;
+         };
+         __CFA_SELECT_GET_TYPE( chan_write(T) );
+} // distribution
+static inline forall( T ) {
+        void ?{}( channel(T) & this, channel(T) this2 ) = void;
+        void ?=?( channel(T) & this, channel(T) this2 ) = void;
+        void ?{}( channel(T) &c, size_t _size ) with(c) {
+                size = _size;
+                front = back = count = 0;
+                if ( size != 0 ) buffer = aalloc( size );
+                prods{};
+                cons{};
+                mutex_lock{};
+                closed = false;
+            #ifdef CHAN_STATS
+                p_blocks = 0;
+                p_ops = 0;
+                c_blocks = 0;
+                c_ops = 0;
+            #endif
+        }
+        void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+        void ^?{}( channel(T) &c ) with(c) {
+            #ifdef CHAN_STATS
+                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);
+                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);
+                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);
+            #endif
+                verifyf( __handle_waituntil_OR( cons ) || __handle_waituntil_OR( prods ) || isEmpty( cons ) && isEmpty( prods ),
+                                 "Attempted to delete channel with waiting threads (Deadlock).\n" );
+                if ( size != 0 ) delete( buffer );
+        }
+        size_t get_count( channel(T) & chan ) with(chan) { return __atomic_load_n( &count, __ATOMIC_RELAXED ); }
+        size_t get_size( channel(T) & chan ) with(chan) { return __atomic_load_n( &size, __ATOMIC_RELAXED ); }
+        bool has_waiters( channel(T) & chan ) with(chan) { return ! isEmpty( cons ) || ! isEmpty( prods ); }
+        bool has_waiting_consumers( channel(T) & chan ) with(chan) { return ! isEmpty( cons ); }
+        bool has_waiting_producers( channel(T) & chan ) with(chan) { return ! isEmpty( prods ); }
+        // closes the channel and notifies all blocked threads
+        void close( channel(T) & chan ) with(chan) {
+                lock( mutex_lock );
+                closed = true;
+                // flush waiting consumers and producers
+                while ( has_waiting_consumers( chan ) ) {
+                        if( ! __handle_waituntil_OR( cons ) ) // ensure we only signal special OR case threads when they win the race
+                                break;  // if __handle_waituntil_OR returns false cons is empty so break
+                        first( cons ).extra = 0p;
+                        wake_one( cons );
+                }
+                while ( has_waiting_producers( chan ) ) {
+                        if( ! __handle_waituntil_OR( prods ) ) // ensure we only signal special OR case threads when they win the race
+                                break;  // if __handle_waituntil_OR returns false prods is empty so break
+                        first( prods ).extra = 0p;
+                        wake_one( prods );
+                }
+                unlock(mutex_lock);
+        }
+        void is_closed( channel(T) & chan ) with(chan) { return closed; }
+        // used to hand an element to a blocked consumer and signal it
+        void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
+                memcpy( first( cons ).extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
+                wake_one( cons );
+        }
+        // used to hand an element to a blocked producer and signal it
+        void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
+                memcpy( (void *)&retval, first( prods ).extra, sizeof(T) );
+                wake_one( prods );
+        }
+        void flush( channel(T) & chan, T elem ) with(chan) {
+                lock( mutex_lock );
+                while ( count == 0 && ! isEmpty( cons ) ) {
+                        __cons_handoff( chan, elem );
+                }
+                unlock( mutex_lock );
+        }
+        // handles buffer insert
+        void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
+                memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
+                count += 1;
+                back++;
+                if ( back == size ) back = 0;
+        }
+        // needed to avoid an extra copy in closed case
+        bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
+                lock( mutex_lock );
+            #ifdef CHAN_STATS
+                p_ops++;
+            #endif
+          ConsEmpty:
+                if ( ! isEmpty( cons ) ) {
+                        if ( ! __handle_waituntil_OR( cons ) ) break ConsEmpty;
+                        __cons_handoff( chan, elem );
+                        unlock( mutex_lock );
+                        return true;
+                }
+                if ( count == size ) { unlock( mutex_lock ); return false; }
+                __buf_insert( chan, elem );
+                unlock( mutex_lock );
+                return true;
+        }
+        // attempts a nonblocking insert
+        // returns true if insert was successful, false otherwise
+        bool try_insert( channel(T) & chan, T elem ) { return __internal_try_insert( chan, elem ); }
+        // handles closed case of insert routine
+        void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
+                channel_closed except{ &channel_closed_vt, &elem, &chan };
+                throwResume except; // throw closed resumption
+                if ( ! __internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
+        }
+        void insert( channel(T) & chan, T elem ) with(chan) {
+                // check for close before acquire mx
+                if ( unlikely(closed) ) {
+                        __closed_insert( chan, elem );
+                        return;
+                }
+                lock( mutex_lock );
+            #ifdef CHAN_STATS
+                if ( ! closed ) p_ops++;
+            #endif
+                // if closed handle
+                if ( unlikely(closed) ) {
+                        unlock( mutex_lock );
+                        __closed_insert( chan, elem );
+                        return;
+                }
+                // buffer count must be zero if cons are blocked (also handles zero-size case)
+          ConsEmpty:
+                if ( ! isEmpty( cons ) ) {
+                        if ( ! __handle_waituntil_OR( cons ) ) break ConsEmpty;
+                        __cons_handoff( chan, elem );
+                        unlock( mutex_lock );
+                        return;
+                }
+                // wait if buffer is full, work will be completed by someone else
+                if ( count == size ) {
+                #ifdef CHAN_STATS
+                        p_blocks++;
+                #endif
+                        // check for if woken due to close
+                        if ( unlikely( block( prods, &elem, mutex_lock ) ) )
+                                __closed_insert( chan, elem );
+                        return;
+                } // if
+                __buf_insert( chan, elem );
+                unlock( mutex_lock );
+        }
+        // does the buffer remove and potentially does waiting producer work
+        void __do_remove( channel(T) & chan, T & retval ) with(chan) {
+                memcpy( (void *)&retval, (void *)&buffer[front], sizeof(T) );
+                count -= 1;
+                front = (front + 1) % size;
+                if (count == size - 1 && ! isEmpty( prods ) ) {
+                        if ( ! __handle_waituntil_OR( prods ) ) return;
+                        __buf_insert( chan, *(T *)first( prods ).extra );  // do waiting producer work
+                        wake_one( prods );
+                }
+        }
+        // needed to avoid an extra copy in closed case and single return val case
+        bool __internal_try_remove( channel(T) & chan, T & retval ) with(chan) {
+                lock( mutex_lock );
+            #ifdef CHAN_STATS
+                c_ops++;
+            #endif
+          ZeroSize:
+                if ( size == 0 && ! isEmpty( prods ) ) {
+                        if ( ! __handle_waituntil_OR( prods ) ) break ZeroSize;
+                        __prods_handoff( chan, retval );
+                        unlock( mutex_lock );
+                        return true;
+                }
+                if ( count == 0 ) { unlock( mutex_lock ); return false; }
+                __do_remove( chan, retval );
+                unlock( mutex_lock );
+                return true;
+        }
+        // attempts a nonblocking remove
+        // returns [T, true] if insert was successful
+        // returns [T, false] if insert was successful (T uninit)
+        [T, bool] try_remove( channel(T) & chan ) {
+                T retval;
+                bool success = __internal_try_remove( chan, retval );
+                return [ retval, success ];
+        }
+        T try_remove( channel(T) & chan ) {
+                T retval;
+                __internal_try_remove( chan, retval );
+                return retval;
+        }
+        // handles closed case of insert routine
+        void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
+                channel_closed except{ &channel_closed_vt, 0p, &chan };
+                throwResume except; // throw resumption
+                if ( ! __internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
+        }
+        T remove( channel(T) & chan ) with(chan) {
+                T retval;
+                if ( unlikely(closed) ) {
+                        __closed_remove( chan, retval );
+                        return retval;
+                }
+                lock( mutex_lock );
+            #ifdef CHAN_STATS
+                if ( ! closed ) c_ops++;
+                #endif
+                if ( unlikely(closed) ) {
+                        unlock( mutex_lock );
+                        __closed_remove( chan, retval );
+                        return retval;
+                }
+                // have to check for the zero size channel case
+          ZeroSize:
+                if ( size == 0 && ! isEmpty( prods ) ) {
+                        if ( ! __handle_waituntil_OR( prods ) ) break ZeroSize;
+                        __prods_handoff( chan, retval );
+                        unlock( mutex_lock );
+                        return retval;
+                }
+                // wait if buffer is empty, work will be completed by someone else
+                if ( count == 0 ) {
+                #ifdef CHAN_STATS
+                        c_blocks++;
+                #endif
+                        // check for if woken due to close
+                        if ( unlikely( block( cons, &retval, mutex_lock ) ) )
+                                __closed_remove( chan, retval );
+                        return retval;
+                }
+                // Remove from buffer
+                __do_remove( chan, retval );
+                unlock( mutex_lock );
+                return retval;
+        }
+        void remove( channel(T) & chan ) { T elem = (T)remove( chan ); }
+        ///////////////////////////////////////////////////////////////////////////////////////////
+        // The following is Go-style operator support for channels
+        ///////////////////////////////////////////////////////////////////////////////////////////
+        void ?<<?( channel(T) & chan, T elem ) { insert( chan, elem ); }
+        void ?<<?( T & ret, channel(T) & chan ) { ret = remove( chan ); }
+        ///////////////////////////////////////////////////////////////////////////////////////////
+        // The following is support for waituntil (select) statements
+        ///////////////////////////////////////////////////////////////////////////////////////////
+        bool unregister_chan( channel(T) & chan, select_node & node ) with(chan) {
+            if ( ! isListed( node ) && ! node.park_counter ) return false; // handle special OR case
+            lock( mutex_lock );
+            if ( isListed( node ) ) { // op wasn't performed
+                remove( node );
+                unlock( mutex_lock );
+                return false;
+            }
+            unlock( mutex_lock );
+            // only return true when not special OR case and status is SAT
+            return ! node.park_counter ? false : *node.clause_status == __SELECT_SAT;
+        }
+        // special case of __handle_waituntil_OR, that does some work to avoid starvation/deadlock case
+        bool __handle_pending( dlist( select_node ) & queue, select_node & mine ) {
+            while ( ! isEmpty( queue ) ) {
+                // if node not a special OR case or if we win the special OR case race break
+                if ( ! first( queue ).clause_status || first( queue ).park_counter || __pending_set_other( first( queue ), mine, ((unsigned long int)(&(first( queue )))) ) )
+                    return true;
+        // our node lost the race when toggling in __pending_set_other
+        if ( *mine.clause_status != __SELECT_PENDING )
+            return false;
+        // otherwise we lost the special OR race so discard node
+        try_pop_front( queue );
+    }
+    return false;
+}
+// type used by select statement to capture a chan read as the selected operation
+struct chan_read {
+    T * ret;
+    channel(T) * chan;
+};
+__CFA_SELECT_GET_TYPE( chan_read(T) );
+static inline void ?{}( chan_read(T) & cr, channel(T) * chan, T * ret ) {
+    cr.chan = chan;
+    cr.ret = ret;
+}
+static inline chan_read(T) ?<<?( T & ret, channel(T) & chan ) { chan_read(T) cr{ &chan, &ret }; return cr; }
+static inline void __handle_select_closed_read( chan_read(T) & this, select_node & node ) with(*this.chan, this) {
+    __closed_remove( *chan, *ret );
+    // if we get here then the insert succeeded
+    __make_select_node_available( node );
+}
+static inline bool register_select( chan_read(T) & this, select_node & node ) with(*this.chan, this) {
+    lock( mutex_lock );
+    node.extra = ret; // set .extra so that if it == 0p later in on_selected it is due to channel close
+    #ifdef CHAN_STATS
+    if ( !closed ) c_ops++;
+    #endif
+    if ( !node.park_counter ) {
+        // are we special case OR and front of cons is also special case OR
+        if ( !unlikely(closed) && !prods`isEmpty && prods`first.clause_status && !prods`first.park_counter ) {
+            if ( !__make_select_node_pending( node ) ) {
+                unlock( mutex_lock );
+                return false;
+            }
+            if ( __handle_pending( prods, node ) ) {
+                __prods_handoff( *chan, *ret );
+                __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
+                unlock( mutex_lock );
+                return true;
+            }
+            if ( *node.clause_status == __SELECT_PENDING )
+                __make_select_node_unsat( node );
+        }
+        // check if we can complete operation. If so race to establish winner in special OR case
+        if ( count != 0 || !prods`isEmpty || unlikely(closed) ) {
+            if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                unlock( mutex_lock );
+                return false;
+            }
+        }
+    }
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __handle_select_closed_read( this, node );
+        return true;
+    }
+    // have to check for the zero size channel case
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( *chan, *ret );
+        __set_avail_then_unlock( node, mutex_lock );
+        return true;
+    }
+    // wait if buffer is empty, work will be completed by someone else
+    if ( count == 0 ) {
+        #ifdef CHAN_STATS
+        c_blocks++;
+        #endif
+                // our node lost the race when toggling in __pending_set_other
+                if ( *mine.clause_status != __SELECT_PENDING )
+                    return false;
+                // otherwise we lost the special OR race so discard node
+                remove_first( queue );
+            }
+            return false;
+        }
+        void ?{}( chan_read(T) & cr, channel(T) * chan, T * ret ) {
+            cr.chan = chan;
+            cr.ret = ret;
+        }
+        chan_read(T) ?<<?( T & ret, channel(T) & chan ) { chan_read(T) cr{ &chan, &ret }; return cr; }
+                void __handle_select_closed_read( chan_read(T) & this, select_node & node ) with(*this.chan, this) {
+            __closed_remove( *chan, *ret );
+            // if we get here then the insert succeeded
+            __make_select_node_available( node );
+        }
+        bool register_select( chan_read(T) & this, select_node & node ) with(*this.chan, this) {
+            lock( mutex_lock );
+            node.extra = ret; // set .extra so that if it == 0p later in on_selected it is due to channel close
+            #ifdef CHAN_STATS
+            if ( ! closed ) c_ops++;
+            #endif
+            if ( ! node.park_counter ) {
+                // are we special case OR and front of cons is also special case OR
+                if ( ! unlikely(closed) && ! isEmpty( prods ) && first( prods ).clause_status && ! first( prods ).park_counter ) {
+                    if ( ! __make_select_node_pending( node ) ) {
+                        unlock( mutex_lock );
+                        return false;
+                    }
+                            if ( __handle_pending( prods, node ) ) {
+                        __prods_handoff( *chan, *ret );
+                        __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
+                        unlock( mutex_lock );
+                        return true;
+                    }
+                    if ( *node.clause_status == __SELECT_PENDING )
+                        __make_select_node_unsat( node );
+                }
+                // check if we can complete operation. If so race to establish winner in special OR case
+                if ( count != 0 || ! isEmpty( prods ) || unlikely(closed) ) {
+                    if ( ! __make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                        unlock( mutex_lock );
+                        return false;
+                    }
+                }
+            }
+            if ( unlikely(closed) ) {
+                unlock( mutex_lock );
+                __handle_select_closed_read( this, node );
+                return true;
+            }
+            // have to check for the zero size channel case
+            ZeroSize:
+                if ( size == 0 && ! isEmpty( prods ) ) {
+                        if ( ! __handle_waituntil_OR( prods ) ) break ZeroSize;
+                        __prods_handoff( *chan, *ret );
+                        __set_avail_then_unlock( node, mutex_lock );
+                        return true;
+                }
+                // wait if buffer is empty, work will be completed by someone else
+                if ( count == 0 ) {
+                #ifdef CHAN_STATS
+                c_blocks++;
+                #endif
+        insert_last( cons, node );
+        unlock( mutex_lock );
+        return false;
+    }
+    // Remove from buffer
+    __do_remove( *chan, *ret );
+    __set_avail_then_unlock( node, mutex_lock );
+    return true;
+}
+static inline bool unregister_select( chan_read(T) & this, select_node & node ) { return unregister_chan( *this.chan, node ); }
+static inline bool on_selected( chan_read(T) & this, select_node & node ) with(this) {
+    if ( unlikely(node.extra == 0p) ) {
+        if ( !exception_in_flight() ) __closed_remove( *chan, *ret ); // check if woken up due to closed channel
+        else return false;
+    }
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+// type used by select statement to capture a chan read as the selected operation that doesn't have a param to read to
+struct chan_read_no_ret {
+    T retval;
+    chan_read( T ) c_read;
+};
+__CFA_SELECT_GET_TYPE( chan_read_no_ret(T) );
+static inline void ?{}( chan_read_no_ret(T) & this, channel(T) & chan ) {
+    this.c_read{ &chan, &this.retval };
+}
+static inline chan_read_no_ret(T) remove( channel(T) & chan ) { chan_read_no_ret(T) c_read{ chan }; return c_read; }
+static inline bool register_select( chan_read_no_ret(T) & this, select_node & node ) {
+    this.c_read.ret = &this.retval;
+    return register_select( this.c_read, node );
+}
+static inline bool unregister_select( chan_read_no_ret(T) & this, select_node & node ) { return unregister_select( this.c_read, node ); }
+static inline bool on_selected( chan_read_no_ret(T) & this, select_node & node ) { return on_selected( this.c_read, node ); }
+// type used by select statement to capture a chan write as the selected operation
+struct chan_write {
+    T elem;
+    channel(T) * chan;
+};
+__CFA_SELECT_GET_TYPE( chan_write(T) );
+static inline void ?{}( chan_write(T) & cw, channel(T) * chan, T elem ) {
+    cw.chan = chan;
+    memcpy( (void *)&cw.elem, (void *)&elem, sizeof(T) );
+}
+static inline chan_write(T) ?<<?( channel(T) & chan, T elem ) { chan_write(T) cw{ &chan, elem }; return cw; }
+static inline chan_write(T) insert( T elem, channel(T) & chan) { chan_write(T) cw{ &chan, elem }; return cw; }
+static inline void __handle_select_closed_write( chan_write(T) & this, select_node & node ) with(*this.chan, this) {
+    __closed_insert( *chan, elem );
+    // if we get here then the insert succeeded
+    __make_select_node_available( node );
+}
+static inline bool register_select( chan_write(T) & this, select_node & node ) with(*this.chan, this) {
+    lock( mutex_lock );
+    node.extra = &elem; // set .extra so that if it == 0p later in on_selected it is due to channel close
+    #ifdef CHAN_STATS
+    if ( !closed ) p_ops++;
+    #endif
+    // special OR case handling
+    if ( !node.park_counter ) {
+        // are we special case OR and front of cons is also special case OR
+        if ( !unlikely(closed) && !cons`isEmpty && cons`first.clause_status && !cons`first.park_counter ) {
+            if ( !__make_select_node_pending( node ) ) {
+                unlock( mutex_lock );
+                return false;
+            }
+            if ( __handle_pending( cons, node ) ) {
+                __cons_handoff( *chan, elem );
+                __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
+                unlock( mutex_lock );
+                return true;
+            }
+            if ( *node.clause_status == __SELECT_PENDING )
+                __make_select_node_unsat( node );
+        }
+        // check if we can complete operation. If so race to establish winner in special OR case
+        if ( count != size || !cons`isEmpty || unlikely(closed) ) {
+            if ( !__make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                unlock( mutex_lock );
+                return false;
+            }
+        }
+    }
+    // if closed handle
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __handle_select_closed_write( this, node );
+        return true;
+    }
+    // handle blocked consumer case via handoff (buffer is implicitly empty)
+    ConsEmpty: if ( !cons`isEmpty ) {
+        if ( !__handle_waituntil_OR( cons ) ) break ConsEmpty;
+        __cons_handoff( *chan, elem );
+        __set_avail_then_unlock( node, mutex_lock );
+        return true;
+    }
+    // insert node in list if buffer is full, work will be completed by someone else
+    if ( count == size ) {
+        #ifdef CHAN_STATS
+        p_blocks++;
+        #endif
+        insert_last( prods, node );
+        unlock( mutex_lock );
+        return false;
+    } // if
+    // otherwise carry out write either via normal insert
+    __buf_insert( *chan, elem );
+    __set_avail_then_unlock( node, mutex_lock );
+    return true;
+}
+static inline bool unregister_select( chan_write(T) & this, select_node & node ) { return unregister_chan( *this.chan, node ); }
+static inline bool on_selected( chan_write(T) & this, select_node & node ) with(this) {
+    if ( unlikely(node.extra == 0p) ) {
+        if ( !exception_in_flight() ) __closed_insert( *chan, elem ); // check if woken up due to closed channel
+        else return false;
+    }
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+} // forall( T )
+                insert_last( cons, node );
+                unlock( mutex_lock );
+                return false;
+            }
+            // Remove from buffer
+            __do_remove( *chan, *ret );
+            __set_avail_then_unlock( node, mutex_lock );
+            return true;
+        }
+        bool unregister_select( chan_read(T) & this, select_node & node ) { return unregister_chan( *this.chan, node ); }
+        bool on_selected( chan_read(T) & this, select_node & node ) with(this) {
+            if ( unlikely(node.extra == 0p) ) {
+                if ( ! exception_in_flight() ) __closed_remove( *chan, *ret ); // check if woken up due to closed channel
+                else return false;
+            }
+            // This is only reachable if not closed or closed exception was handled
+            return true;
+        }
+        void ?{}( chan_read_no_ret(T) & this, channel(T) & chan ) {
+            this.c_read{ &chan, &this.retval };
+        }
+        chan_read_no_ret(T) remove( channel(T) & chan ) { chan_read_no_ret(T) c_read{ chan }; return c_read; }
+        bool register_select( chan_read_no_ret(T) & this, select_node & node ) {
+            this.c_read.ret = &this.retval;
+            return register_select( this.c_read, node );
+        }
+        bool unregister_select( chan_read_no_ret(T) & this, select_node & node ) { return unregister_select( this.c_read, node ); }
+        bool on_selected( chan_read_no_ret(T) & this, select_node & node ) { return on_selected( this.c_read, node ); }
+        void ?{}( chan_write(T) & cw, channel(T) * chan, T elem ) {
+            cw.chan = chan;
+            memcpy( (void *)&cw.elem, (void *)&elem, sizeof(T) );
+        }
+        chan_write(T) ?<<?( channel(T) & chan, T elem ) { chan_write(T) cw{ &chan, elem }; return cw; }
+        chan_write(T) insert( T elem, channel(T) & chan) { chan_write(T) cw{ &chan, elem }; return cw; }
+        void __handle_select_closed_write( chan_write(T) & this, select_node & node ) with(*this.chan, this) {
+            __closed_insert( *chan, elem );
+            // if we get here then the insert succeeded
+            __make_select_node_available( node );
+        }
+        bool register_select( chan_write(T) & this, select_node & node ) with(*this.chan, this) {
+            lock( mutex_lock );
+            node.extra = &elem; // set .extra so that if it == 0p later in on_selected it is due to channel close
+            #ifdef CHAN_STATS
+            if ( ! closed ) p_ops++;
+            #endif
+            // special OR case handling
+            if ( ! node.park_counter ) {
+                // are we special case OR and front of cons is also special case OR
+                if ( ! unlikely(closed) && ! isEmpty( cons ) && first( cons ).clause_status && ! first( cons ).park_counter ) {
+                    if ( ! __make_select_node_pending( node ) ) {
+                        unlock( mutex_lock );
+                        return false;
+                    }
+                    if ( __handle_pending( cons, node ) ) {
+                                        __cons_handoff( *chan, elem );
+                                        __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
+                                        unlock( mutex_lock );
+                                        return true;
+                                }
+                                if ( *node.clause_status == __SELECT_PENDING )
+                                        __make_select_node_unsat( node );
+                        }
+                        // check if we can complete operation. If so race to establish winner in special OR case
+                        if ( count != size || ! isEmpty( cons ) || unlikely(closed) ) {
+                                if ( ! __make_select_node_available( node ) ) { // we didn't win the race so give up on registering
+                                        unlock( mutex_lock );
+                                        return false;
+                                }
+                        }
+                }
+                // if closed handle
+                if ( unlikely(closed) ) {
+                        unlock( mutex_lock );
+                        __handle_select_closed_write( this, node );
+                        return true;
+                }
+                // handle blocked consumer case via handoff (buffer is implicitly empty)
+    ConsEmpty:
+                if ( ! isEmpty( cons ) ) {
+                        if ( ! __handle_waituntil_OR( cons ) ) break ConsEmpty;
+                        __cons_handoff( *chan, elem );
+                        __set_avail_then_unlock( node, mutex_lock );
+                        return true;
+                }
+                // insert node in list if buffer is full, work will be completed by someone else
+                if ( count == size ) {
+                #ifdef CHAN_STATS
+                        p_blocks++;
+                #endif
+                        insert_last( prods, node );
+                        unlock( mutex_lock );
+                        return false;
+                } // if
+                // otherwise carry out write either via normal insert
+                __buf_insert( *chan, elem );
+                __set_avail_then_unlock( node, mutex_lock );
+                return true;
+        }
+        bool unregister_select( chan_write(T) & this, select_node & node ) { return unregister_chan( *this.chan, node ); }
+        bool on_selected( chan_write(T) & this, select_node & node ) with(this) {
+                if ( unlikely(node.extra == 0p) ) {
+                        if ( ! exception_in_flight() ) __closed_insert( *chan, elem ); // check if woken up due to closed channel
+                        else return false;
+                }
+                // This is only reachable if not closed or closed exception was handled
+                return true;
+        }
+} // distribution

Note: See TracChangeset for help on using the changeset viewer.

Context Navigation

Changeset 6b33e89 for libcfa/src/concurrency/channel.hfa

Legend:

libcfa/src/concurrency/channel.hfa

Download in other formats: