remap: Change remapping function argument type from void to int16_t / float as approp...

[pulseaudio] / src / pulsecore / thread-mq.c

diff --git a/src/pulsecore/thread-mq.c b/src/pulsecore/thread-mq.c
index 73997a74367401479b4cb9489a9132a695ac8efc..d34b22b30fe14849d8c978e84d5f1bdb9290e065 100644 (file)
--- a/src/pulsecore/thread-mq.c
+++ b/src/pulsecore/thread-mq.c
@@ -26,30 +26,30 @@
 #include <unistd.h>
 #include <errno.h>
 
-#include <pulse/xmalloc.h>
-
-#include <pulsecore/atomic.h>
-#include <pulsecore/once.h>
-#include <pulsecore/log.h>
 #include <pulsecore/thread.h>
 #include <pulsecore/semaphore.h>
 #include <pulsecore/macro.h>
-#include <pulsecore/core-util.h>
-#include <pulsecore/flist.h>
+
+#include <pulse/mainloop-api.h>
 
 #include "thread-mq.h"
 
 PA_STATIC_TLS_DECLARE_NO_FREE(thread_mq);
 
-static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io_event_flags_t events, void *userdata) {
+static void asyncmsgq_read_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
     pa_thread_mq *q = userdata;
     pa_asyncmsgq *aq;
 
-    pa_assert(pa_asyncmsgq_read_fd(q->outq) == fd);
     pa_assert(events == PA_IO_EVENT_INPUT);
 
-    pa_asyncmsgq_ref(aq = q->outq);
-    pa_asyncmsgq_write_after_poll(aq);
+    if (pa_asyncmsgq_read_fd(q->outq) == fd)
+        pa_asyncmsgq_ref(aq = q->outq);
+    else if (pa_asyncmsgq_read_fd(q->inq) == fd)
+        pa_asyncmsgq_ref(aq = q->inq);
+    else
+        pa_assert_not_reached();
+
+    pa_asyncmsgq_read_after_poll(aq);
 
     for (;;) {
         pa_msgobject *object;
@@ -62,6 +62,12 @@ static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io
         while (pa_asyncmsgq_get(aq, &object, &code, &data, &offset, &chunk, 0) >= 0) {
             int ret;
 
+            if (!object && code == PA_MESSAGE_SHUTDOWN) {
+                pa_asyncmsgq_done(aq, 0);
+                api->quit(api, 0);
+                break;
+            }
+
             ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
             pa_asyncmsgq_done(aq, ret);
         }
@@ -73,7 +79,7 @@ static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io
     pa_asyncmsgq_unref(aq);
 }
 
-static void asyncmsgq_write_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io_event_flags_t events, void *userdata) {
+static void asyncmsgq_write_inq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
     pa_thread_mq *q = userdata;
 
     pa_assert(pa_asyncmsgq_write_fd(q->inq) == fd);
@@ -83,19 +89,53 @@ static void asyncmsgq_write_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_i
     pa_asyncmsgq_write_before_poll(q->inq);
 }
 
+static void asyncmsgq_write_outq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
+    pa_thread_mq *q = userdata;
+
+    pa_assert(pa_asyncmsgq_write_fd(q->outq) == fd);
+    pa_assert(events == PA_IO_EVENT_INPUT);
+
+    pa_asyncmsgq_write_after_poll(q->outq);
+    pa_asyncmsgq_write_before_poll(q->outq);
+}
+
+void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop) {
+    pa_assert(q);
+    pa_assert(main_mainloop);
+    pa_assert(thread_mainloop);
+
+    pa_assert_se(q->inq = pa_asyncmsgq_new(0));
+    pa_assert_se(q->outq = pa_asyncmsgq_new(0));
+
+    q->main_mainloop = main_mainloop;
+    q->thread_mainloop = thread_mainloop;
+
+    pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
+    pa_asyncmsgq_write_before_poll(q->outq);
+    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));
+    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));
+
+    pa_asyncmsgq_read_before_poll(q->inq);
+    pa_asyncmsgq_write_before_poll(q->inq);
+    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));
+    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));
+}
+
 void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll) {
     pa_assert(q);
     pa_assert(mainloop);
 
-    q->mainloop = mainloop;
+    q->main_mainloop = mainloop;
+    q->thread_mainloop = NULL;
+
     pa_assert_se(q->inq = pa_asyncmsgq_new(0));
     pa_assert_se(q->outq = pa_asyncmsgq_new(0));
 
     pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
-    pa_assert_se(q->read_event = mainloop->io_new(mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
+    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));
 
     pa_asyncmsgq_write_before_poll(q->inq);
-    pa_assert_se(q->write_event = mainloop->io_new(mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_cb, q));
+    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));
 
     pa_rtpoll_item_new_asyncmsgq_read(rtpoll, PA_RTPOLL_EARLY, q->inq);
     pa_rtpoll_item_new_asyncmsgq_write(rtpoll, PA_RTPOLL_LATE, q->outq);
@@ -111,17 +151,24 @@ void pa_thread_mq_done(pa_thread_mq *q) {
      * dispatching anyway. */
 
     if (!pa_asyncmsgq_dispatching(q->outq))
-        pa_asyncmsgq_flush(q->outq, TRUE);
+        pa_asyncmsgq_flush(q->outq, true);
+
+    q->main_mainloop->io_free(q->read_main_event);
+    q->main_mainloop->io_free(q->write_main_event);
+    q->read_main_event = q->write_main_event = NULL;
 
-    q->mainloop->io_free(q->read_event);
-    q->mainloop->io_free(q->write_event);
-    q->read_event = q->write_event = NULL;
+    if (q->thread_mainloop) {
+        q->thread_mainloop->io_free(q->read_thread_event);
+        q->thread_mainloop->io_free(q->write_thread_event);
+        q->read_thread_event = q->write_thread_event = NULL;
+    }
 
     pa_asyncmsgq_unref(q->inq);
     pa_asyncmsgq_unref(q->outq);
     q->inq = q->outq = NULL;
 
-    q->mainloop = NULL;
+    q->main_mainloop = NULL;
+    q->thread_mainloop = NULL;
 }
 
 void pa_thread_mq_install(pa_thread_mq *q) {