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

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,

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,

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) {

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");

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;

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

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))) {

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;

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(

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 =

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",

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__;

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__;

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));

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));
 

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);

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);

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);

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);

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\".");

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);

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;
 

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);
 

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 =

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;
 

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) {

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 */