code.delx.au - pulseaudio/blob - src/pulsecore/thread-mq.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/thread.h>
  30 #include <pulsecore/semaphore.h>
  31 #include <pulsecore/macro.h>
  32
  33 #include <pulse/mainloop-api.h>
  34
  35 #include "thread-mq.h"
  36
  37 PA_STATIC_TLS_DECLARE_NO_FREE(thread_mq);
  38
  39 static void asyncmsgq_read_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
  40     pa_thread_mq *q = userdata;
  41     pa_asyncmsgq *aq;
  42
  43     pa_assert(events == PA_IO_EVENT_INPUT);
  44
  45     if (pa_asyncmsgq_read_fd(q->outq) == fd)
  46         pa_asyncmsgq_ref(aq = q->outq);
  47     else if (pa_asyncmsgq_read_fd(q->inq) == fd)
  48         pa_asyncmsgq_ref(aq = q->inq);
  49     else
  50         pa_assert_not_reached();
  51
  52     pa_asyncmsgq_read_after_poll(aq);
  53
  54     for (;;) {
  55         pa_msgobject *object;
  56         int code;
  57         void *data;
  58         int64_t offset;
  59         pa_memchunk chunk;
  60
  61         /* Check whether there is a message for us to process */
  62         while (pa_asyncmsgq_get(aq, &object, &code, &data, &offset, &chunk, 0) >= 0) {
  63             int ret;
  64
  65             if (!object && code == PA_MESSAGE_SHUTDOWN) {
  66                 pa_asyncmsgq_done(aq, 0);
  67                 api->quit(api, 0);
  68                 break;
  69             }
  70
  71             ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
  72             pa_asyncmsgq_done(aq, ret);
  73         }
  74
  75         if (pa_asyncmsgq_read_before_poll(aq) == 0)
  76             break;
  77     }
  78
  79     pa_asyncmsgq_unref(aq);
  80 }
  81
  82 static void asyncmsgq_write_inq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
  83     pa_thread_mq *q = userdata;
  84
  85     pa_assert(pa_asyncmsgq_write_fd(q->inq) == fd);
  86     pa_assert(events == PA_IO_EVENT_INPUT);
  87
  88     pa_asyncmsgq_write_after_poll(q->inq);
  89     pa_asyncmsgq_write_before_poll(q->inq);
  90 }
  91
  92 static void asyncmsgq_write_outq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
  93     pa_thread_mq *q = userdata;
  94
  95     pa_assert(pa_asyncmsgq_write_fd(q->outq) == fd);
  96     pa_assert(events == PA_IO_EVENT_INPUT);
  97
  98     pa_asyncmsgq_write_after_poll(q->outq);
  99     pa_asyncmsgq_write_before_poll(q->outq);
 100 }
 101
 102 void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop) {
 103     pa_assert(q);
 104     pa_assert(main_mainloop);
 105     pa_assert(thread_mainloop);
 106
 107     pa_assert_se(q->inq = pa_asyncmsgq_new(0));
 108     pa_assert_se(q->outq = pa_asyncmsgq_new(0));
 109
 110     q->main_mainloop = main_mainloop;
 111     q->thread_mainloop = thread_mainloop;
 112
 113     pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
 114     pa_asyncmsgq_write_before_poll(q->outq);
 115     pa_assert_se(q->read_main_event = main_mainloop->io_new(main_mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
 116     pa_assert_se(q->write_thread_event = main_mainloop->io_new(thread_mainloop, pa_asyncmsgq_write_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_write_outq_cb, q));
 117
 118     pa_asyncmsgq_read_before_poll(q->inq);
 119     pa_asyncmsgq_write_before_poll(q->inq);
 120     pa_assert_se(q->read_thread_event = thread_mainloop->io_new(thread_mainloop, pa_asyncmsgq_read_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
 121     pa_assert_se(q->write_main_event = main_mainloop->io_new(main_mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_inq_cb, q));
 122 }
 123
 124 void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll) {
 125     pa_assert(q);
 126     pa_assert(mainloop);
 127
 128     q->main_mainloop = mainloop;
 129     q->thread_mainloop = NULL;
 130
 131     pa_assert_se(q->inq = pa_asyncmsgq_new(0));
 132     pa_assert_se(q->outq = pa_asyncmsgq_new(0));
 133
 134     pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
 135     pa_assert_se(q->read_main_event = mainloop->io_new(mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
 136
 137     pa_asyncmsgq_write_before_poll(q->inq);
 138     pa_assert_se(q->write_main_event = mainloop->io_new(mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_inq_cb, q));
 139
 140     pa_rtpoll_item_new_asyncmsgq_read(rtpoll, PA_RTPOLL_EARLY, q->inq);
 141     pa_rtpoll_item_new_asyncmsgq_write(rtpoll, PA_RTPOLL_LATE, q->outq);
 142 }
 143
 144 void pa_thread_mq_done(pa_thread_mq *q) {
 145     pa_assert(q);
 146
 147     /* Since we are called from main context we can be sure that the
 148      * inq is empty. However, the outq might still contain messages
 149      * for the main loop, which we need to dispatch (e.g. release
 150      * msgs, other stuff). Hence do so if we aren't currently
 151      * dispatching anyway. */
 152
 153     if (!pa_asyncmsgq_dispatching(q->outq))
 154         pa_asyncmsgq_flush(q->outq, true);
 155
 156     q->main_mainloop->io_free(q->read_main_event);
 157     q->main_mainloop->io_free(q->write_main_event);
 158     q->read_main_event = q->write_main_event = NULL;
 159
 160     if (q->thread_mainloop) {
 161         q->thread_mainloop->io_free(q->read_thread_event);
 162         q->thread_mainloop->io_free(q->write_thread_event);
 163         q->read_thread_event = q->write_thread_event = NULL;
 164     }
 165
 166     pa_asyncmsgq_unref(q->inq);
 167     pa_asyncmsgq_unref(q->outq);
 168     q->inq = q->outq = NULL;
 169
 170     q->main_mainloop = NULL;
 171     q->thread_mainloop = NULL;
 172 }
 173
 174 void pa_thread_mq_install(pa_thread_mq *q) {
 175     pa_assert(q);
 176
 177     pa_assert(!(PA_STATIC_TLS_GET(thread_mq)));
 178     PA_STATIC_TLS_SET(thread_mq, q);
 179 }
 180
 181 pa_thread_mq *pa_thread_mq_get(void) {
 182     return PA_STATIC_TLS_GET(thread_mq);
 183 }