Changeset b110bcc for libcfa/src/concurrency/channel.hfa

Timestamp:

Apr 21, 2023, 5:36:12 PM (2 years ago)

Author:

JiadaL <j82liang@…>

Branches:

ADT, master

Children:

28f8f15, 6e4c44d

Parents:

2ed94a9 (diff), 699a97d (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

File:

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

libcfa/src/concurrency/channel.hfa

@@ +1 @@
+#pragma once
+
 #include <locks.hfa>
-
-struct no_reacq_lock {
-    inline exp_backoff_then_block_lock;
+#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;
 };
-
-// have to override these by hand to get around plan 9 inheritance bug where resolver can't find the appropriate routine to call
-static inline void   ?{}( no_reacq_lock & this ) { ((exp_backoff_then_block_lock &)this){}; }
-static inline bool   try_lock(no_reacq_lock & this) { return try_lock(((exp_backoff_then_block_lock &)this)); }
-static inline void   lock(no_reacq_lock & this) { lock(((exp_backoff_then_block_lock &)this)); }
-static inline void   unlock(no_reacq_lock & this) { unlock(((exp_backoff_then_block_lock &)this)); }
-static inline void   on_notify(no_reacq_lock & this, struct thread$ * t ) { on_notify(((exp_backoff_then_block_lock &)this), t); }
-static inline size_t on_wait(no_reacq_lock & this) { return on_wait(((exp_backoff_then_block_lock &)this)); }
-// override wakeup so that we don't reacquire the lock if using a condvar
-static inline void   on_wakeup( no_reacq_lock & this, size_t recursion ) {}
+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 );
+}
+
+// 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 );
+    unlock( lock );
+    park();
+    return w.elem == 0p;
+}
+
+// void * used for some fields since exceptions don't work with parametric polymorphism currently
+exception channel_closed {
+    // on failed insert elem is a ptr to the element attempting to be inserted
+    // on failed remove elem ptr is 0p
+    // on resumption of a failed insert this elem will be inserted
+    // so a user may modify it in the resumption handler
+    void * elem;
+
+    // pointer to chan that is closed
+    void * closed_chan;
+};
+vtable(channel_closed) channel_closed_vt;
+
+// #define CHAN_STATS // define this to get channel stats printed in dtor
 
 forall( T ) {
-struct channel {
-    size_t size;
-    size_t front, back, count;
+
+struct __attribute__((aligned(128))) channel {
+    size_t size, front, back, count;
     T * buffer;
-    fast_cond_var( no_reacq_lock ) prods, cons;
-    no_reacq_lock mutex_lock;
+    dlist( wait_link ) prods, cons; // lists of blocked threads
+    go_mutex mutex_lock;            // MX lock
+    bool closed;                    // indicates channel close/open
+    #ifdef CHAN_STATS
+    size_t blocks, operations;      // counts total ops and ops resulting in a blocked thd
+    #endif
 };
 
@@ -27 +70 @@
     size = _size;
     front = back = count = 0;
-    buffer = anew( size );
+    buffer = aalloc( size );
     prods{};
     cons{};
     mutex_lock{};
+    closed = false;
+    #ifdef CHAN_STATS
+    blocks = 0;
+    operations = 0;
+    #endif
 }
 
 static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
-static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
+static inline void ^?{}( channel(T) &c ) with(c) {
+    #ifdef CHAN_STATS
+    printf("Channel %p Blocks: %lu, Operations: %lu, %.2f%% of ops blocked\n", &c, blocks, operations, ((double)blocks)/operations * 100);
+    #endif
+    verifyf( cons`isEmpty && prods`isEmpty, "Attempted to delete channel with waiting threads (Deadlock).\n" );
+    delete( buffer );
+}
 static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
 static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
-static inline bool has_waiters( channel(T) & chan ) with(chan) { return !empty( cons ) || !empty( prods ); }
-static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !empty( cons ); }
-static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !empty( prods ); }
-
-static inline void insert_( channel(T) & chan, T elem ) with(chan) {
+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 ) ) {
+        cons`first.elem = 0p;
+        wake_one( cons );
+    }
+    while ( has_waiting_producers( chan ) ) {
+        prods`first.elem = 0p;
+        wake_one( prods );
+    }
+    unlock(mutex_lock);
+}
+
+static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
+
+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 );
+    }
+    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;
@@ -48 +131 @@
 }
 
+// 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) {
+    lock( mutex_lock );
+    #ifdef CHAN_STATS
+    operations++;
+    #endif
+    if ( count == size ) { unlock( mutex_lock ); return false; }
+    __do_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) {
-    lock( mutex_lock );
+    // check for close before acquire mx
+    if ( unlikely(closed) ) {
+        __closed_insert( chan, elem );
+        return;
+    } 
+
+    lock( mutex_lock );
+
+    #ifdef CHAN_STATS
+    if ( !closed ) operations++;
+    #endif
+
+    // if closed handle
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __closed_insert( chan, elem );
+        return;
+    }
 
     // have to check for the zero size channel case
-    if ( size == 0 && !empty( cons ) ) {
-        memcpy((void *)front( cons ), (void *)&elem, sizeof(T));
-        notify_one( cons );
+    if ( size == 0 && !cons`isEmpty ) {
+        memcpy(cons`first.elem, (void *)&elem, sizeof(T));
+        wake_one( cons );
         unlock( mutex_lock );
-        return;
+        return true;
     }
 
     // wait if buffer is full, work will be completed by someone else
-    if ( count == size ) { 
-        wait( prods, mutex_lock, (uintptr_t)&elem );
+    if ( count == size ) {
+        #ifdef CHAN_STATS
+        blocks++;
+        #endif
+
+        // check for if woken due to close
+        if ( unlikely( block( prods, &elem, mutex_lock ) ) )
+            __closed_insert( chan, elem );
         return;
     } // if
 
-    if ( count == 0 && !empty( cons ) )
-        // do waiting consumer work
-        memcpy((void *)front( cons ), (void *)&elem, sizeof(T)); 
-    else insert_( chan, elem );
+    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 );
     
-    notify_one( cons );
-    unlock( mutex_lock );
-}
-
-static inline T remove( channel(T) & chan ) with(chan) {
-    lock( mutex_lock );
-    T retval;
-
-    // have to check for the zero size channel case
-    if ( size == 0 && !empty( prods ) ) {
-        memcpy((void *)&retval, (void *)front( prods ), sizeof(T));
-        notify_one( prods );
-        unlock( mutex_lock );
-        return retval;
-    }
-
-    // wait if buffer is empty, work will be completed by someone else
-    if (count == 0) { 
-        wait( cons, mutex_lock, (uintptr_t)&retval );
-        return retval;
-    }
-
-    // Remove from buffer
+    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));
     count -= 1;
     front = (front + 1) % size;
-
-    if (count == size - 1 && !empty( prods ) ) 
-        insert_( chan, *((T *)front( prods )) );  // do waiting producer work
-
-    notify_one( prods );
-    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) {
+    __buf_remove( chan, retval );
+    if (count == size - 1 && !prods`isEmpty ) {
+        __buf_insert( chan, *(T *)prods`first.elem );  // 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
+    operations++;
+    #endif
+    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;
+    return [ retval, __internal_try_remove( chan, retval ) ];
+}
+
+static inline T try_remove( channel(T) & chan, T elem ) {
+    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 ) operations++;
+    #endif
+
+    if ( unlikely(closed) ) {
+        unlock( mutex_lock );
+        __closed_remove( chan, retval );
+        return retval;
+    } 
+
+    // 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 );
+        unlock( mutex_lock );
+        return retval;
+    }
+
+    // wait if buffer is empty, work will be completed by someone else
+    if (count == 0) {
+        #ifdef CHAN_STATS
+        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;
+}
 } // forall( T )