code.delx.au - pulseaudio/blob - src/pulsecore/asyncmsgq.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 <unistd.h>
  29 #include <errno.h>
  30
  31 #include <pulsecore/atomic.h>
  32 #include <pulsecore/log.h>
  33 #include <pulsecore/thread.h>
  34 #include <pulsecore/semaphore.h>
  35 #include <pulsecore/macro.h>
  36 #include <pulsecore/core-util.h>
  37 #include <pulsecore/flist.h>
  38 #include <pulse/xmalloc.h>
  39
  40 #include "asyncmsgq.h"
  41
  42 PA_STATIC_FLIST_DECLARE(asyncmsgq, 0, pa_xfree);
  43
  44 struct asyncmsgq_item {
  45     int code;
  46     pa_msgobject *object;
  47     void *userdata;
  48     pa_free_cb_t free_cb;
  49     pa_memchunk memchunk;
  50     pa_semaphore *semaphore;
  51     int ret;
  52 };
  53
  54 struct pa_asyncmsgq {
  55     pa_asyncq *asyncq;
  56     pa_mutex *mutex; /* only for the writer side */
  57
  58     struct asyncmsgq_item *current;
  59 };
  60
  61 pa_asyncmsgq *pa_asyncmsgq_new(unsigned size) {
  62     pa_asyncmsgq *a;
  63
  64     a = pa_xnew(pa_asyncmsgq, 1);
  65
  66     pa_assert_se(a->asyncq = pa_asyncq_new(size));
  67     pa_assert_se(a->mutex = pa_mutex_new(0));
  68     a->current = NULL;
  69
  70     return a;
  71 }
  72
  73 void pa_asyncmsgq_free(pa_asyncmsgq *a) {
  74     struct asyncmsgq_item *i;
  75     pa_assert(a);
  76
  77     while ((i = pa_asyncq_pop(a->asyncq, 0))) {
  78
  79         pa_assert(!i->semaphore);
  80
  81         if (i->object)
  82             pa_msgobject_unref(i->object);
  83
  84         if (i->memchunk.memblock)
  85             pa_memblock_unref(i->memchunk.memblock);
  86
  87         if (i->free_cb)
  88             i->free_cb(i->userdata);
  89
  90         if (pa_flist_push(PA_STATIC_FLIST_GET(asyncmsgq), i) < 0)
  91             pa_xfree(i);
  92     }
  93
  94     pa_asyncq_free(a->asyncq, NULL);
  95     pa_mutex_free(a->mutex);
  96     pa_xfree(a);
  97 }
  98
  99 void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *chunk, pa_free_cb_t free_cb) {
 100     struct asyncmsgq_item *i;
 101     pa_assert(a);
 102
 103     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(asyncmsgq))))
 104         i = pa_xnew(struct asyncmsgq_item, 1);
 105
 106     i->code = code;
 107     i->object = object ? pa_msgobject_ref(object) : NULL;
 108     i->userdata = (void*) userdata;
 109     i->free_cb = free_cb;
 110     if (chunk) {
 111         pa_assert(chunk->memblock);
 112         i->memchunk = *chunk;
 113         pa_memblock_ref(i->memchunk.memblock);
 114     } else
 115         pa_memchunk_reset(&i->memchunk);
 116     i->semaphore = NULL;
 117
 118     /* Thus mutex makes the queue multiple-writer safe. This lock is only used on the writing side */
 119     pa_mutex_lock(a->mutex);
 120     pa_assert_se(pa_asyncq_push(a->asyncq, i, 1) == 0);
 121     pa_mutex_unlock(a->mutex);
 122 }
 123
 124 int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *chunk) {
 125     struct asyncmsgq_item i;
 126     pa_assert(a);
 127
 128     i.code = code;
 129     i.object = object;
 130     i.userdata = (void*) userdata;
 131     i.free_cb = NULL;
 132     i.ret = -1;
 133     if (chunk) {
 134         pa_assert(chunk->memblock);
 135         i.memchunk = *chunk;
 136     } else
 137         pa_memchunk_reset(&i.memchunk);
 138     pa_assert_se(i.semaphore = pa_semaphore_new(0));
 139
 140     /* Thus mutex makes the queue multiple-writer safe. This lock is only used on the writing side */
 141     pa_mutex_lock(a->mutex);
 142     pa_assert_se(pa_asyncq_push(a->asyncq, &i, 1) == 0);
 143     pa_mutex_unlock(a->mutex);
 144
 145     pa_semaphore_wait(i.semaphore);
 146     pa_semaphore_free(i.semaphore);
 147
 148     return i.ret;
 149 }
 150
 151 int pa_asyncmsgq_get(pa_asyncmsgq *a, pa_msgobject **object, int *code, void **userdata, pa_memchunk *chunk, int wait) {
 152     pa_assert(a);
 153     pa_assert(code);
 154     pa_assert(!a->current);
 155
 156     if (!(a->current = pa_asyncq_pop(a->asyncq, wait))) {
 157 /*         pa_log("failure"); */
 158         return -1;
 159     }
 160
 161 /*     pa_log("success"); */
 162
 163     *code = a->current->code;
 164     if (userdata)
 165         *userdata = a->current->userdata;
 166     if (object) {
 167         if ((*object = a->current->object))
 168             pa_msgobject_assert_ref(*object);
 169     }
 170     if (chunk)
 171         *chunk = a->current->memchunk;
 172
 173     pa_log_debug("Get q=%p object=%p (%s) code=%i data=%p chunk.length=%u", (void*) a, (void*) a->current->object, a->current->object ? a->current->object->parent.type_name : NULL, a->current->code, (void*) a->current->userdata, a->current->memchunk.length);
 174
 175     return 0;
 176 }
 177
 178 void pa_asyncmsgq_done(pa_asyncmsgq *a, int ret) {
 179     pa_assert(a);
 180     pa_assert(a->current);
 181
 182     if (a->current->semaphore) {
 183         a->current->ret = ret;
 184         pa_semaphore_post(a->current->semaphore);
 185     } else {
 186
 187         if (a->current->free_cb)
 188             a->current->free_cb(a->current->userdata);
 189
 190         if (a->current->object)
 191             pa_msgobject_unref(a->current->object);
 192
 193         if (a->current->memchunk.memblock)
 194             pa_memblock_unref(a->current->memchunk.memblock);
 195
 196         if (pa_flist_push(PA_STATIC_FLIST_GET(asyncmsgq), a->current) < 0)
 197             pa_xfree(a->current);
 198     }
 199
 200     a->current = NULL;
 201 }
 202
 203 int pa_asyncmsgq_wait_for(pa_asyncmsgq *a, int code) {
 204     int c;
 205     pa_assert(a);
 206
 207     do {
 208         pa_msgobject *o;
 209         void *data;
 210         pa_memchunk chunk;
 211         int ret;
 212
 213         if (pa_asyncmsgq_get(a, &o, &c, &data, &chunk, 1) < 0)
 214             return -1;
 215
 216         ret = pa_asyncmsgq_dispatch(o, c, data, &chunk);
 217         pa_asyncmsgq_done(a, ret);
 218
 219     } while (c != code);
 220
 221     return 0;
 222 }
 223
 224 int pa_asyncmsgq_get_fd(pa_asyncmsgq *a) {
 225     pa_assert(a);
 226
 227     return pa_asyncq_get_fd(a->asyncq);
 228 }
 229
 230 int pa_asyncmsgq_before_poll(pa_asyncmsgq *a) {
 231     pa_assert(a);
 232
 233     return pa_asyncq_before_poll(a->asyncq);
 234 }
 235
 236 void pa_asyncmsgq_after_poll(pa_asyncmsgq *a) {
 237     pa_assert(a);
 238
 239     pa_asyncq_after_poll(a->asyncq);
 240 }
 241
 242 int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, pa_memchunk *memchunk) {
 243
 244     if (object)
 245         return object->process_msg(object, code, userdata, memchunk);
 246
 247     return 0;
 248 }