tunnel-sink-new: reinit module

When configured, reinitialize the module instead of exiting.  This
allows a restart/reconnect, but the module to appear to always be alive
when the user does: "pactl list modules".  (The sink will still not
exist until the tcp connection is established.)

Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>

src/modules/meson.build

@@ -55,7 +55,7 @@ all_modules = [
   [ 'module-switch-on-connect', 'module-switch-on-connect.c' ],
   [ 'module-switch-on-port-available', 'module-switch-on-port-available.c' ],
   [ 'module-tunnel-sink', 'module-tunnel.c', [], ['-DTUNNEL_SINK=1'], [x11_dep] ],
-  [ 'module-tunnel-sink-new', 'module-tunnel-sink-new.c' ],
+  [ 'module-tunnel-sink-new', ['module-tunnel-sink-new.c', 'restart-module.c'] ],
   [ 'module-tunnel-source', 'module-tunnel.c', [], [], [x11_dep] ],
   [ 'module-tunnel-source-new', 'module-tunnel-source-new.c' ],
   [ 'module-virtual-sink', 'module-virtual-sink.c' ],

src/modules/module-tunnel-sink-new.c

@@ -21,6 +21,8 @@
 #include <config.h>
 #endif
 
+#include "restart-module.h"
+
 #include <pulse/context.h>
 #include <pulse/timeval.h>
 #include <pulse/xmalloc.h>
@@ -50,6 +52,7 @@ PA_MODULE_USAGE(
         "sink=<name of the remote sink> "
         "sink_name=<name for the local sink> "
         "sink_properties=<properties for the local sink> "
+        "reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
         "format=<sample format> "
         "channels=<number of channels> "
         "rate=<sample rate> "
@@ -60,6 +63,8 @@ PA_MODULE_USAGE(
 #define MAX_LATENCY_USEC (200 * PA_USEC_PER_MSEC)
 #define TUNNEL_THREAD_FAILED_MAINLOOP 1
 
+static int do_init(pa_module *m);
+static void do_done(pa_module *m);
 static void stream_state_cb(pa_stream *stream, void *userdata);
 static void stream_changed_buffer_attr_cb(pa_stream *stream, void *userdata);
 static void stream_set_buffer_attr_cb(pa_stream *stream, int success, void *userdata);
@@ -75,6 +80,7 @@ PA_DEFINE_PRIVATE_CLASS(tunnel_msg, pa_msgobject);
 
 enum {
     TUNNEL_MESSAGE_CREATE_SINK_REQUEST,
+    TUNNEL_MESSAGE_MAYBE_RESTART,
 };
 
 enum {
@@ -108,6 +114,8 @@ struct userdata {
     pa_channel_map channel_map;
 
     tunnel_msg *msg;
+
+    pa_usec_t reconnect_interval_us;
 };
 
 static const char* const valid_modargs[] = {
@@ -120,7 +128,7 @@ static const char* const valid_modargs[] = {
     "rate",
     "channel_map",
     "cookie",
-   /* "reconnect", reconnect if server comes back again - unimplemented */
+    "reconnect_interval_ms",
     NULL,
 };
 
@@ -166,6 +174,7 @@ static pa_proplist* tunnel_new_proplist(struct userdata *u) {
 static void thread_func(void *userdata) {
     struct userdata *u = userdata;
     pa_proplist *proplist;
+
     pa_assert(u);
 
     pa_log_debug("Thread starting up");
@@ -192,7 +201,7 @@ static void thread_func(void *userdata) {
                            u->remote_server,
                            PA_CONTEXT_NOAUTOSPAWN,
                            NULL) < 0) {
-        pa_log("Failed to connect libpulse context");
+        pa_log("Failed to connect libpulse context: %s", pa_strerror(pa_context_errno(u->context)));
         goto fail;
     }
 
@@ -244,7 +253,10 @@ static void thread_func(void *userdata) {
         }
     }
 fail:
-    pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+    /* send a message to the ctl thread to ask it to either terminate us, or
+     * restart us, but either way this thread will exit, so then wait for the
+     * shutdown message */
+    pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->msg), TUNNEL_MESSAGE_MAYBE_RESTART, u, 0, NULL, NULL);
     pa_asyncmsgq_wait_for(u->thread_mq->inq, PA_MESSAGE_SHUTDOWN);
 
 finish:
@@ -323,6 +335,17 @@ static void stream_overflow_callback(pa_stream *stream, void *userdata) {
     pa_log_info("Server signalled buffer overrun.");
 }
 
+/* Do a reinit of the module.  Note that u will be freed as a result of this
+ * call. */
+static void maybe_restart(struct userdata *u) {
+    if (u->reconnect_interval_us > 0) {
+        pa_restart_module_reinit(u->module, do_init, do_done, u->reconnect_interval_us);
+    } else {
+        /* exit the module */
+        pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+    }
+}
+
 static void on_sink_created(struct userdata *u) {
     pa_proplist *proplist;
     pa_buffer_attr bufferattr;
@@ -596,16 +619,20 @@ static int tunnel_process_msg(pa_msgobject *o, int code, void *data, int64_t off
         case TUNNEL_MESSAGE_CREATE_SINK_REQUEST:
             create_sink(u);
             break;
+        case TUNNEL_MESSAGE_MAYBE_RESTART:
+            maybe_restart(u);
+            break;
     }
 
     return 0;
 }
 
-int pa__init(pa_module *m) {
+static int do_init(pa_module *m) {
     struct userdata *u = NULL;
     pa_modargs *ma = NULL;
     const char *remote_server = NULL;
     char *default_sink_name = NULL;
+    uint32_t reconnect_interval_ms = 0;
 
     pa_assert(m);
 
@@ -676,6 +703,9 @@ int pa__init(pa_module *m) {
         goto fail;
     }
 
+    pa_modargs_get_value_u32(ma, "reconnect_interval_ms", &reconnect_interval_ms);
+    u->reconnect_interval_us = reconnect_interval_ms * PA_USEC_PER_MSEC;
+
     if (!(u->thread = pa_thread_new("tunnel-sink", thread_func, u))) {
         pa_log("Failed to create thread.");
         goto fail;
@@ -693,13 +723,11 @@ fail:
     if (default_sink_name)
         pa_xfree(default_sink_name);
 
-    pa__done(m);
-
     return -1;
 }
 
-void pa__done(pa_module *m) {
+static void do_done(pa_module *m) {
-    struct userdata *u;
+    struct userdata *u = NULL;
 
     pa_assert(m);
 
@@ -748,4 +776,25 @@ void pa__done(pa_module *m) {
     pa_xfree(u->msg);
 
     pa_xfree(u);
+
+    m->userdata = NULL;
+}
+
+int pa__init(pa_module *m) {
+    int ret;
+
+    pa_assert(m);
+
+    ret = do_init(m);
+
+    if (ret < 0)
+        pa__done(m);
+
+    return ret;
+}
+
+void pa__done(pa_module *m) {
+    pa_assert(m);
+
+    do_done(m);
 }

src/modules/restart-module.c

@@ -0,0 +1,88 @@
+/***
+    This file is part of PulseAudio.
+
+    Copyright 2022 Craig Howard
+
+    PulseAudio is free software; you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published
+    by the Free Software Foundation; either version 2.1 of the License,
+    or (at your option) any later version.
+
+    PulseAudio is distributed in the hope that it will be useful, but
+    WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+    General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "restart-module.h"
+
+#include <pulse/timeval.h>
+#include <pulse/xmalloc.h>
+#include <pulse/mainloop.h>
+
+#include <pulsecore/core.h>
+#include <pulsecore/thread-mq.h>
+
+struct reinit_data {
+    init_cb do_init;
+    done_cb do_done;
+
+    pa_usec_t restart_usec;
+    pa_module *module;
+};
+
+static void call_init(pa_mainloop_api *mainloop, pa_time_event *e, const struct timeval *tv, void *userdata) {
+    struct reinit_data *rd = userdata;
+    int ret;
+
+    /* now that restart_usec has elapsed, we call do_init to restart the module */
+    ret = rd->do_init(rd->module);
+
+    /* if the init failed, we got here because the caller wanted to restart, so
+     * setup another restart */
+    if (ret < 0)
+        pa_restart_module_reinit(rd->module, rd->do_init, rd->do_done, rd->restart_usec);
+
+    pa_xfree(rd);
+}
+
+static void do_reinit(pa_mainloop_api *mainloop, void *userdata) {
+    struct reinit_data *rd = userdata;
+    struct timeval tv;
+
+    pa_assert_ctl_context();
+
+    /* call do_done on the module, which will effectively tear it down; all
+     * that remains is the pa_module */
+    rd->do_done(rd->module);
+
+    /* after restart_usec, call do_init to restart the module */
+    pa_gettimeofday(&tv);
+    pa_timeval_add(&tv, rd->restart_usec);
+    mainloop->time_new(mainloop, &tv, call_init, rd);
+}
+
+void pa_restart_module_reinit(pa_module *m, init_cb do_init, done_cb do_done, pa_usec_t restart_usec) {
+    struct reinit_data *rd;
+
+    pa_assert_ctl_context();
+
+    pa_log_info("Starting reinit for %s", m->name);
+
+    rd = pa_xnew0(struct reinit_data, 1);
+    rd->do_init = do_init;
+    rd->do_done = do_done;
+    rd->restart_usec = restart_usec;
+    rd->module = m;
+
+    /* defer actually doing a reinit, so that we can safely exit whatever call
+     * chain we're in before we effectively reinit the module */
+    pa_mainloop_api_once(m->core->mainloop, do_reinit, rd);
+}

src/modules/restart-module.h

@@ -0,0 +1,41 @@
+/***
+    This file is part of PulseAudio.
+
+    Copyright 2022 Craig Howard
+
+    PulseAudio is free software; you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published
+    by the Free Software Foundation; either version 2.1 of the License,
+    or (at your option) any later version.
+
+    PulseAudio is distributed in the hope that it will be useful, but
+    WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+    General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef RESTART_MODULE_H
+#define RESTART_MODULE_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <pulse/timeval.h>
+
+#include <pulsecore/core.h>
+#include <pulsecore/thread-mq.h>
+
+typedef int (*init_cb)(pa_module *m);
+typedef void (*done_cb)(pa_module *m);
+
+void pa_restart_module_reinit(pa_module *m, init_cb do_init, done_cb do_done, pa_usec_t restart_usec);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif