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