code.delx.au - pulseaudio/blob - src/pulsecore/shmasyncq.c

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2006 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 <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 <pulse/xmalloc.h>
  35
  36 #include "fdsem.h"
  37
  38 /* For debugging purposes we can define _Y to put and extra thread
  39  * yield between each operation. */
  40
  41 /* #define PROFILE */
  42
  43 #ifdef PROFILE
  44 #define _Y pa_thread_yield()
  45 #else
  46 #define _Y do { } while(0)
  47 #endif
  48
  49 struct pa_shmasyncq {
  50     pa_fdsem *read_fdsem, *write_fdsem;
  51     pa_shmasyncq_data *data;
  52 };
  53
  54 static int is_power_of_two(unsigned size) {
  55     return !(size & (size - 1));
  56 }
  57
  58 static int reduce(pa_shmasyncq *l, int value) {
  59     return value & (unsigned) (l->n_elements - 1);
  60 }
  61
  62 static pa_atomic_t* get_cell(pa_shmasyncq *l, unsigned i) {
  63     pa_assert(i < l->data->n_elements);
  64
  65     return (pa_atomic_t*) ((uint8*t) l->data + PA_ALIGN(sizeof(pa_shmasyncq_data)) + i * (PA_ALIGN(sizeof(pa_atomic_t)) + PA_ALIGN(element_size)))
  66 }
  67
  68 static void *get_cell_data(pa_atomic_t *a) {
  69     return (uint8_t*) a + PA_ALIGN(sizeof(atomic_t));
  70 }
  71
  72 pa_shmasyncq *pa_shmasyncq_new(unsigned n_elements, size_t element_size, void *data, int fd[2]) {
  73     pa_shmasyncq *l;
  74
  75     pa_assert(n_elements > 0);
  76     pa_assert(is_power_of_two(n_elements));
  77     pa_assert(element_size > 0);
  78     pa_assert(data);
  79     pa_assert(fd);
  80
  81     l = pa_xnew(pa_shmasyncq, 1);
  82
  83     l->data = data;
  84     memset(data, 0, PA_SHMASYNCQ_SIZE(n_elements, element_size));
  85
  86     l->data->n_elements = n_elements;
  87     l->data->element_size = element_size;
  88
  89     if (!(l->read_fdsem = pa_fdsem_new_shm(&d->read_fdsem_data, &fd[0]))) {
  90         pa_xfree(l);
  91         return NULL;
  92     }
  93
  94     if (!(l->write_fdsem = pa_fdsem_new(&d->write_fdsem_data, &fd[1]))) {
  95         pa_fdsem_free(l->read_fdsem);
  96         pa_xfree(l);
  97         return NULL;
  98     }
  99
 100     return l;
 101 }
 102
 103 void pa_shmasyncq_free(pa_shmasyncq *l, pa_free_cb_t free_cb) {
 104     pa_assert(l);
 105
 106     if (free_cb) {
 107         void *p;
 108
 109         while ((p = pa_shmasyncq_pop(l, 0)))
 110             free_cb(p);
 111     }
 112
 113     pa_fdsem_free(l->read_fdsem);
 114     pa_fdsem_free(l->write_fdsem);
 115     pa_xfree(l);
 116 }
 117
 118 int pa_shmasyncq_push(pa_shmasyncq*l, void *p, int wait) {
 119     int idx;
 120     pa_atomic_ptr_t *cells;
 121
 122     pa_assert(l);
 123     pa_assert(p);
 124
 125     cells = PA_SHMASYNCQ_CELLS(l);
 126
 127     _Y;
 128     idx = reduce(l, l->write_idx);
 129
 130     if (!pa_atomic_ptr_cmpxchg(&cells[idx], NULL, p)) {
 131
 132         if (!wait)
 133             return -1;
 134
 135 /*         pa_log("sleeping on push"); */
 136
 137         do {
 138             pa_fdsem_wait(l->read_fdsem);
 139         } while (!pa_atomic_ptr_cmpxchg(&cells[idx], NULL, p));
 140     }
 141
 142     _Y;
 143     l->write_idx++;
 144
 145     pa_fdsem_post(l->write_fdsem);
 146
 147     return 0;
 148 }
 149
 150 void* pa_shmasyncq_pop(pa_shmasyncq*l, int wait) {
 151     int idx;
 152     void *ret;
 153     pa_atomic_ptr_t *cells;
 154
 155     pa_assert(l);
 156
 157     cells = PA_SHMASYNCQ_CELLS(l);
 158
 159     _Y;
 160     idx = reduce(l, l->read_idx);
 161
 162     if (!(ret = pa_atomic_ptr_load(&cells[idx]))) {
 163
 164         if (!wait)
 165             return NULL;
 166
 167 /*         pa_log("sleeping on pop"); */
 168
 169         do {
 170             pa_fdsem_wait(l->write_fdsem);
 171         } while (!(ret = pa_atomic_ptr_load(&cells[idx])));
 172     }
 173
 174     pa_assert(ret);
 175
 176     /* Guaranteed to succeed if we only have a single reader */
 177     pa_assert_se(pa_atomic_ptr_cmpxchg(&cells[idx], ret, NULL));
 178
 179     _Y;
 180     l->read_idx++;
 181
 182     pa_fdsem_post(l->read_fdsem);
 183
 184     return ret;
 185 }
 186
 187 int pa_shmasyncq_get_fd(pa_shmasyncq *q) {
 188     pa_assert(q);
 189
 190     return pa_fdsem_get(q->write_fdsem);
 191 }
 192
 193 int pa_shmasyncq_before_poll(pa_shmasyncq *l) {
 194     int idx;
 195     pa_atomic_ptr_t *cells;
 196
 197     pa_assert(l);
 198
 199     cells = PA_SHMASYNCQ_CELLS(l);
 200
 201     _Y;
 202     idx = reduce(l, l->read_idx);
 203
 204     for (;;) {
 205         if (pa_atomic_ptr_load(&cells[idx]))
 206             return -1;
 207
 208         if (pa_fdsem_before_poll(l->write_fdsem) >= 0)
 209             return 0;
 210     }
 211
 212     return 0;
 213 }
 214
 215 void pa_shmasyncq_after_poll(pa_shmasyncq *l) {
 216     pa_assert(l);
 217
 218     pa_fdsem_after_poll(l->write_fdsem);
 219 }