source: libcfa/src/concurrency/channel.hfa @ 9319a23

#pragma once

#include <locks.hfa>
#include <list.hfa>

#define __COOP_CHANNEL
#ifdef __PREVENTION_CHANNEL
forall( T ) {
struct channel {
    size_t size, count, front, back;
    T * buffer;
    thread$ * chair;
    T * chair_elem;
    exp_backoff_then_block_lock c_lock, p_lock;
    __spinlock_t mutex_lock;
    char __padding[64]; // avoid false sharing in arrays of channels
};

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{};
}

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 chair != 0p; }

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 );

    // have to check for the zero size channel case
    if ( size == 0 && chair != 0p ) {
        memcpy((void *)chair_elem, (void *)&elem, sizeof(T));
        unpark( chair );
        chair = 0p;
        unlock( mutex_lock );
        unlock( p_lock );
        unlock( c_lock );
        return;
    }

    // wait if buffer is full, work will be completed by someone else
    if ( count == size ) {
        chair = active_thread();
        chair_elem = &elem;
        unlock( mutex_lock );
        park( );
        return;
    } // if

    if ( chair != 0p ) {
        memcpy((void *)chair_elem, (void *)&elem, sizeof(T));
        unpark( chair );
        chair = 0p;
        unlock( mutex_lock );
        unlock( p_lock );
        unlock( c_lock );
        return;
    }
    insert_( chan, elem );

    unlock( mutex_lock );
    unlock( p_lock );
}

static inline T remove( channel(T) & chan ) with(chan) {
    lock( c_lock );
    lock( mutex_lock __cfaabi_dbg_ctx2 );
    T retval;

    // have to check for the zero size channel case
    if ( size == 0 && chair != 0p ) {
        memcpy((void *)&retval, (void *)chair_elem, sizeof(T));
        unpark( chair );
        chair = 0p;
        unlock( mutex_lock );
        unlock( p_lock );
        unlock( c_lock );
        return retval;
    }

    // wait if buffer is empty, work will be completed by someone else
    if ( count == 0 ) {
        chair = active_thread();
        chair_elem = &retval;
        unlock( mutex_lock );
        park( );
        return retval;
    }

    // Remove from buffer
    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
    count -= 1;
    front++;
    if ( front == size ) front = 0;

    if ( chair != 0p ) {
        insert_( chan, *chair_elem );  // do waiting producer work
        unpark( chair );
        chair = 0p;
        unlock( mutex_lock );
        unlock( p_lock );
        unlock( c_lock );
        return retval;
    }

    unlock( mutex_lock );
    unlock( c_lock );
    return retval;
}

} // forall( T )
#endif

#ifdef __COOP_CHANNEL

// 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;
}

forall( T ) {

struct channel {
    size_t size;
    size_t front, back, count;
    T * buffer;
    dlist( wait_link ) 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 !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; }

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 wake_one( dlist( wait_link ) & queue ) {
    wait_link & popped = try_pop_front( queue );
    unpark( popped.t );
}

static inline void block( dlist( wait_link ) & queue, void * elem_ptr, exp_backoff_then_block_lock & lock ) {
    wait_link w{ active_thread(), elem_ptr };
    insert_last( queue, w );
    unlock( lock );
    park();
}

static inline void insert( channel(T) & chan, T elem ) with(chan) {
    lock( mutex_lock );

    // 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;
    }

    // wait if buffer is full, work will be completed by someone else
    if ( count == size ) {
        block( prods, &elem, mutex_lock );
        return;
    } // if

    if ( count == 0 && !cons`isEmpty ) {
        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
        wake_one( cons );
    } else insert_( chan, elem );
    
    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 && !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) {
        block( cons, &retval, mutex_lock );
        return retval;
    }

    // Remove from buffer
    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
    count -= 1;
    front = (front + 1) % size;

    if (count == size - 1 && !prods`isEmpty ) {
        insert_( chan, *(T *)prods`first.elem );  // do waiting producer work
        wake_one( prods );
    }

    unlock( mutex_lock );
    return retval;
}
} // 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