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