Ignore:
File:
1 edited

Legend:

Unmodified
Added
Removed

  • libcfa/src/concurrency/channel.hfa

    rd30e3eb r1d245ea  
    22
    33#include <locks.hfa>
    4 #include <list.hfa>
    5 
    6 // #define __PREVENTION_CHANNEL
     4
     5struct no_reacq_lock {
     6    inline exp_backoff_then_block_lock;
     7};
     8
     9// have to override these by hand to get around plan 9 inheritance bug where resolver can't find the appropriate routine to call
     10static inline void   ?{}( no_reacq_lock & this ) { ((exp_backoff_then_block_lock &)this){}; }
     11static inline bool   try_lock(no_reacq_lock & this) { return try_lock(((exp_backoff_then_block_lock &)this)); }
     12static inline void   lock(no_reacq_lock & this) { lock(((exp_backoff_then_block_lock &)this)); }
     13static inline void   unlock(no_reacq_lock & this) { unlock(((exp_backoff_then_block_lock &)this)); }
     14static inline void   on_notify(no_reacq_lock & this, struct thread$ * t ) { on_notify(((exp_backoff_then_block_lock &)this), t); }
     15static inline size_t on_wait(no_reacq_lock & this) { return on_wait(((exp_backoff_then_block_lock &)this)); }
     16// override wakeup so that we don't reacquire the lock if using a condvar
     17static inline void   on_wakeup( no_reacq_lock & this, size_t recursion ) {}
     18
     19#define __PREVENTION_CHANNEL
    720#ifdef __PREVENTION_CHANNEL
    821forall( T ) {
    922struct channel {
    10     size_t size, count, front, back;
     23    size_t size;
     24    size_t front, back, count;
    1125    T * buffer;
    1226    thread$ * chair;
    … …  
    7387        return;
    7488    }
    75     insert_( chan, elem );
     89    else insert_( chan, elem );
    7690
    7791    unlock( mutex_lock );
    … …  
    96110
    97111    // wait if buffer is empty, work will be completed by someone else
    98     if ( count == 0 ) {
     112    if ( count == 0 ) { 
    99113        chair = active_thread();
    100114        chair_elem = &retval;
    … …  
    107121    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
    108122    count -= 1;
    109     front++;
    110     if ( front == size ) front = 0;
     123    front = (front + 1) % size;
    111124
    112125    if ( chair != 0p ) {
    … …  
    129142
    130143#ifndef __PREVENTION_CHANNEL
    131 
    132 // link field used for threads waiting on channel
    133 struct wait_link {
    134     // used to put wait_link on a dl queue
    135     inline dlink(wait_link);
    136 
    137     // waiting thread
    138     struct thread$ * t;
    139 
    140     // shadow field
    141     void * elem;
    142 };
    143 P9_EMBEDDED( wait_link, dlink(wait_link) )
    144 
    145 static inline void ?{}( wait_link & this, thread$ * t, void * elem ) {
    146     this.t = t;
    147     this.elem = elem;
    148 }
    149 
    150144forall( T ) {
    151 
    152145struct channel {
    153146    size_t size;
    154147    size_t front, back, count;
    155148    T * buffer;
    156     dlist( wait_link ) prods, cons;
    157     exp_backoff_then_block_lock mutex_lock;
     149    fast_cond_var( no_reacq_lock ) prods, cons;
     150    no_reacq_lock mutex_lock;
    158151};
    159152
    … …  
    171164static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
    172165static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
    173 static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
    174 static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
    175 static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
     166static inline bool has_waiters( channel(T) & chan ) with(chan) { return !empty( cons ) || !empty( prods ); }
     167static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !empty( cons ); }
     168static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !empty( prods ); }
    176169
    177170static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
    … …  
    182175}
    183176
    184 static inline void wake_one( dlist( wait_link ) & queue ) {
    185     wait_link & popped = try_pop_front( queue );
    186     unpark( popped.t );
    187 }
    188 
    189 static inline void block( dlist( wait_link ) & queue, void * elem_ptr, exp_backoff_then_block_lock & lock ) {
    190     wait_link w{ active_thread(), elem_ptr };
    191     insert_last( queue, w );
    192     unlock( lock );
    193     park();
    194 }
    195177
    196178static inline void insert( channel(T) & chan, T elem ) with(chan) {
    … …  
    198180
    199181    // have to check for the zero size channel case
    200     if ( size == 0 && !cons`isEmpty ) {
    201         memcpy(cons`first.elem, (void *)&elem, sizeof(T));
    202         wake_one( cons );
     182    if ( size == 0 && !empty( cons ) ) {
     183        memcpy((void *)front( cons ), (void *)&elem, sizeof(T));
     184        notify_one( cons );
    203185        unlock( mutex_lock );
    204186        return;
    … …  
    206188
    207189    // wait if buffer is full, work will be completed by someone else
    208     if ( count == size ) {
    209         block( prods, &elem, mutex_lock );
     190    if ( count == size ) { 
     191        wait( prods, mutex_lock, (uintptr_t)&elem );
    210192        return;
    211193    } // if
    212194
    213     if ( count == 0 && !cons`isEmpty ) {
    214         memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
    215         wake_one( cons );
    216     } else insert_( chan, elem );
     195    if ( count == 0 && !empty( cons ) )
     196        // do waiting consumer work
     197        memcpy((void *)front( cons ), (void *)&elem, sizeof(T));
     198    else insert_( chan, elem );
    217199   
     200    notify_one( cons );
    218201    unlock( mutex_lock );
    219202}
    … …  
    224207
    225208    // have to check for the zero size channel case
    226     if ( size == 0 && !prods`isEmpty ) {
    227         memcpy((void *)&retval, (void *)prods`first.elem, sizeof(T));
    228         wake_one( prods );
     209    if ( size == 0 && !empty( prods ) ) {
     210        memcpy((void *)&retval, (void *)front( prods ), sizeof(T));
     211        notify_one( prods );
    229212        unlock( mutex_lock );
    230213        return retval;
    … …  
    232215
    233216    // wait if buffer is empty, work will be completed by someone else
    234     if (count == 0) {
    235         block( cons, &retval, mutex_lock );
     217    if (count == 0) { 
     218        wait( cons, mutex_lock, (uintptr_t)&retval );
    236219        return retval;
    237220    }
    … …  
    242225    front = (front + 1) % size;
    243226
    244     if (count == size - 1 && !prods`isEmpty ) {
    245         insert_( chan, *(T *)prods`first.elem );  // do waiting producer work
    246         wake_one( prods );
    247     }
    248 
     227    if (count == size - 1 && !empty( prods ) )
     228        insert_( chan, *((T *)front( prods )) );  // do waiting producer work
     229
     230    notify_one( prods );
    249231    unlock( mutex_lock );
    250232    return retval;
    251233}
     234
    252235} // forall( T )
    253236#endif
Note: See TracChangeset for help on using the changeset viewer.