BNX2X: disables TPA
[akaros.git] / kern / src / apipe.c
1 /* Copyright (c) 2013 The Regents of the University of California
2  * Barret Rhoden <brho@cs.berkeley.edu>
3  * See LICENSE for details.
4  *
5  * Atomic pipes.  Multi-reader, multi-writer pipes, similar to sys_pipe except
6  * that they operate on fixed sized chunks of data.
7  *
8  * A note on broadcast wakeups.  We broadcast in a few places.  If we don't,
9  * then all paths (like error paths) will have to signal.  Not a big deal
10  * either way, but just need to catch all the cases.  Other non-obvious
11  * cases are that read and write methods need to wake other readers and
12  * writers (in the absence of a broadcast wakeup) */
13
14 #include <apipe.h>
15 #include <ros/ring_buffer.h>
16 #include <string.h>
17 #include <stdio.h>
18
19 void apipe_init(struct atomic_pipe *ap, void *buf, size_t buf_sz,
20                 size_t elem_sz)
21 {
22         ap->ap_buf = buf;
23         /* power of two number of elements in the ring. */
24         ap->ap_ring_sz = ROUNDDOWNPWR2(buf_sz / elem_sz);
25         ap->ap_elem_sz = elem_sz;
26         ap->ap_rd_off = 0;
27         ap->ap_wr_off = 0;
28         ap->ap_nr_readers = 1;
29         ap->ap_nr_writers = 1;
30         /* Three CVs, all using the same lock. */
31         spinlock_init(&ap->ap_lock);
32         cv_init_with_lock(&ap->ap_priority_reader, &ap->ap_lock);
33         cv_init_with_lock(&ap->ap_general_readers, &ap->ap_lock);
34         cv_init_with_lock(&ap->ap_writers, &ap->ap_lock);
35         ap->ap_has_priority_reader = FALSE;
36 }
37
38 void apipe_open_reader(struct atomic_pipe *ap)
39 {
40         spin_lock(&ap->ap_lock);
41         ap->ap_nr_readers++;
42         spin_unlock(&ap->ap_lock);
43 }
44
45 void apipe_open_writer(struct atomic_pipe *ap)
46 {
47         spin_lock(&ap->ap_lock);
48         ap->ap_nr_writers++;
49         spin_unlock(&ap->ap_lock);
50 }
51
52 /* Helper: Wake the appropriate readers.  When there's a priority reader, only
53  * that one wakes up.  It's up to the priority reader to wake the other readers,
54  * by clearing has_prior and calling this again. */
55 static void __apipe_wake_readers(struct atomic_pipe *ap)
56 {
57         if (ap->ap_has_priority_reader)
58                 __cv_signal(&ap->ap_priority_reader);
59         else
60                 __cv_broadcast(&ap->ap_general_readers);
61 }
62
63 /* When closing, there might be others blocked waiting for us.  For example,
64  * a writer could have blocked on a full pipe, waiting for us to read.  Instead
65  * of reading, the last reader closes.  The writer needs to be woken up so it
66  * can return 0. */
67 void apipe_close_reader(struct atomic_pipe *ap)
68 {
69         spin_lock(&ap->ap_lock);
70         ap->ap_nr_readers--;
71         __cv_broadcast(&ap->ap_writers);
72         spin_unlock(&ap->ap_lock);
73 }
74
75 void apipe_close_writer(struct atomic_pipe *ap)
76 {
77         spin_lock(&ap->ap_lock);
78         ap->ap_nr_writers--;
79         __apipe_wake_readers(ap);
80         spin_unlock(&ap->ap_lock);
81 }
82
83 /* read a pipe that is already locked. */
84 int apipe_read_locked(struct atomic_pipe *ap, void *buf, size_t nr_elem)
85 {
86         size_t rd_idx;
87         int nr_copied = 0;
88
89         for (int i = 0; i < nr_elem; i++) {
90                 /* readers that call read_locked directly might have failed to check for
91                  * emptiness, so we'll double check early. */
92                 if (__ring_empty(ap->ap_wr_off, ap->ap_rd_off))
93                         break;
94                 /* power of 2 elements in the ring buffer, index is the lower n bits */
95                 rd_idx = ap->ap_rd_off & (ap->ap_ring_sz - 1);
96                 memcpy(buf, ap->ap_buf + rd_idx * ap->ap_elem_sz, ap->ap_elem_sz);
97                 ap->ap_rd_off++;
98                 buf += ap->ap_elem_sz;
99                 nr_copied++;
100         }
101         /* We could have multiple writers blocked.  Just broadcast for them all.
102          * Alternatively, we could signal one, and then it's on the writers to
103          * signal further writers (see the note at the top of this file). */
104         __cv_broadcast(&ap->ap_writers);
105         return nr_copied;
106 }
107
108
109 int apipe_read(struct atomic_pipe *ap, void *buf, size_t nr_elem)
110 {
111         size_t rd_idx;
112         int nr_copied = 0;
113
114         spin_lock(&ap->ap_lock);
115         /* Need to wait til the priority reader is gone, and the ring isn't empty.
116          * If we do this as two steps, (either of priority check or empty check
117          * first), there's a chance the second one will fail, and when we sleep and
118          * wake up, the first condition could have changed.  (An alternative would
119          * be to block priority readers too, by promoting ourselves to a priority
120          * reader). */
121         while (ap->ap_has_priority_reader ||
122                __ring_empty(ap->ap_wr_off, ap->ap_rd_off)) {
123                 if (!ap->ap_nr_writers) {
124                         spin_unlock(&ap->ap_lock);
125                         return 0;
126                 }
127                 cv_wait(&ap->ap_general_readers);
128                 cpu_relax();
129         }
130         /* This read call wakes up writers */
131         nr_copied = apipe_read_locked(ap, buf, nr_elem);
132         /* If the writer didn't broadcast, we'd need to wake other readers (imagine
133          * a long queue of blocked readers, and a queue filled by one massive
134          * write).  (same with the error case). */
135         spin_unlock(&ap->ap_lock);
136         return nr_copied;
137 }
138
139 int apipe_write(struct atomic_pipe *ap, void *buf, size_t nr_elem)
140 {
141         size_t wr_idx;
142         int nr_copied = 0;
143
144         spin_lock(&ap->ap_lock);
145         /* not sure if we want to check for readers first or not */
146         while (__ring_full(ap->ap_ring_sz, ap->ap_wr_off, ap->ap_rd_off)) {
147                 if (!ap->ap_nr_readers) {
148                         spin_unlock(&ap->ap_lock);
149                         return 0;
150                 }
151                 cv_wait(&ap->ap_writers);
152                 cpu_relax();
153         }
154         for (int i = 0; i < nr_elem; i++) {
155                 /* power of 2 elements in the ring buffer, index is the lower n bits */
156                 wr_idx = ap->ap_wr_off & (ap->ap_ring_sz - 1);
157                 memcpy(ap->ap_buf + wr_idx * ap->ap_elem_sz, buf, ap->ap_elem_sz);
158                 ap->ap_wr_off++;
159                 buf += ap->ap_elem_sz;
160                 nr_copied++;
161                 if (__ring_full(ap->ap_ring_sz, ap->ap_wr_off, ap->ap_rd_off))
162                         break;
163         }
164         /* We only need to wake readers, since the reader that woke us used a
165          * broadcast.  o/w, we'd need to wake the next writer.  (same goes for the
166          * error case). */
167         __apipe_wake_readers(ap);
168         spin_unlock(&ap->ap_lock);
169         return nr_copied;
170 }
171
172 void *apipe_head(struct atomic_pipe *ap)
173 {
174         if (__ring_empty(ap->ap_wr_off, ap->ap_rd_off))
175                 return 0;
176         return ap->ap_buf + (ap->ap_rd_off & (ap->ap_ring_sz - 1)) * ap->ap_elem_sz;
177 }
178
179 /* 
180  * Read data from the pipe until a condition is satisfied.
181  * f is the function that determines the condition. f saves its
182  * state in arg. When f returns non-zero, this function exits,
183  * and returns the value to its caller. Note that f can return -1
184  * to indicate an error. But returning zero will keep you trapped in
185  * this function. The intent here is to ensure one-reader-at-a-time
186  * operation.
187  */
188 int apipe_read_cond(struct atomic_pipe *ap,
189                     int(*f)(struct atomic_pipe *pipe, void *arg), void *arg)
190 {
191         size_t rd_idx;
192         int ret;
193
194         spin_lock(&ap->ap_lock);
195         /* Can only have one priority reader at a time.  Wait our turn. */
196         while (ap->ap_has_priority_reader) {
197                 cv_wait(&ap->ap_general_readers);
198                 cpu_relax();
199         }
200         ap->ap_has_priority_reader = TRUE;
201         while (1) {
202                 /* Each time there is a need to check the pipe, call
203                  * f. f will maintain its state in arg. It is expected that f
204                  * will dequeue elements from the pipe as they are available.
205                  * N.B. this is being done for protocols like IPV4 that can
206                  * fragment an RPC request. For IPV6, it is likely that this
207                  * will end up looking like a blocking read. Thus was it ever
208                  * with legacy code. F is supposed to call apipe_read_locked().
209                  */
210                 ret = f(ap, arg);
211                 if (ret)
212                         break;
213                 /* if nr_writers goes to zero, that's bad.  return -1 because they're
214                  * going to have to clean up.  We should have been able to call f once
215                  * though, to pull out any remaining elements.  The main concern here is
216                  * sleeping on the cv when no one (no writers) will wake us. */
217                 if (!ap->ap_nr_writers) {
218                         ret = -1;
219                         goto out;
220                 }
221                 cv_wait(&ap->ap_priority_reader);
222                 cpu_relax();
223         }
224 out:
225         /* All out paths need to wake other readers.  When we were woken up, there
226          * was no broadcast sent to the other readers.  Plus, there may be other
227          * potential priority readers. */
228         ap->ap_has_priority_reader = FALSE;
229         __apipe_wake_readers(ap);
230         /* FYI, writers were woken up after an actual read.  If we had an error (ret
231          * == -1), there should be no writers. */
232         spin_unlock(&ap->ap_lock);
233         return ret;
234 }