don't assume that pa_asyncq_new() always succeeds

Bug 96741 shows a case where an assertion is hit, because
pa_asyncq_new() failed due to running out of file descriptors.
pa_asyncq_new() is used in only one place (not counting the call in
asyncq-test): pa_asyncmsgq_new(). Now pa_asyncmsgq_new() can fail too,
which requires error handling in many places. One of those places is
pa_thread_mq_init(), which can now fail too, and that needs additional
error handling in many more places. Luckily there weren't any places
where adding better error handling wouldn't have been easy, so there are
many changes in this patch, but they are not complicated.

BugLink: https://bugs.freedesktop.org/show_bug.cgi?id=96741

27 files changed, 205 insertions, 32 deletions

diff --git a/src/modules/alsa/alsa-sink.c b/src/modules/alsa/alsa-sink.c
index 63674e28f..886c735f9 100644
--- a/src/modules/alsa/alsa-sink.c
+++ b/src/modules/alsa/alsa-sink.c
@@ -2115,7 +2115,11 @@ pa_sink *pa_alsa_sink_new(pa_module *m, pa_modargs *ma, const char*driver, pa_ca
     u->first = true;
     u->rewind_safeguard = rewind_safeguard;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->smoother = pa_smoother_new(
             SMOOTHER_ADJUST_USEC,
diff --git a/src/modules/alsa/alsa-source.c b/src/modules/alsa/alsa-source.c
index 0820b48f1..b788df2b3 100644
--- a/src/modules/alsa/alsa-source.c
+++ b/src/modules/alsa/alsa-source.c
@@ -1823,7 +1823,11 @@ pa_source *pa_alsa_source_new(pa_module *m, pa_modargs *ma, const char*driver, p
     u->fixed_latency_range = fixed_latency_range;
     u->first = true;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->smoother = pa_smoother_new(
             SMOOTHER_ADJUST_USEC,
diff --git a/src/modules/bluetooth/module-bluez4-device.c b/src/modules/bluetooth/module-bluez4-device.c
index 7f9335bda..ac4ed63f0 100644
--- a/src/modules/bluetooth/module-bluez4-device.c
+++ b/src/modules/bluetooth/module-bluez4-device.c
@@ -1949,7 +1949,11 @@ static int start_thread(struct userdata *u) {
     pa_assert(!u->rtpoll_item);
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        return -1;
+    }
 
     if (USE_SCO_OVER_PCM(u)) {
         if (sco_over_pcm_state_update(u, false) < 0) {
diff --git a/src/modules/bluetooth/module-bluez5-device.c b/src/modules/bluetooth/module-bluez5-device.c
index 0914bfb58..065fcaa61 100644
--- a/src/modules/bluetooth/module-bluez5-device.c
+++ b/src/modules/bluetooth/module-bluez5-device.c
@@ -1525,7 +1525,11 @@ static int start_thread(struct userdata *u) {
     pa_assert(!u->rtpoll_item);
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        return -1;
+    }
 
     if (!(u->thread = pa_thread_new("bluetooth", thread_func, u))) {
         pa_log_error("Failed to create IO thread");
diff --git a/src/modules/echo-cancel/module-echo-cancel.c b/src/modules/echo-cancel/module-echo-cancel.c
index c873648dd..dfd05b647 100644
--- a/src/modules/echo-cancel/module-echo-cancel.c
+++ b/src/modules/echo-cancel/module-echo-cancel.c
@@ -1757,6 +1757,11 @@ int pa__init(pa_module*m) {
         goto fail;
 
     u->asyncmsgq = pa_asyncmsgq_new(0);
+    if (!u->asyncmsgq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     u->need_realign = true;
 
     source_output_ss = source_ss;
diff --git a/src/modules/jack/module-jack-sink.c b/src/modules/jack/module-jack-sink.c
index 38ba9ba76..4d314930b 100644
--- a/src/modules/jack/module-jack-sink.c
+++ b/src/modules/jack/module-jack-sink.c
@@ -320,10 +320,18 @@ int pa__init(pa_module*m) {
     u->module = m;
     u->saved_frame_time_valid = false;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     /* The queue linking the JACK thread and our RT thread */
     u->jack_msgq = pa_asyncmsgq_new(0);
+    if (!u->jack_msgq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
 
     /* The msgq from the JACK RT thread should have an even higher
      * priority than the normal message queues, to match the guarantee
diff --git a/src/modules/jack/module-jack-source.c b/src/modules/jack/module-jack-source.c
index 1130947d0..e45f3048f 100644
--- a/src/modules/jack/module-jack-source.c
+++ b/src/modules/jack/module-jack-source.c
@@ -274,9 +274,18 @@ int pa__init(pa_module*m) {
     u->module = m;
     u->saved_frame_time_valid = false;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->jack_msgq = pa_asyncmsgq_new(0);
+    if (!u->jack_msgq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     u->rtpoll_item = pa_rtpoll_item_new_asyncmsgq_read(u->rtpoll, PA_RTPOLL_EARLY-1, u->jack_msgq);
 
     if (!(u->client = jack_client_open(client_name, server_name ? JackServerName : JackNullOption, &status, server_name))) {
diff --git a/src/modules/macosx/module-coreaudio-device.c b/src/modules/macosx/module-coreaudio-device.c
index d91c6564c..502fc511d 100644
--- a/src/modules/macosx/module-coreaudio-device.c
+++ b/src/modules/macosx/module-coreaudio-device.c
@@ -825,8 +825,18 @@ int pa__init(pa_module *m) {
     pa_card_put(u->card);
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->async_msgq = pa_asyncmsgq_new(0);
+    if (!u->async_msgq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     pa_rtpoll_item_new_asyncmsgq_read(u->rtpoll, PA_RTPOLL_EARLY-1, u->async_msgq);
 
     PA_LLIST_HEAD_INIT(coreaudio_sink, u->sinks);
@@ -910,9 +920,11 @@ void pa__done(pa_module *m) {
         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
         pa_thread_free(u->thread);
         pa_thread_mq_done(&u->thread_mq);
-        pa_asyncmsgq_unref(u->async_msgq);
     }
 
+    if (u->async_msgq)
+        pa_asyncmsgq_unref(u->async_msgq);
+
     /* free sinks */
     for (ca_sink = u->sinks; ca_sink;) {
         coreaudio_sink *next = ca_sink->next;
diff --git a/src/modules/module-combine-sink.c b/src/modules/module-combine-sink.c
index b6322c6b7..250240a7a 100644
--- a/src/modules/module-combine-sink.c
+++ b/src/modules/module-combine-sink.c
@@ -1019,9 +1019,25 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
 
     o = pa_xnew0(struct output, 1);
     o->userdata = u;
+
     o->audio_inq = pa_asyncmsgq_new(0);
+    if (!o->audio_inq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     o->control_inq = pa_asyncmsgq_new(0);
+    if (!o->control_inq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     o->outq = pa_asyncmsgq_new(0);
+    if (!o->outq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
+
     o->sink = sink;
     o->memblockq = pa_memblockq_new(
             "module-combine-sink output memblockq",
@@ -1038,6 +1054,11 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
     update_description(u);
 
     return o;
+
+fail:
+    output_free(o);
+
+    return NULL;
 }
 
 /* Called from main context */
@@ -1280,7 +1301,12 @@ int pa__init(pa_module*m) {
     u->core = m->core;
     u->module = m;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->resample_method = resample_method;
     u->outputs = pa_idxset_new(NULL, NULL);
     u->thread_info.smoother = pa_smoother_new(
diff --git a/src/modules/module-esound-sink.c b/src/modules/module-esound-sink.c
index 76e380e54..2ce0c8549 100644
--- a/src/modules/module-esound-sink.c
+++ b/src/modules/module-esound-sink.c
@@ -564,7 +564,12 @@ int pa__init(pa_module*m) {
     u->offset = 0;
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->rtpoll_item = NULL;
 
     u->format =
diff --git a/src/modules/module-loopback.c b/src/modules/module-loopback.c
index 246d622d5..9410c98a0 100644
--- a/src/modules/module-loopback.c
+++ b/src/modules/module-loopback.c
@@ -943,6 +943,10 @@ int pa__init(pa_module *m) {
     pa_memblock_unref(silence.memblock);
 
     u->asyncmsgq = pa_asyncmsgq_new(0);
+    if (!u->asyncmsgq) {
+        pa_log("pa_asyncmsgq_new() failed.");
+        goto fail;
+    }
 
     if (!pa_proplist_contains(u->source_output->proplist, PA_PROP_MEDIA_NAME))
         pa_proplist_setf(u->source_output->proplist, PA_PROP_MEDIA_NAME, "Loopback to %s",
diff --git a/src/modules/module-null-sink.c b/src/modules/module-null-sink.c
index 5b98ba5cb..b8157e8ff 100644
--- a/src/modules/module-null-sink.c
+++ b/src/modules/module-null-sink.c
@@ -269,7 +269,11 @@ int pa__init(pa_module*m) {
     u->core = m->core;
     u->module = m;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     pa_sink_new_data_init(&data);
     data.driver = __FILE__;
diff --git a/src/modules/module-null-source.c b/src/modules/module-null-source.c
index 0c590bd29..a75a04f5a 100644
--- a/src/modules/module-null-source.c
+++ b/src/modules/module-null-source.c
@@ -200,7 +200,11 @@ int pa__init(pa_module*m) {
     u->core = m->core;
     u->module = m;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     pa_source_new_data_init(&data);
     data.driver = __FILE__;
diff --git a/src/modules/module-pipe-sink.c b/src/modules/module-pipe-sink.c
index 0eefdf170..da6502115 100644
--- a/src/modules/module-pipe-sink.c
+++ b/src/modules/module-pipe-sink.c
@@ -247,7 +247,12 @@ int pa__init(pa_module *m) {
     m->userdata = u;
     pa_memchunk_reset(&u->memchunk);
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->write_type = 0;
 
     u->filename = pa_runtime_path(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
diff --git a/src/modules/module-pipe-source.c b/src/modules/module-pipe-source.c
index 1218674f9..f39fc557a 100644
--- a/src/modules/module-pipe-source.c
+++ b/src/modules/module-pipe-source.c
@@ -234,7 +234,11 @@ int pa__init(pa_module *m) {
     u->module = m;
     pa_memchunk_reset(&u->memchunk);
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->filename = pa_runtime_path(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
 
diff --git a/src/modules/module-sine-source.c b/src/modules/module-sine-source.c
index 55d70c79a..cdeb2c03d 100644
--- a/src/modules/module-sine-source.c
+++ b/src/modules/module-sine-source.c
@@ -226,7 +226,11 @@ int pa__init(pa_module*m) {
     u->core = m->core;
     u->module = m;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->peek_index = 0;
     pa_memchunk_sine(&u->memchunk, m->core->mempool, ss.rate, frequency);
diff --git a/src/modules/module-solaris.c b/src/modules/module-solaris.c
index 2fa0bffa6..ccff69fc9 100644
--- a/src/modules/module-solaris.c
+++ b/src/modules/module-solaris.c
@@ -911,7 +911,11 @@ int pa__init(pa_module *m) {
     pa_memchunk_reset(&u->memchunk);
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     u->rtpoll_item = NULL;
     build_pollfd(u);
diff --git a/src/modules/module-tunnel-sink-new.c b/src/modules/module-tunnel-sink-new.c
index 7f835433e..92f99df79 100644
--- a/src/modules/module-tunnel-sink-new.c
+++ b/src/modules/module-tunnel-sink-new.c
@@ -497,7 +497,11 @@ int pa__init(pa_module *m) {
     u->remote_sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 
     u->thread_mq = pa_xnew0(pa_thread_mq, 1);
-    pa_thread_mq_init_thread_mainloop(u->thread_mq, m->core->mainloop, u->thread_mainloop_api);
+
+    if (pa_thread_mq_init_thread_mainloop(u->thread_mq, m->core->mainloop, u->thread_mainloop_api) < 0) {
+        pa_log("pa_thread_mq_init_thread_mainloop() failed.");
+        goto fail;
+    }
 
     /* Create sink */
     pa_sink_new_data_init(&sink_data);
diff --git a/src/modules/module-tunnel-source-new.c b/src/modules/module-tunnel-source-new.c
index 0f72dbf9f..e159c33f1 100644
--- a/src/modules/module-tunnel-source-new.c
+++ b/src/modules/module-tunnel-source-new.c
@@ -496,7 +496,11 @@ int pa__init(pa_module *m) {
     u->remote_source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 
     u->thread_mq = pa_xnew0(pa_thread_mq, 1);
-    pa_thread_mq_init_thread_mainloop(u->thread_mq, m->core->mainloop, u->thread_mainloop_api);
+
+    if (pa_thread_mq_init_thread_mainloop(u->thread_mq, m->core->mainloop, u->thread_mainloop_api) < 0) {
+        pa_log("pa_thread_mq_init_thread_mainloop() failed.");
+        goto fail;
+    }
 
     /* Create source */
     pa_source_new_data_init(&source_data);
diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c
index 5c8b84ae5..e08816bd2 100644
--- a/src/modules/module-tunnel.c
+++ b/src/modules/module-tunnel.c
@@ -1968,7 +1968,11 @@ int pa__init(pa_module*m) {
     u->counter = u->counter_delta = 0;
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     if (pa_modargs_get_value_boolean(ma, "auto", &automatic) < 0) {
         pa_log("Failed to parse argument \"auto\".");
diff --git a/src/modules/module-waveout.c b/src/modules/module-waveout.c
index ab3ea74cb..0b219f1fb 100644
--- a/src/modules/module-waveout.c
+++ b/src/modules/module-waveout.c
@@ -684,7 +684,11 @@ int pa__init(pa_module *m) {
         sink_get_volume_cb(u->sink);
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
 
     if (u->sink) {
         pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
diff --git a/src/modules/module-zeroconf-publish.c b/src/modules/module-zeroconf-publish.c
index 0d908409e..e9710292a 100644
--- a/src/modules/module-zeroconf-publish.c
+++ b/src/modules/module-zeroconf-publish.c
@@ -826,7 +826,11 @@ int pa__init(pa_module*m) {
     u->mainloop = pa_threaded_mainloop_new();
     u->api = pa_threaded_mainloop_get_api(u->mainloop);
 
-    pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll);
+    if (pa_thread_mq_init(&u->thread_mq, u->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->msg = pa_msgobject_new(avahi_msg);
     u->msg->parent.process_msg = avahi_process_msg;
 
diff --git a/src/modules/oss/module-oss.c b/src/modules/oss/module-oss.c
index 8537dd807..8a5a69279 100644
--- a/src/modules/oss/module-oss.c
+++ b/src/modules/oss/module-oss.c
@@ -1270,7 +1270,12 @@ int pa__init(pa_module*m) {
     u->orig_frag_size = orig_frag_size;
     u->use_mmap = use_mmap;
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->rtpoll_item = NULL;
     build_pollfd(u);
 
diff --git a/src/modules/raop/module-raop-sink.c b/src/modules/raop/module-raop-sink.c
index 603703873..7a97e8304 100644
--- a/src/modules/raop/module-raop-sink.c
+++ b/src/modules/raop/module-raop-sink.c
@@ -551,7 +551,12 @@ int pa__init(pa_module*m) {
     u->encoding_ratio = 1.0;
 
     u->rtpoll = pa_rtpoll_new();
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
+
+    if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+
     u->rtpoll_item = NULL;
 
     /*u->format =
diff --git a/src/pulsecore/asyncmsgq.c b/src/pulsecore/asyncmsgq.c
index dd33e30ac..47371ae19 100644
--- a/src/pulsecore/asyncmsgq.c
+++ b/src/pulsecore/asyncmsgq.c
@@ -58,12 +58,17 @@ struct pa_asyncmsgq {
 };
 
 pa_asyncmsgq *pa_asyncmsgq_new(unsigned size) {
+    pa_asyncq *asyncq;
     pa_asyncmsgq *a;
 
+    asyncq = pa_asyncq_new(size);
+    if (!asyncq)
+        return NULL;
+
     a = pa_xnew(pa_asyncmsgq, 1);
 
     PA_REFCNT_INIT(a);
-    pa_assert_se(a->asyncq = pa_asyncq_new(size));
+    a->asyncq = asyncq;
     pa_assert_se(a->mutex = pa_mutex_new(false, true));
     a->current = NULL;
 
diff --git a/src/pulsecore/thread-mq.c b/src/pulsecore/thread-mq.c
index f4489ac31..eaa866370 100644
--- a/src/pulsecore/thread-mq.c
+++ b/src/pulsecore/thread-mq.c
@@ -97,13 +97,20 @@ static void asyncmsgq_write_outq_cb(pa_mainloop_api *api, pa_io_event *e, int fd
     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) {
+int 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));
+    pa_zero(*q);
+
+    q->inq = pa_asyncmsgq_new(0);
+    if (!q->inq)
+        goto fail;
+
+    q->outq = pa_asyncmsgq_new(0);
+    if (!q->outq)
+        goto fail;
 
     q->main_mainloop = main_mainloop;
     q->thread_mainloop = thread_mainloop;
@@ -117,17 +124,31 @@ void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_ma
     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));
+
+    return 0;
+
+fail:
+    pa_thread_mq_done(q);
+
+    return -1;
 }
 
-void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll) {
+int pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll) {
     pa_assert(q);
     pa_assert(mainloop);
 
+    pa_zero(*q);
+
     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));
+    q->inq = pa_asyncmsgq_new(0);
+    if (!q->inq)
+        goto fail;
+
+    q->outq = pa_asyncmsgq_new(0);
+    if (!q->outq)
+        goto fail;
 
     pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
     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));
@@ -137,6 +158,13 @@ void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rt
 
     pa_rtpoll_item_new_asyncmsgq_read(rtpoll, PA_RTPOLL_EARLY, q->inq);
     pa_rtpoll_item_new_asyncmsgq_write(rtpoll, PA_RTPOLL_LATE, q->outq);
+
+    return 0;
+
+fail:
+    pa_thread_mq_done(q);
+
+    return -1;
 }
 
 void pa_thread_mq_done(pa_thread_mq *q) {
diff --git a/src/pulsecore/thread-mq.h b/src/pulsecore/thread-mq.h
index b3a4a6953..f6daa7fec 100644
--- a/src/pulsecore/thread-mq.h
+++ b/src/pulsecore/thread-mq.h
@@ -36,8 +36,8 @@ typedef struct pa_thread_mq {
     pa_io_event *read_thread_event, *write_thread_event;
 } pa_thread_mq;
 
-void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll);
-void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop);
+int pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll);
+int pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop);
 void pa_thread_mq_done(pa_thread_mq *q);
 
 /* Install the specified pa_thread_mq object for the current thread */