src/pulsecore/flist.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2006 Lennart Poettering
   7
   8   PulseAudio is free software; you can redistribute it and/or modify
   9   it under the terms of the GNU Lesser General Public License as
  10   published by the Free Software Foundation; either version 2.1 of the
  11   License, or (at your option) any later version.
  12
  13   PulseAudio is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   Lesser General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public
  19   License along with PulseAudio; if not, write to the Free Software
  20   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  21   USA.
  22 ***/
  23
  24 #ifdef HAVE_CONFIG_H
  25 #include <config.h>
  26 #endif
  27
  28 #include <assert.h>
  29
  30 #include <pulsecore/atomic.h>
  31 #include <pulsecore/log.h>
  32 #include <pulsecore/thread.h>
  33 #include <pulse/xmalloc.h>
  34
  35 #include "flist.h"
  36
  37 /* Algorithm is not perfect, in a few corner cases it will fail to pop
  38  * from the flist although it isn't empty, and fail to push into the
  39  * flist, although it isn't full.
  40  *
  41  * We keep a fixed size array of entries, each item is either marked
  42  * UNUSED, USED or BUSY and contains a user data pointer. When pushing
  43  * into the queue we look for an UNUSED cell and mark it BUSY with a
  44  * CAS operation. If successful we use it and mark it USED, otherwise
  45  * we go on and look for the next UNUSED cell. The algorithm for
  46  * popping an item from the queue is practically inverse: look for a
  47  * USED cell and and mark it BUSY with a CAS operation, after reading
  48  * from it mark it UNUSED again.
  49  *
  50  * To accelerate finding of used/unused cells we maintain a read and a
  51  * write index which is used like a ring buffer. After each push we
  52  * increase the write index and after each pop we increase the read
  53  * index.
  54  *
  55  * The indexes are incremented atomically and are never truncated to
  56  * the buffer size. Instead we assume that the buffer size is a power
  57  * of two and that the truncation can thus be done by applying a
  58  * simple AND on read.
  59  *
  60  * To make sure that we do not look for empty cells indefinitely we
  61  * maintain a length value which stores the number of used cells. From
  62  * this value the number of unused cells is easily calculated. Please
  63  * note that the length value is not updated atomically with the read
  64  * and write index and might thus be a few cells off the real
  65  * value. To deal with this we always look for N_EXTRA_SCAN extra
  66  * cells when pushing/popping entries.
  67  *
  68  * It might make sense to replace this implementation with a link list
  69  * stack or queue, which however requires DCAS to be simple. Patches
  70  * welcome.
  71  *
  72  * Please note that this algorithm is home grown.*/
  73
  74 #define FLIST_SIZE 128
  75 #define N_EXTRA_SCAN 2
  76
  77 /* For debugging purposes we can define _Y to put and extra thread
  78  * yield between each operation. */
  79
  80 #ifdef PROFILE
  81 #define _Y pa_thread_yield()
  82 #else
  83 #define _Y do { } while(0)
  84 #endif
  85
  86 enum {
  87     STATE_UNUSED,
  88     STATE_USED,
  89     STATE_BUSY
  90 };
  91
  92 struct cell {
  93     pa_atomic_int_t state;
  94     void *data;
  95 };
  96
  97 struct pa_flist {
  98     struct cell *cells;
  99     unsigned size;
 100     pa_atomic_int_t length;
 101     pa_atomic_int_t read_idx;
 102     pa_atomic_int_t write_idx;
 103 };
 104
 105 static int is_power_of_two(unsigned size) {
 106     return !(size & (size - 1));
 107 }
 108
 109 pa_flist *pa_flist_new(unsigned size) {
 110     pa_flist *l;
 111
 112     if (!size)
 113         size = FLIST_SIZE;
 114
 115     assert(is_power_of_two(size));
 116
 117     l = pa_xnew(pa_flist, 1);
 118
 119     l->size = size;
 120     l->cells = pa_xnew0(struct cell, size);
 121
 122     pa_atomic_store(&l->read_idx, 0);
 123     pa_atomic_store(&l->write_idx, 0);
 124     pa_atomic_store(&l->length, 0);
 125
 126     return l;
 127 }
 128
 129 static int reduce(pa_flist *l, int value) {
 130     return value & (unsigned) (l->size - 1);
 131 }
 132
 133 void pa_flist_free(pa_flist *l, pa_free_cb_t free_cb) {
 134     assert(l);
 135
 136     if (free_cb) {
 137         int len, idx;
 138
 139         idx = reduce(l, pa_atomic_load(&l->read_idx));
 140         len = pa_atomic_load(&l->length);
 141
 142         for (; len > 0; len--) {
 143
 144             if (pa_atomic_load(&l->cells[idx].state) == STATE_USED)
 145                 free_cb(l->cells[idx].data);
 146
 147             idx = reduce(l, idx + 1);
 148         }
 149     }
 150
 151     pa_xfree(l->cells);
 152     pa_xfree(l);
 153 }
 154
 155 int pa_flist_push(pa_flist*l, void *p) {
 156     int idx, len, n;
 157
 158     assert(l);
 159     assert(p);
 160
 161     n = len = (int) l->size - pa_atomic_load(&l->length) + N_EXTRA_SCAN;
 162     _Y;
 163     idx = reduce(l, pa_atomic_load(&l->write_idx));
 164
 165     for (; n > 0 ; n--) {
 166         _Y;
 167
 168         if (pa_atomic_cmpxchg(&l->cells[idx].state, STATE_UNUSED, STATE_BUSY)) {
 169             _Y;
 170             pa_atomic_inc(&l->write_idx);
 171             _Y;
 172             l->cells[idx].data = p;
 173             _Y;
 174             pa_atomic_store(&l->cells[idx].state, STATE_USED);
 175             _Y;
 176             pa_atomic_inc(&l->length);
 177             return 0;
 178         }
 179
 180         _Y;
 181         idx = reduce(l, idx + 1);
 182     }
 183
 184 #ifdef PROFILE
 185     if (len > N_EXTRA_SCAN)
 186         pa_log("WARNING: Didn't  find free cell after %u iterations.", len);
 187 #endif
 188
 189     return -1;
 190 }
 191
 192 void* pa_flist_pop(pa_flist*l) {
 193     int idx, len, n;
 194
 195     assert(l);
 196
 197     n = len = pa_atomic_load(&l->length) + N_EXTRA_SCAN;
 198     _Y;
 199     idx = reduce(l, pa_atomic_load(&l->read_idx));
 200
 201     for (; n > 0 ; n--) {
 202         _Y;
 203
 204         if (pa_atomic_cmpxchg(&l->cells[idx].state, STATE_USED, STATE_BUSY)) {
 205             void *p;
 206             _Y;
 207             pa_atomic_inc(&l->read_idx);
 208             _Y;
 209             p = l->cells[idx].data;
 210             _Y;
 211             pa_atomic_store(&l->cells[idx].state, STATE_UNUSED);
 212             _Y;
 213
 214             pa_atomic_dec(&l->length);
 215             return p;
 216         }
 217
 218         _Y;
 219         idx = reduce(l, idx+1);
 220     }
 221
 222 #ifdef PROFILE
 223     if (len > N_EXTRA_SCAN)
 224         pa_log("WARNING: Didn't find used cell after %u iterations.", len);
 225 #endif
 226
 227     return NULL;
 228 }