src/pulsecore/asyncmsgq.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, 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/macro.h>
  30 #include <pulsecore/log.h>
  31 #include <pulsecore/semaphore.h>
  32 #include <pulsecore/macro.h>
  33 #include <pulsecore/mutex.h>
  34 #include <pulsecore/flist.h>
  35
  36 #include "asyncmsgq.h"
  37
  38 PA_STATIC_FLIST_DECLARE(asyncmsgq, 0, pa_xfree);
  39 PA_STATIC_FLIST_DECLARE(semaphores, 0, (void(*)(void*)) pa_semaphore_free);
  40
  41 struct asyncmsgq_item {
  42     int code;
  43     pa_msgobject *object;
  44     void *userdata;
  45     pa_free_cb_t free_cb;
  46     int64_t offset;
  47     pa_memchunk memchunk;
  48     pa_semaphore *semaphore;
  49     int ret;
  50 };
  51
  52 struct pa_asyncmsgq {
  53     PA_REFCNT_DECLARE;
  54     pa_asyncq *asyncq;
  55     pa_mutex *mutex; /* only for the writer side */
  56
  57     struct asyncmsgq_item *current;
  58 };
  59
  60 pa_asyncmsgq *pa_asyncmsgq_new(unsigned size) {
  61     pa_asyncq *asyncq;
  62     pa_asyncmsgq *a;
  63
  64     asyncq = pa_asyncq_new(size);
  65     if (!asyncq)
  66         return NULL;
  67
  68     a = pa_xnew(pa_asyncmsgq, 1);
  69
  70     PA_REFCNT_INIT(a);
  71     a->asyncq = asyncq;
  72     pa_assert_se(a->mutex = pa_mutex_new(false, true));
  73     a->current = NULL;
  74
  75     return a;
  76 }
  77
  78 static void asyncmsgq_free(pa_asyncmsgq *a) {
  79     struct asyncmsgq_item *i;
  80     pa_assert(a);
  81
  82     while ((i = pa_asyncq_pop(a->asyncq, false))) {
  83
  84         pa_assert(!i->semaphore);
  85
  86         if (i->object)
  87             pa_msgobject_unref(i->object);
  88
  89         if (i->memchunk.memblock)
  90             pa_memblock_unref(i->memchunk.memblock);
  91
  92         if (i->free_cb)
  93             i->free_cb(i->userdata);
  94
  95         if (pa_flist_push(PA_STATIC_FLIST_GET(asyncmsgq), i) < 0)
  96             pa_xfree(i);
  97     }
  98
  99     pa_asyncq_free(a->asyncq, NULL);
 100     pa_mutex_free(a->mutex);
 101     pa_xfree(a);
 102 }
 103
 104 pa_asyncmsgq* pa_asyncmsgq_ref(pa_asyncmsgq *q) {
 105     pa_assert(PA_REFCNT_VALUE(q) > 0);
 106
 107     PA_REFCNT_INC(q);
 108     return q;
 109 }
 110
 111 void pa_asyncmsgq_unref(pa_asyncmsgq* q) {
 112     pa_assert(PA_REFCNT_VALUE(q) > 0);
 113
 114     if (PA_REFCNT_DEC(q) <= 0)
 115         asyncmsgq_free(q);
 116 }
 117
 118 void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *chunk, pa_free_cb_t free_cb) {
 119     struct asyncmsgq_item *i;
 120     pa_assert(PA_REFCNT_VALUE(a) > 0);
 121
 122     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(asyncmsgq))))
 123         i = pa_xnew(struct asyncmsgq_item, 1);
 124
 125     i->code = code;
 126     i->object = object ? pa_msgobject_ref(object) : NULL;
 127     i->userdata = (void*) userdata;
 128     i->free_cb = free_cb;
 129     i->offset = offset;
 130     if (chunk) {
 131         pa_assert(chunk->memblock);
 132         i->memchunk = *chunk;
 133         pa_memblock_ref(i->memchunk.memblock);
 134     } else
 135         pa_memchunk_reset(&i->memchunk);
 136     i->semaphore = NULL;
 137
 138     /* This mutex makes the queue multiple-writer safe. This lock is only used on the writing side */
 139     pa_mutex_lock(a->mutex);
 140     pa_asyncq_post(a->asyncq, i);
 141     pa_mutex_unlock(a->mutex);
 142 }
 143
 144 int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *chunk) {
 145     struct asyncmsgq_item i;
 146     pa_assert(PA_REFCNT_VALUE(a) > 0);
 147
 148     i.code = code;
 149     i.object = object;
 150     i.userdata = (void*) userdata;
 151     i.free_cb = NULL;
 152     i.ret = -1;
 153     i.offset = offset;
 154     if (chunk) {
 155         pa_assert(chunk->memblock);
 156         i.memchunk = *chunk;
 157     } else
 158         pa_memchunk_reset(&i.memchunk);
 159
 160     if (!(i.semaphore = pa_flist_pop(PA_STATIC_FLIST_GET(semaphores))))
 161         i.semaphore = pa_semaphore_new(0);
 162
 163     /* This mutex makes the queue multiple-writer safe. This lock is only used on the writing side */
 164     pa_mutex_lock(a->mutex);
 165     pa_assert_se(pa_asyncq_push(a->asyncq, &i, true) == 0);
 166     pa_mutex_unlock(a->mutex);
 167
 168     pa_semaphore_wait(i.semaphore);
 169
 170     if (pa_flist_push(PA_STATIC_FLIST_GET(semaphores), i.semaphore) < 0)
 171         pa_semaphore_free(i.semaphore);
 172
 173     return i.ret;
 174 }
 175
 176 int pa_asyncmsgq_get(pa_asyncmsgq *a, pa_msgobject **object, int *code, void **userdata, int64_t *offset, pa_memchunk *chunk, bool wait_op) {
 177     pa_assert(PA_REFCNT_VALUE(a) > 0);
 178     pa_assert(!a->current);
 179
 180     if (!(a->current = pa_asyncq_pop(a->asyncq, wait_op))) {
 181 /*         pa_log("failure"); */
 182         return -1;
 183     }
 184
 185 /*     pa_log("success"); */
 186
 187     if (code)
 188         *code = a->current->code;
 189     if (userdata)
 190         *userdata = a->current->userdata;
 191     if (offset)
 192         *offset = a->current->offset;
 193     if (object) {
 194         if ((*object = a->current->object))
 195             pa_msgobject_assert_ref(*object);
 196     }
 197     if (chunk)
 198         *chunk = a->current->memchunk;
 199
 200 /*     pa_log_debug("Get q=%p object=%p (%s) code=%i data=%p chunk.length=%lu", */
 201 /*                  (void*) a, */
 202 /*                  (void*) a->current->object, */
 203 /*                  a->current->object ? a->current->object->parent.type_name : NULL, */
 204 /*                  a->current->code, */
 205 /*                  (void*) a->current->userdata, */
 206 /*                  (unsigned long) a->current->memchunk.length); */
 207
 208     return 0;
 209 }
 210
 211 void pa_asyncmsgq_done(pa_asyncmsgq *a, int ret) {
 212     pa_assert(PA_REFCNT_VALUE(a) > 0);
 213     pa_assert(a);
 214     pa_assert(a->current);
 215
 216     if (a->current->semaphore) {
 217         a->current->ret = ret;
 218         pa_semaphore_post(a->current->semaphore);
 219     } else {
 220
 221         if (a->current->free_cb)
 222             a->current->free_cb(a->current->userdata);
 223
 224         if (a->current->object)
 225             pa_msgobject_unref(a->current->object);
 226
 227         if (a->current->memchunk.memblock)
 228             pa_memblock_unref(a->current->memchunk.memblock);
 229
 230         if (pa_flist_push(PA_STATIC_FLIST_GET(asyncmsgq), a->current) < 0)
 231             pa_xfree(a->current);
 232     }
 233
 234     a->current = NULL;
 235 }
 236
 237 int pa_asyncmsgq_wait_for(pa_asyncmsgq *a, int code) {
 238     int c;
 239     pa_assert(PA_REFCNT_VALUE(a) > 0);
 240
 241     pa_asyncmsgq_ref(a);
 242
 243     do {
 244         pa_msgobject *o;
 245         void *data;
 246         int64_t offset;
 247         pa_memchunk chunk;
 248         int ret;
 249
 250         if (pa_asyncmsgq_get(a, &o, &c, &data, &offset, &chunk, true) < 0)
 251             return -1;
 252
 253         ret = pa_asyncmsgq_dispatch(o, c, data, offset, &chunk);
 254         pa_asyncmsgq_done(a, ret);
 255
 256     } while (c != code);
 257
 258     pa_asyncmsgq_unref(a);
 259
 260     return 0;
 261 }
 262
 263 int pa_asyncmsgq_process_one(pa_asyncmsgq *a) {
 264     pa_msgobject *object;
 265     int code;
 266     void *data;
 267     pa_memchunk chunk;
 268     int64_t offset;
 269     int ret;
 270
 271     pa_assert(PA_REFCNT_VALUE(a) > 0);
 272
 273     if (pa_asyncmsgq_get(a, &object, &code, &data, &offset, &chunk, false) < 0)
 274         return 0;
 275
 276     pa_asyncmsgq_ref(a);
 277     ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 278     pa_asyncmsgq_done(a, ret);
 279     pa_asyncmsgq_unref(a);
 280
 281     return 1;
 282 }
 283
 284 int pa_asyncmsgq_read_fd(pa_asyncmsgq *a) {
 285     pa_assert(PA_REFCNT_VALUE(a) > 0);
 286
 287     return pa_asyncq_read_fd(a->asyncq);
 288 }
 289
 290 int pa_asyncmsgq_read_before_poll(pa_asyncmsgq *a) {
 291     pa_assert(PA_REFCNT_VALUE(a) > 0);
 292
 293     return pa_asyncq_read_before_poll(a->asyncq);
 294 }
 295
 296 void pa_asyncmsgq_read_after_poll(pa_asyncmsgq *a) {
 297     pa_assert(PA_REFCNT_VALUE(a) > 0);
 298
 299     pa_asyncq_read_after_poll(a->asyncq);
 300 }
 301
 302 int pa_asyncmsgq_write_fd(pa_asyncmsgq *a) {
 303     pa_assert(PA_REFCNT_VALUE(a) > 0);
 304
 305     return pa_asyncq_write_fd(a->asyncq);
 306 }
 307
 308 void pa_asyncmsgq_write_before_poll(pa_asyncmsgq *a) {
 309     pa_assert(PA_REFCNT_VALUE(a) > 0);
 310
 311     pa_asyncq_write_before_poll(a->asyncq);
 312 }
 313
 314 void pa_asyncmsgq_write_after_poll(pa_asyncmsgq *a) {
 315     pa_assert(PA_REFCNT_VALUE(a) > 0);
 316
 317     pa_asyncq_write_after_poll(a->asyncq);
 318 }
 319
 320 int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *memchunk) {
 321
 322     if (object)
 323         return object->process_msg(object, code, userdata, offset, pa_memchunk_isset(memchunk) ? memchunk : NULL);
 324
 325     return 0;
 326 }
 327
 328 void pa_asyncmsgq_flush(pa_asyncmsgq *a, bool run) {
 329     pa_assert(PA_REFCNT_VALUE(a) > 0);
 330
 331     for (;;) {
 332         pa_msgobject *object;
 333         int code;
 334         void *data;
 335         int64_t offset;
 336         pa_memchunk chunk;
 337         int ret;
 338
 339         if (pa_asyncmsgq_get(a, &object, &code, &data, &offset, &chunk, false) < 0)
 340             return;
 341
 342         if (!run) {
 343             pa_asyncmsgq_done(a, -1);
 344             continue;
 345         }
 346
 347         pa_asyncmsgq_ref(a);
 348         ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 349         pa_asyncmsgq_done(a, ret);
 350         pa_asyncmsgq_unref(a);
 351     }
 352 }
 353
 354 bool pa_asyncmsgq_dispatching(pa_asyncmsgq *a) {
 355     pa_assert(PA_REFCNT_VALUE(a) > 0);
 356
 357     return !!a->current;
 358 }