Branch data Line data Source code
1 : : /***
2 : : This file is part of PulseAudio.
3 : :
4 : : Copyright 2006-2008 Lennart Poettering
5 : :
6 : : PulseAudio is free software; you can redistribute it and/or modify
7 : : it under the terms of the GNU Lesser General Public License as
8 : : published by the Free Software Foundation; either version 2.1 of the
9 : : License, or (at your option) any later version.
10 : :
11 : : PulseAudio is distributed in the hope that it will be useful, but
12 : : WITHOUT ANY WARRANTY; without even the implied warranty of
13 : : MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 : : Lesser General Public License for more details.
15 : :
16 : : You should have received a copy of the GNU Lesser General Public
17 : : License along with PulseAudio; if not, write to the Free Software
18 : : Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
19 : : USA.
20 : : ***/
21 : :
22 : : #ifdef HAVE_CONFIG_H
23 : : #include <config.h>
24 : : #endif
25 : :
26 : : #include <unistd.h>
27 : : #include <errno.h>
28 : :
29 : : #include <pulse/xmalloc.h>
30 : :
31 : : #include <pulsecore/atomic.h>
32 : : #include <pulsecore/log.h>
33 : : #include <pulsecore/thread.h>
34 : : #include <pulsecore/macro.h>
35 : : #include <pulsecore/core-util.h>
36 : : #include <pulsecore/llist.h>
37 : : #include <pulsecore/flist.h>
38 : : #include <pulsecore/fdsem.h>
39 : :
40 : : #include "asyncq.h"
41 : :
42 : : #define ASYNCQ_SIZE 256
43 : :
44 : : /* For debugging purposes we can define _Y to put an extra thread
45 : : * yield between each operation. */
46 : :
47 : : /* #define PROFILE */
48 : :
49 : : #ifdef PROFILE
50 : : #define _Y pa_thread_yield()
51 : : #else
52 : : #define _Y do { } while(0)
53 : : #endif
54 : :
55 : : struct localq {
56 : : void *data;
57 : : PA_LLIST_FIELDS(struct localq);
58 : : };
59 : :
60 : : struct pa_asyncq {
61 : : unsigned size;
62 : : unsigned read_idx;
63 : : unsigned write_idx;
64 : : pa_fdsem *read_fdsem, *write_fdsem;
65 : :
66 : : PA_LLIST_HEAD(struct localq, localq);
67 : : struct localq *last_localq;
68 : : pa_bool_t waiting_for_post;
69 : : };
70 : :
71 [ - + ][ # # ]: 21 : PA_STATIC_FLIST_DECLARE(localq, 0, pa_xfree);
72 : :
73 : : #define PA_ASYNCQ_CELLS(x) ((pa_atomic_ptr_t*) ((uint8_t*) (x) + PA_ALIGN(sizeof(struct pa_asyncq))))
74 : :
75 : 0 : static unsigned reduce(pa_asyncq *l, unsigned value) {
76 : 0 : return value & (unsigned) (l->size - 1);
77 : : }
78 : :
79 : 2 : pa_asyncq *pa_asyncq_new(unsigned size) {
80 : : pa_asyncq *l;
81 : :
82 [ + - ]: 2 : if (!size)
83 : 2 : size = ASYNCQ_SIZE;
84 : :
85 [ - + ]: 2 : pa_assert(pa_is_power_of_two(size));
86 : :
87 : 2 : l = pa_xmalloc0(PA_ALIGN(sizeof(pa_asyncq)) + (sizeof(pa_atomic_ptr_t) * size));
88 : :
89 : 2 : l->size = size;
90 : :
91 : 2 : PA_LLIST_HEAD_INIT(struct localq, l->localq);
92 : 2 : l->last_localq = NULL;
93 : 2 : l->waiting_for_post = FALSE;
94 : :
95 [ - + ]: 2 : if (!(l->read_fdsem = pa_fdsem_new())) {
96 : 0 : pa_xfree(l);
97 : 0 : return NULL;
98 : : }
99 : :
100 [ - + ]: 2 : if (!(l->write_fdsem = pa_fdsem_new())) {
101 : 0 : pa_fdsem_free(l->read_fdsem);
102 : 0 : pa_xfree(l);
103 : 2 : return NULL;
104 : : }
105 : :
106 : : return l;
107 : : }
108 : :
109 : 2 : void pa_asyncq_free(pa_asyncq *l, pa_free_cb_t free_cb) {
110 : : struct localq *q;
111 [ - + ]: 2 : pa_assert(l);
112 : :
113 [ - + ]: 2 : if (free_cb) {
114 : : void *p;
115 : :
116 [ # # ]: 2 : while ((p = pa_asyncq_pop(l, 0)))
117 : 0 : free_cb(p);
118 : : }
119 : :
120 [ - + ]: 2 : while ((q = l->localq)) {
121 [ # # ]: 0 : if (free_cb)
122 : 0 : free_cb(q->data);
123 : :
124 [ # # ][ # # ]: 0 : PA_LLIST_REMOVE(struct localq, l->localq, q);
[ # # ][ # # ]
125 : :
126 [ # # ]: 0 : if (pa_flist_push(PA_STATIC_FLIST_GET(localq), q) < 0)
127 : 2 : pa_xfree(q);
128 : : }
129 : :
130 : 2 : pa_fdsem_free(l->read_fdsem);
131 : 2 : pa_fdsem_free(l->write_fdsem);
132 : 2 : pa_xfree(l);
133 : 2 : }
134 : :
135 : 1005 : static int push(pa_asyncq*l, void *p, pa_bool_t wait_op) {
136 : : unsigned idx;
137 : : pa_atomic_ptr_t *cells;
138 : :
139 [ - + ]: 1005 : pa_assert(l);
140 [ - + ]: 1005 : pa_assert(p);
141 : :
142 : 1005 : cells = PA_ASYNCQ_CELLS(l);
143 : :
144 : : _Y;
145 : 2010 : idx = reduce(l, l->write_idx);
146 : :
147 [ + + ]: 1005 : if (!pa_atomic_ptr_cmpxchg(&cells[idx], NULL, p)) {
148 : :
149 [ + - ]: 1 : if (!wait_op)
150 : : return -1;
151 : :
152 : : /* pa_log("sleeping on push"); */
153 : :
154 : : do {
155 : 1 : pa_fdsem_wait(l->read_fdsem);
156 [ - + ]: 1 : } while (!pa_atomic_ptr_cmpxchg(&cells[idx], NULL, p));
157 : : }
158 : :
159 : : _Y;
160 : 1005 : l->write_idx++;
161 : :
162 : 1005 : pa_fdsem_post(l->write_fdsem);
163 : :
164 : 1005 : return 0;
165 : : }
166 : :
167 : 1008 : static pa_bool_t flush_postq(pa_asyncq *l, pa_bool_t wait_op) {
168 : : struct localq *q;
169 : :
170 [ + - ]: 1008 : pa_assert(l);
171 : :
172 [ - + ]: 1008 : while ((q = l->last_localq)) {
173 : :
174 [ # # ]: 0 : if (push(l, q->data, wait_op) < 0)
175 : : return FALSE;
176 : :
177 : 0 : l->last_localq = q->prev;
178 : :
179 [ # # ][ # # ]: 0 : PA_LLIST_REMOVE(struct localq, l->localq, q);
[ # # ][ # # ]
180 : :
181 [ # # ]: 0 : if (pa_flist_push(PA_STATIC_FLIST_GET(localq), q) < 0)
182 : 1008 : pa_xfree(q);
183 : : }
184 : :
185 : : return TRUE;
186 : : }
187 : :
188 : 1005 : int pa_asyncq_push(pa_asyncq*l, void *p, pa_bool_t wait_op) {
189 [ - + ]: 1005 : pa_assert(l);
190 : :
191 [ + - ]: 1005 : if (!flush_postq(l, wait_op))
192 : : return -1;
193 : :
194 : 1005 : return push(l, p, wait_op);
195 : : }
196 : :
197 : 3 : void pa_asyncq_post(pa_asyncq*l, void *p) {
198 : : struct localq *q;
199 : :
200 [ - + ]: 3 : pa_assert(l);
201 [ - + ]: 3 : pa_assert(p);
202 : :
203 [ + - ]: 3 : if (flush_postq(l, FALSE))
204 [ - + ]: 3 : if (pa_asyncq_push(l, p, FALSE) >= 0)
205 : : return;
206 : :
207 : : /* OK, we couldn't push anything in the queue. So let's queue it
208 : : * locally and push it later */
209 : :
210 [ # # ]: 0 : if (pa_log_ratelimit(PA_LOG_WARN))
211 : 0 : pa_log_warn("q overrun, queuing locally");
212 : :
213 [ # # ]: 0 : if (!(q = pa_flist_pop(PA_STATIC_FLIST_GET(localq))))
214 : 0 : q = pa_xnew(struct localq, 1);
215 : :
216 : 0 : q->data = p;
217 [ # # ][ # # ]: 0 : PA_LLIST_PREPEND(struct localq, l->localq, q);
218 : :
219 [ # # ]: 0 : if (!l->last_localq)
220 : 3 : l->last_localq = q;
221 : :
222 : : return;
223 : : }
224 : :
225 : 1006 : void* pa_asyncq_pop(pa_asyncq*l, pa_bool_t wait_op) {
226 : : unsigned idx;
227 : : void *ret;
228 : : pa_atomic_ptr_t *cells;
229 : :
230 [ - + ]: 1006 : pa_assert(l);
231 : :
232 : 1006 : cells = PA_ASYNCQ_CELLS(l);
233 : :
234 : : _Y;
235 : 2012 : idx = reduce(l, l->read_idx);
236 : :
237 [ + + ]: 1006 : if (!(ret = pa_atomic_ptr_load(&cells[idx]))) {
238 : :
239 [ + + ]: 4 : if (!wait_op)
240 : : return NULL;
241 : :
242 : : /* pa_log("sleeping on pop"); */
243 : :
244 : : do {
245 : 3 : pa_fdsem_wait(l->write_fdsem);
246 [ - + ]: 3 : } while (!(ret = pa_atomic_ptr_load(&cells[idx])));
247 : : }
248 : :
249 [ - + ]: 1005 : pa_assert(ret);
250 : :
251 : : /* Guaranteed to succeed if we only have a single reader */
252 [ - + ]: 1005 : pa_assert_se(pa_atomic_ptr_cmpxchg(&cells[idx], ret, NULL));
253 : :
254 : : _Y;
255 : 1005 : l->read_idx++;
256 : :
257 : 1005 : pa_fdsem_post(l->read_fdsem);
258 : :
259 : 1006 : return ret;
260 : : }
261 : :
262 : 0 : int pa_asyncq_read_fd(pa_asyncq *q) {
263 [ # # ]: 0 : pa_assert(q);
264 : :
265 : 0 : return pa_fdsem_get(q->write_fdsem);
266 : : }
267 : :
268 : 0 : int pa_asyncq_read_before_poll(pa_asyncq *l) {
269 : : unsigned idx;
270 : : pa_atomic_ptr_t *cells;
271 : :
272 [ # # ]: 0 : pa_assert(l);
273 : :
274 : 0 : cells = PA_ASYNCQ_CELLS(l);
275 : :
276 : : _Y;
277 : 0 : idx = reduce(l, l->read_idx);
278 : :
279 : : for (;;) {
280 [ # # ]: 0 : if (pa_atomic_ptr_load(&cells[idx]))
281 : : return -1;
282 : :
283 [ # # ]: 0 : if (pa_fdsem_before_poll(l->write_fdsem) >= 0)
284 : : return 0;
285 : : }
286 : : }
287 : :
288 : 0 : void pa_asyncq_read_after_poll(pa_asyncq *l) {
289 [ # # ]: 0 : pa_assert(l);
290 : :
291 : 0 : pa_fdsem_after_poll(l->write_fdsem);
292 : 0 : }
293 : :
294 : 0 : int pa_asyncq_write_fd(pa_asyncq *q) {
295 [ # # ]: 0 : pa_assert(q);
296 : :
297 : 0 : return pa_fdsem_get(q->read_fdsem);
298 : : }
299 : :
300 : 0 : void pa_asyncq_write_before_poll(pa_asyncq *l) {
301 [ # # ]: 0 : pa_assert(l);
302 : :
303 : : for (;;) {
304 : :
305 [ # # ]: 0 : if (flush_postq(l, FALSE))
306 : : break;
307 : :
308 [ # # ]: 0 : if (pa_fdsem_before_poll(l->read_fdsem) >= 0) {
309 : 0 : l->waiting_for_post = TRUE;
310 : 0 : break;
311 : : }
312 : : }
313 : 0 : }
314 : :
315 : 0 : void pa_asyncq_write_after_poll(pa_asyncq *l) {
316 [ # # ]: 0 : pa_assert(l);
317 : :
318 [ # # ]: 0 : if (l->waiting_for_post) {
319 : 0 : pa_fdsem_after_poll(l->read_fdsem);
320 : 0 : l->waiting_for_post = FALSE;
321 : : }
322 : 0 : }
|
