Changes in libcfa/src/concurrency/channel.hfa [a45e21c:beeff61e]

File:

• libcfa/src/concurrency/channel.hfa (modified) (18 diffs)

libcfa/src/concurrency/channel.hfa

@@ -4 +4 @@
 #include <list.hfa>
 #include <mutex_stmt.hfa>
-
-// link field used for threads waiting on channel
-struct wait_link {
-    // used to put wait_link on a dl queue
-    inline dlink(wait_link);
-
-    // waiting thread
-    struct thread$ * t;
-
-    // shadow field
-    void * elem;
-};
-P9_EMBEDDED( wait_link, dlink(wait_link) )
-
-static inline void ?{}( wait_link & this, thread$ * t, void * elem ) {
-    this.t = t;
-    this.elem = elem;
-}
-
-// wake one thread from the list
-static inline void wake_one( dlist( wait_link ) & queue ) {
-    wait_link & popped = try_pop_front( queue );
-    unpark( popped.t );
-}
+#include "select.hfa"
 
 // returns true if woken due to shutdown
 // blocks thread on list and releases passed lock
-static inline bool block( dlist( wait_link ) & queue, void * elem_ptr, go_mutex & lock ) {
-    wait_link w{ active_thread(), elem_ptr };
-    insert_last( queue, w );
+static inline bool block( dlist( select_node ) & queue, void * elem_ptr, go_mutex & lock ) {
+    select_node sn{ active_thread(), elem_ptr };
+    insert_last( queue, sn );
     unlock( lock );
     park();
-    return w.elem == 0p;
+    return sn.extra == 0p;
+}
+
+// Waituntil support (un)register_select helper routine
+// Sets select node avail if not special OR case and then unlocks
+static inline void __set_avail_then_unlock( select_node & node, go_mutex & mutex_lock ) {
+    if ( node.park_counter ) __make_select_node_available( node );
+    unlock( mutex_lock );
 }
 
@@ -59 +43 @@
     size_t size, front, back, count;
     T * buffer;
-    dlist( wait_link ) prods, cons; // lists of blocked threads
+    dlist( select_node ) prods, cons; // lists of blocked threads
     go_mutex mutex_lock;            // MX lock
     bool closed;                    // indicates channel close/open
@@ -70 +54 @@
     size = _size;
     front = back = count = 0;
-    buffer = aalloc( size );
+    if ( size != 0 ) buffer = aalloc( size );
     prods{};
     cons{};
@@ -87 +71 @@
     #endif
     verifyf( cons`isEmpty && prods`isEmpty, "Attempted to delete channel with waiting threads (Deadlock).\n" );
-    delete( buffer );
+    if ( size != 0 ) delete( buffer );
 }
 static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
@@ -102 +86 @@
     // flush waiting consumers and producers
     while ( has_waiting_consumers( chan ) ) {
-        cons`first.elem = 0p;
+        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 ) ) {
-        prods`first.elem = 0p;
+        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 );
     }
@@ -114 +102 @@
 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 ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
+        __cons_handoff( chan, elem );
     }
     unlock( mutex_lock );
@@ -125 +124 @@
 // handles buffer insert
 static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
-    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    memcpy( (void *)&buffer[back], (void *)&elem, sizeof(T) );
     count += 1;
     back++;
@@ -131 +130 @@
 }
 
-// does the buffer insert or hands elem directly to consumer if one is waiting
-static inline void __do_insert( channel(T) & chan, T & elem ) with(chan) {
-    if ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    } else __buf_insert( chan, elem );
-}
-
 // needed to avoid an extra copy in closed case
 static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
@@ -145 +136 @@
     operations++;
     #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; }
-    __do_insert( chan, elem );
+
+    __buf_insert( chan, elem );
     unlock( mutex_lock );
     return true;
@@ -157 +157 @@
 // 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 };
+    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
@@ -182 +182 @@
     }
 
-    // have to check for the zero size channel case
-    if ( size == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T));
-        wake_one( cons );
-        unlock( mutex_lock );
-        return true;
+    // 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;
     }
 
@@ -202 +202 @@
     } // if
 
-    if ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    } else __buf_insert( chan, elem );
-    
-    unlock( mutex_lock );
-    return;
-}
-
-// handles buffer remove
-static inline void __buf_remove( channel(T) & chan, T & retval ) with(chan) {
-    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    __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;
-}
-
-// does the buffer remove and potentially does waiting producer work
-static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
-    __buf_remove( chan, retval );
     if (count == size - 1 && !prods`isEmpty ) {
-        __buf_insert( chan, *(T *)prods`first.elem );  // do waiting producer work
+        if ( !__handle_waituntil_OR( prods ) ) return;
+        __buf_insert( chan, *(T *)prods`first.extra );  // do waiting producer work
         wake_one( prods );
     }
@@ -233 +224 @@
     operations++;
     #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 );
@@ -244 +244 @@
 static inline [T, bool] try_remove( channel(T) & chan ) {
     T retval;
-    return [ retval, __internal_try_remove( chan, retval ) ];
-}
-
-static inline T try_remove( channel(T) & chan, T elem ) {
+    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 );
@@ -255 +256 @@
 // 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 };
+    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
@@ -279 +280 @@
 
     // have to check for the zero size channel case
-    if ( size == 0 && !prods`isEmpty ) {
-        memcpy((void *)&retval, (void *)prods`first.elem, sizeof(T));
-        wake_one( prods );
+    ZeroSize: if ( size == 0 && !prods`isEmpty ) {
+        if ( !__handle_waituntil_OR( prods ) ) break ZeroSize;
+        __prods_handoff( chan, retval );
         unlock( mutex_lock );
         return retval;
@@ -287 +288 @@
 
     // wait if buffer is empty, work will be completed by someone else
-    if (count == 0) {
+    if ( count == 0 ) {
         #ifdef CHAN_STATS
         blocks++;
@@ -299 +300 @@
     // Remove from buffer
     __do_remove( chan, retval );
-
     unlock( mutex_lock );
     return retval;
 }
+
+///////////////////////////////////////////////////////////////////////////////////////////
+// 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
+        #ifdef CHAN_STATS
+        operations--;
+        #endif
+        remove( node );
+        unlock( mutex_lock );
+        return false;
+    }
+    unlock( mutex_lock );
+
+    // only return true when not special OR case, not exceptional calse and status is SAT
+    return ( node.extra == 0p || !node.park_counter ) ? false : *node.clause_status == __SELECT_SAT;
+}
+
+// type used by select statement to capture a chan read as the selected operation
+struct chan_read {
+    channel(T) & chan;
+    T & ret;
+};
+
+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) {
+    // mutex(sout) sout | "register_read";
+    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 ) operations++;
+    #endif
+
+    // check if we can complete operation. If so race to establish winner in special OR case
+    if ( !node.park_counter && ( 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
+        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 ( node.extra == 0p ) // check if woken up due to closed channel
+        __closed_remove( chan, ret );
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+
+// type used by select statement to capture a chan write as the selected operation
+struct chan_write {
+    channel(T) & chan;
+    T elem;
+};
+
+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) ?>>?( 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) {
+    // mutex(sout) sout | "register_write";
+    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 ) operations++;
+    #endif
+
+    // check if we can complete operation. If so race to establish winner in special OR case
+    if ( !node.park_counter && ( 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 ) ) {
+            // mutex(sout) sout | "empty";
+            break ConsEmpty;
+        }
+        // mutex(sout) sout | "signal";
+        __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
+        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 ( node.extra == 0p ) // check if woken up due to closed channel
+        __closed_insert( chan, elem );
+
+    // This is only reachable if not closed or closed exception was handled
+    return true;
+}
+
+
 } // forall( T )
+
+
+