| 1 | #include <locks.hfa>
|
|---|
| 2 |
|
|---|
| 3 | struct no_reacq_lock {
|
|---|
| 4 | inline exp_backoff_then_block_lock;
|
|---|
| 5 | };
|
|---|
| 6 |
|
|---|
| 7 | // have to override these by hand to get around plan 9 inheritance bug where resolver can't find the appropriate routine to call
|
|---|
| 8 | static inline void ?{}( no_reacq_lock & this ) { ((exp_backoff_then_block_lock &)this){}; }
|
|---|
| 9 | static inline bool try_lock(no_reacq_lock & this) { return try_lock(((exp_backoff_then_block_lock &)this)); }
|
|---|
| 10 | static inline void lock(no_reacq_lock & this) { lock(((exp_backoff_then_block_lock &)this)); }
|
|---|
| 11 | static inline void unlock(no_reacq_lock & this) { unlock(((exp_backoff_then_block_lock &)this)); }
|
|---|
| 12 | static inline void on_notify(no_reacq_lock & this, struct thread$ * t ) { on_notify(((exp_backoff_then_block_lock &)this), t); }
|
|---|
| 13 | static inline size_t on_wait(no_reacq_lock & this) { return on_wait(((exp_backoff_then_block_lock &)this)); }
|
|---|
| 14 | // override wakeup so that we don't reacquire the lock if using a condvar
|
|---|
| 15 | static inline void on_wakeup( no_reacq_lock & this, size_t recursion ) {}
|
|---|
| 16 |
|
|---|
| 17 | forall( T ) {
|
|---|
| 18 | struct __attribute__ ((aligned (64))) channel {
|
|---|
| 19 | size_t size;
|
|---|
| 20 | size_t front, back, count;
|
|---|
| 21 | T * buffer;
|
|---|
| 22 | fast_cond_var( no_reacq_lock ) prods, cons;
|
|---|
| 23 | no_reacq_lock mutex_lock;
|
|---|
| 24 | };
|
|---|
| 25 |
|
|---|
| 26 | static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
|
|---|
| 27 | size = _size;
|
|---|
| 28 | front = back = count = 0;
|
|---|
| 29 | buffer = anew( size );
|
|---|
| 30 | prods{};
|
|---|
| 31 | cons{};
|
|---|
| 32 | mutex_lock{};
|
|---|
| 33 | }
|
|---|
| 34 |
|
|---|
| 35 | static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
|
|---|
| 36 | static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
|
|---|
| 37 | inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
|
|---|
| 38 | inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
|
|---|
| 39 |
|
|---|
| 40 | inline void insert_( channel(T) & chan, T elem ) with(chan) {
|
|---|
| 41 | memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
|
|---|
| 42 | count += 1;
|
|---|
| 43 | back++;
|
|---|
| 44 | if ( back == size ) back = 0;
|
|---|
| 45 | }
|
|---|
| 46 |
|
|---|
| 47 |
|
|---|
| 48 | inline void insert( channel(T) & chan, T elem ) with(chan) {
|
|---|
| 49 | lock( mutex_lock );
|
|---|
| 50 |
|
|---|
| 51 | // wait if buffer is full, work will be completed by someone else
|
|---|
| 52 | if ( count == size ) {
|
|---|
| 53 | wait( prods, mutex_lock, (uintptr_t)&elem );
|
|---|
| 54 | return;
|
|---|
| 55 | } // if
|
|---|
| 56 |
|
|---|
| 57 | if ( count == 0 && !empty( prods ) )
|
|---|
| 58 | // do waiting consumer work
|
|---|
| 59 | memcpy((void *)front( prods ), (void *)&elem, sizeof(T));
|
|---|
| 60 | else insert_( chan, elem );
|
|---|
| 61 |
|
|---|
| 62 | notify_one( prods );
|
|---|
| 63 | unlock( mutex_lock );
|
|---|
| 64 | }
|
|---|
| 65 |
|
|---|
| 66 | inline T remove( channel(T) & chan ) with(chan) {
|
|---|
| 67 | lock( mutex_lock );
|
|---|
| 68 | T retval;
|
|---|
| 69 |
|
|---|
| 70 | // wait if buffer is empty, work will be completed by someone else
|
|---|
| 71 | if (count == 0) {
|
|---|
| 72 | wait( prods, mutex_lock, (uintptr_t)&retval );
|
|---|
| 73 | return retval;
|
|---|
| 74 | }
|
|---|
| 75 |
|
|---|
| 76 | // Remove from buffer
|
|---|
| 77 | memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
|
|---|
| 78 | count -= 1;
|
|---|
| 79 | front = (front + 1) % size;
|
|---|
| 80 |
|
|---|
| 81 | if (count == size - 1 && !empty( prods ) )
|
|---|
| 82 | insert_( chan, *((T *)front( prods )) ); // do waiting producer work
|
|---|
| 83 |
|
|---|
| 84 | notify_one( prods );
|
|---|
| 85 | unlock( mutex_lock );
|
|---|
| 86 | return retval;
|
|---|
| 87 | }
|
|---|
| 88 |
|
|---|
| 89 | } // forall( T )
|
|---|
