src/pulsecore/asyncq.c

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