Changes in / [1633e04:75d874a]

File:

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

libcfa/src/concurrency/channel.hfa

@@ -4 +4 @@
 #include <list.hfa>
 
-// #define __PREVENTION_CHANNEL
+#define __COOP_CHANNEL
 #ifdef __PREVENTION_CHANNEL
 forall( T ) {
@@ -14 +14 @@
     exp_backoff_then_block_lock c_lock, p_lock;
     __spinlock_t mutex_lock;
-    char __padding[64]; // avoid false sharing in arrays
+    char __padding[64]; // avoid false sharing in arrays of channels
 };
 
@@ -20 +20 @@
     size = _size;
     front = back = count = 0;
-    buffer = anew( size );
+    buffer = aalloc( size );
     chair = 0p;
     mutex_lock{};
@@ -128 +128 @@
 #endif
 
-#ifndef __PREVENTION_CHANNEL
+#ifdef __COOP_CHANNEL
 
 // link field used for threads waiting on channel
@@ -161 +161 @@
     size = _size;
     front = back = count = 0;
-    buffer = anew( size );
+    buffer = aalloc( size );
     prods{};
     cons{};
@@ -252 +252 @@
 } // forall( T )
 #endif
+
+#ifdef __BARGE_CHANNEL
+forall( T ) {
+struct channel {
+    size_t size;
+    size_t front, back, count;
+    T * buffer;
+    fast_cond_var( exp_backoff_then_block_lock ) prods, cons;
+    exp_backoff_then_block_lock mutex_lock;
+};
+
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    buffer = aalloc( size );
+    prods{};
+    cons{};
+    mutex_lock{};
+}
+
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) { 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) {
+    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+
+
+static inline void insert( channel(T) & chan, T elem ) with(chan) {
+    lock( mutex_lock );
+
+    while ( count == size ) { 
+        wait( prods, mutex_lock );
+    } // if
+
+    insert_( chan, elem );
+    
+    if ( !notify_one( cons ) && count < size )
+        notify_one( prods );
+
+    unlock( mutex_lock );
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
+    lock( mutex_lock );
+    T retval;
+
+    while (count == 0) { 
+        wait( cons, mutex_lock );
+    }
+
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front = (front + 1) % size;
+
+    if ( !notify_one( prods ) && count > 0 )
+        notify_one( cons );
+
+    unlock( mutex_lock );
+    return retval;
+}
+
+} // forall( T )
+#endif
+
+#ifdef __NO_WAIT_CHANNEL
+forall( T ) {
+struct channel {
+    size_t size;
+    size_t front, back, count;
+    T * buffer;
+    thread$ * chair;
+    T * chair_elem;
+    exp_backoff_then_block_lock c_lock, p_lock;
+    __spinlock_t mutex_lock;
+};
+
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    buffer = aalloc( size );
+    chair = 0p;
+    mutex_lock{};
+    c_lock{};
+    p_lock{};
+    lock( c_lock );
+}
+
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) { 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 c_lock.lock_value != 0; }
+
+static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
+    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+
+static inline void insert( channel(T) & chan, T elem ) with( chan ) {
+    lock( p_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+
+    insert_( chan, elem );
+
+    if ( count != size )
+        unlock( p_lock );
+
+    if ( count == 1 )
+        unlock( c_lock );
+        
+    unlock( mutex_lock );
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
+    lock( c_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+    T retval;
+
+    // Remove from buffer
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front = (front + 1) % size;
+
+    if ( count != 0 )
+        unlock( c_lock );
+
+    if ( count == size - 1 )
+        unlock( p_lock );
+        
+    unlock( mutex_lock );
+    return retval;
+}
+
+} // forall( T )
+#endif