Fixes apipe_read_cond()
[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                 /* power of 2 elements in the ring buffer, index is the lower n bits */
91                 rd_idx = ap->ap_rd_off & (ap->ap_ring_sz - 1);
92                 memcpy(buf, ap->ap_buf + rd_idx * ap->ap_elem_sz, ap->ap_elem_sz);
93                 ap->ap_rd_off++;
94                 buf += ap->ap_elem_sz;
95                 nr_copied++;
96                 if (__ring_empty(ap->ap_wr_off, ap->ap_rd_off))
97                         break;
98         }
99         /* We could have multiple writers blocked.  Just broadcast for them all.
100          * Alternatively, we could signal one, and then it's on the writers to
101          * signal further writers (see the note at the top of this file). */
102         __cv_broadcast(&ap->ap_writers);
103         return nr_copied;
104 }
105
106
107 int apipe_read(struct atomic_pipe *ap, void *buf, size_t nr_elem)
108 {
109         size_t rd_idx;
110         int nr_copied = 0;
111
112         spin_lock(&ap->ap_lock);
113         /* Need to wait til the priority reader is gone, and the ring isn't empty.
114          * If we do this as two steps, (either of priority check or empty check
115          * first), there's a chance the second one will fail, and when we sleep and
116          * wake up, the first condition could have changed.  (An alternative would
117          * be to block priority readers too, by promoting ourselves to a priority
118          * reader). */
119         while (ap->ap_has_priority_reader ||
120                __ring_empty(ap->ap_wr_off, ap->ap_rd_off)) {
121                 if (!ap->ap_nr_writers) {
122                         spin_unlock(&ap->ap_lock);
123                         return 0;
124                 }
125                 cv_wait(&ap->ap_general_readers);
126                 cpu_relax();
127         }
128         /* This read call wakes up writers */
129         nr_copied = apipe_read_locked(ap, buf, nr_elem);
130         /* If the writer didn't broadcast, we'd need to wake other readers (imagine
131          * a long queue of blocked readers, and a queue filled by one massive
132          * write).  (same with the error case). */
133         spin_unlock(&ap->ap_lock);
134         return nr_copied;
135 }
136
137 int apipe_write(struct atomic_pipe *ap, void *buf, size_t nr_elem)
138 {
139         size_t wr_idx;
140         int nr_copied = 0;
141
142         spin_lock(&ap->ap_lock);
143         /* not sure if we want to check for readers first or not */
144         while (__ring_full(ap->ap_ring_sz, ap->ap_wr_off, ap->ap_rd_off)) {
145                 if (!ap->ap_nr_readers) {
146                         spin_unlock(&ap->ap_lock);
147                         return 0;
148                 }
149                 cv_wait(&ap->ap_writers);
150                 cpu_relax();
151         }
152         for (int i = 0; i < nr_elem; i++) {
153                 /* power of 2 elements in the ring buffer, index is the lower n bits */
154                 wr_idx = ap->ap_wr_off & (ap->ap_ring_sz - 1);
155                 memcpy(ap->ap_buf + wr_idx * ap->ap_elem_sz, buf, ap->ap_elem_sz);
156                 ap->ap_wr_off++;
157                 buf += ap->ap_elem_sz;
158                 nr_copied++;
159                 if (__ring_full(ap->ap_ring_sz, ap->ap_wr_off, ap->ap_rd_off))
160                         break;
161         }
162         /* We only need to wake readers, since the reader that woke us used a
163          * broadcast.  o/w, we'd need to wake the next writer.  (same goes for the
164          * error case). */
165         __apipe_wake_readers(ap);
166         spin_unlock(&ap->ap_lock);
167         return nr_copied;
168 }
169
170 void *apipe_head(struct atomic_pipe *ap)
171 {
172         if (__ring_empty(ap->ap_wr_off, ap->ap_rd_off))
173                 return 0;
174         return ap->ap_buf + (ap->ap_rd_off & (ap->ap_ring_sz - 1)) * ap->ap_elem_sz;
175 }
176
177 /* 
178  * Read data from the pipe until a condition is satisfied.
179  * f is the function that determines the condition. f saves its
180  * state in arg. When f returns non-zero, this function exits,
181  * and returns the value to its caller. Note that f can return -1
182  * to indicate an error. But returning zero will keep you trapped in
183  * this function. The intent here is to ensure one-reader-at-a-time
184  * operation.
185  */
186 int apipe_read_cond(struct atomic_pipe *ap,
187                     int(*f)(struct atomic_pipe *pipe, void *arg), void *arg)
188 {
189         size_t rd_idx;
190         int ret;
191
192         spin_lock(&ap->ap_lock);
193         /* Can only have one priority reader at a time.  Wait our turn. */
194         while (ap->ap_has_priority_reader) {
195                 cv_wait(&ap->ap_general_readers);
196                 cpu_relax();
197         }
198         ap->ap_has_priority_reader = TRUE;
199         while (1) {
200                 /* Each time there is a need to check the pipe, call
201                  * f. f will maintain its state in arg. It is expected that f
202                  * will dequeue elements from the pipe as they are available.
203                  * N.B. this is being done for protocols like IPV4 that can
204                  * fragment an RPC request. For IPV6, it is likely that this
205                  * will end up looking like a blocking read. Thus was it ever
206                  * with legacy code. F is supposed to call apipe_read_locked().
207                  */
208                 ret = f(ap, arg);
209                 if (ret)
210                         break;
211                 /* if nr_writers goes to zero, that's bad.  return -1 because they're
212                  * going to have to clean up.  We should have been able to call f once
213                  * though, to pull out any remaining elements.  The main concern here is
214                  * sleeping on the cv when no one (no writers) will wake us. */
215                 if (!ap->ap_nr_writers) {
216                         ret = -1;
217                         goto out;
218                 }
219                 cv_wait(&ap->ap_priority_reader);
220                 cpu_relax();
221         }
222 out:
223         /* All out paths need to wake other readers.  When we were woken up, there
224          * was no broadcast sent to the other readers.  Plus, there may be other
225          * potential priority readers. */
226         ap->ap_has_priority_reader = FALSE;
227         __apipe_wake_readers(ap);
228         /* FYI, writers were woken up after an actual read.  If we had an error (ret
229          * == -1), there should be no writers. */
230         spin_unlock(&ap->ap_lock);
231         return ret;
232 }