From 34d00afc7467f80ef9be3c8591b49020ec7d6ee1 Mon Sep 17 00:00:00 2001 From: Craig Howard Date: Wed, 23 Mar 2022 17:25:57 -0700 Subject: tunnel-sink-new: create sink *after* connection The io thread, after connection, sends a message asking for a sink to be created. After the ctl thread is done with creation, it sends a message back to the io thread so it can continue. This ensures that the sink only exists when it's connected to something. Part-of: --- src/modules/module-tunnel-sink-new.c | 198 +++++++++++++++++++++++------------ 1 file changed, 133 insertions(+), 65 deletions(-) diff --git a/src/modules/module-tunnel-sink-new.c b/src/modules/module-tunnel-sink-new.c index 6f5fccbee..4de08893c 100644 --- a/src/modules/module-tunnel-sink-new.c +++ b/src/modules/module-tunnel-sink-new.c @@ -66,6 +66,21 @@ static void stream_set_buffer_attr_cb(pa_stream *stream, int success, void *user static void context_state_cb(pa_context *c, void *userdata); static void sink_update_requested_latency_cb(pa_sink *s); +struct tunnel_msg { + pa_msgobject parent; +}; + +typedef struct tunnel_msg tunnel_msg; +PA_DEFINE_PRIVATE_CLASS(tunnel_msg, pa_msgobject); + +enum { + TUNNEL_MESSAGE_CREATE_SINK_REQUEST, +}; + +enum { + TUNNEL_MESSAGE_SINK_CREATED = PA_SINK_MESSAGE_MAX, +}; + struct userdata { pa_module *module; pa_sink *sink; @@ -81,6 +96,7 @@ struct userdata { bool update_stream_bufferattr_after_connect; bool connected; + bool shutting_down; char *cookie_file; char *remote_server; @@ -90,6 +106,8 @@ struct userdata { pa_proplist *sink_proplist; pa_sample_spec sample_spec; pa_channel_map channel_map; + + tunnel_msg *msg; }; static const char* const valid_modargs[] = { @@ -188,7 +206,7 @@ static void thread_func(void *userdata) { goto fail; } - if (PA_UNLIKELY(u->sink->thread_info.rewind_requested)) + if (u->sink && PA_UNLIKELY(u->sink->thread_info.rewind_requested)) pa_sink_process_rewind(u->sink, 0); if (u->connected && @@ -305,6 +323,67 @@ static void stream_overflow_callback(pa_stream *stream, void *userdata) { pa_log_info("Server signalled buffer overrun."); } +static void on_sink_created(struct userdata *u) { + pa_proplist *proplist; + pa_buffer_attr bufferattr; + pa_usec_t requested_latency; + char *username = pa_get_user_name_malloc(); + char *hostname = pa_get_host_name_malloc(); + /* TODO: old tunnel put here the remote sink_name into stream name e.g. 'Null Output for lynxis@lazus' */ + char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname); + pa_xfree(hostname); + pa_xfree(username); + + pa_assert_io_context(); + + /* if we still don't have a sink, then sink creation failed, and we should + * kill this io thread */ + if (!u->sink) { + pa_log_error("Could not create a sink."); + u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP); + return; + } + + proplist = tunnel_new_proplist(u); + u->stream = pa_stream_new_with_proplist(u->context, + stream_name, + &u->sink->sample_spec, + &u->sink->channel_map, + proplist); + pa_proplist_free(proplist); + pa_xfree(stream_name); + + if (!u->stream) { + pa_log_error("Could not create a stream."); + u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP); + return; + } + + requested_latency = pa_sink_get_requested_latency_within_thread(u->sink); + if (requested_latency == (pa_usec_t) -1) + requested_latency = u->sink->thread_info.max_latency; + + reset_bufferattr(&bufferattr); + bufferattr.tlength = pa_usec_to_bytes(requested_latency, &u->sink->sample_spec); + + pa_log_debug("tlength requested at %lu.", (unsigned long) bufferattr.tlength); + + pa_stream_set_state_callback(u->stream, stream_state_cb, u); + pa_stream_set_buffer_attr_callback(u->stream, stream_changed_buffer_attr_cb, u); + pa_stream_set_underflow_callback(u->stream, stream_underflow_callback, u); + pa_stream_set_overflow_callback(u->stream, stream_overflow_callback, u); + if (pa_stream_connect_playback(u->stream, + u->remote_sink_name, + &bufferattr, + PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_DONT_MOVE | PA_STREAM_START_CORKED | PA_STREAM_AUTO_TIMING_UPDATE, + NULL, + NULL) < 0) { + pa_log_error("Could not connect stream."); + u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP); + } + u->connected = true; +} + static void context_state_cb(pa_context *c, void *userdata) { struct userdata *u = userdata; pa_assert(u); @@ -315,60 +394,18 @@ static void context_state_cb(pa_context *c, void *userdata) { case PA_CONTEXT_AUTHORIZING: case PA_CONTEXT_SETTING_NAME: break; - case PA_CONTEXT_READY: { - pa_proplist *proplist; - pa_buffer_attr bufferattr; - pa_usec_t requested_latency; - char *username = pa_get_user_name_malloc(); - char *hostname = pa_get_host_name_malloc(); - /* TODO: old tunnel put here the remote sink_name into stream name e.g. 'Null Output for lynxis@lazus' */ - char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname); - pa_xfree(hostname); - pa_xfree(username); - + case PA_CONTEXT_READY: + /* now that we're connected, ask the control thread to create a sink for + * us, and wait for that to complete before proceeding, we'll + * receive TUNNEL_MESSAGE_SINK_CREATED in response when the sink is + * created (see sink_process_msg_cb()) */ pa_log_debug("Connection successful. Creating stream."); pa_assert(!u->stream); + pa_assert(!u->sink); - proplist = tunnel_new_proplist(u); - u->stream = pa_stream_new_with_proplist(u->context, - stream_name, - &u->sink->sample_spec, - &u->sink->channel_map, - proplist); - pa_proplist_free(proplist); - pa_xfree(stream_name); - - if (!u->stream) { - pa_log_error("Could not create a stream."); - u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP); - return; - } - - requested_latency = pa_sink_get_requested_latency_within_thread(u->sink); - if (requested_latency == (pa_usec_t) -1) - requested_latency = u->sink->thread_info.max_latency; - - reset_bufferattr(&bufferattr); - bufferattr.tlength = pa_usec_to_bytes(requested_latency, &u->sink->sample_spec); - - pa_log_debug("tlength requested at %lu.", (unsigned long) bufferattr.tlength); - - pa_stream_set_state_callback(u->stream, stream_state_cb, userdata); - pa_stream_set_buffer_attr_callback(u->stream, stream_changed_buffer_attr_cb, userdata); - pa_stream_set_underflow_callback(u->stream, stream_underflow_callback, userdata); - pa_stream_set_overflow_callback(u->stream, stream_overflow_callback, userdata); - if (pa_stream_connect_playback(u->stream, - u->remote_sink_name, - &bufferattr, - PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_DONT_MOVE | PA_STREAM_START_CORKED | PA_STREAM_AUTO_TIMING_UPDATE, - NULL, - NULL) < 0) { - pa_log_error("Could not connect stream."); - u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP); - } - u->connected = true; + pa_log_debug("Asking ctl thread to create sink."); + pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->msg), TUNNEL_MESSAGE_CREATE_SINK_REQUEST, u, 0, NULL, NULL); break; - } case PA_CONTEXT_FAILED: pa_log_debug("Context failed: %s.", pa_strerror(pa_context_errno(u->context))); u->connected = false; @@ -454,6 +491,9 @@ static int sink_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t of *((int64_t*) data) = remote_latency; return 0; } + case TUNNEL_MESSAGE_SINK_CREATED: + on_sink_created(u); + return 0; } return pa_sink_process_msg(o, code, data, offset, chunk); } @@ -492,7 +532,14 @@ static int sink_set_state_in_io_thread_cb(pa_sink *s, pa_sink_state_t new_state, return 0; } -static int create_sink(struct userdata *u) { +/* Creates a sink in the main thread. + * + * This method is called when we receive a message from the io thread that a + * connection has been established with the server. We defer creation of the + * sink until the connection is established, because we don't have a sink if + * the remote server isn't there. + */ +static void create_sink(struct userdata *u) { pa_sink_new_data sink_data; pa_assert_ctl_context(); @@ -510,11 +557,9 @@ static int create_sink(struct userdata *u) { if (!(u->sink = pa_sink_new(u->module->core, &sink_data, PA_SINK_LATENCY | PA_SINK_DYNAMIC_LATENCY | PA_SINK_NETWORK))) { pa_log("Failed to create sink."); - goto fail; + goto finish; } - pa_sink_new_data_done(&sink_data); - u->sink->userdata = u; u->sink->parent.process_msg = sink_process_msg_cb; u->sink->set_state_in_io_thread = sink_set_state_in_io_thread_cb; @@ -525,12 +570,35 @@ static int create_sink(struct userdata *u) { pa_sink_set_asyncmsgq(u->sink, u->thread_mq->inq); pa_sink_set_rtpoll(u->sink, u->rtpoll); - return 0; + pa_sink_put(u->sink); -fail: +finish: pa_sink_new_data_done(&sink_data); - return -1; + /* tell any interested io threads that the sink they asked for has now been + * created (even if we failed, we still notify the thread, so they can + * either handle or kill the thread, rather than deadlock waiting for a + * message that will never come */ + pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), TUNNEL_MESSAGE_SINK_CREATED, u, 0, NULL); +} + +/* Runs in PA mainloop context */ +static int tunnel_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) { + struct userdata *u = (struct userdata *) data; + + pa_assert(u); + pa_assert_ctl_context(); + + if (u->shutting_down) + return 0; + + switch (code) { + case TUNNEL_MESSAGE_CREATE_SINK_REQUEST: + create_sink(u); + break; + } + + return 0; } int pa__init(pa_module *m) { @@ -580,6 +648,9 @@ int pa__init(pa_module *m) { goto fail; } + u->msg = pa_msgobject_new(tunnel_msg); + u->msg->parent.process_msg = tunnel_process_msg; + /* The rtpoll created here is never run. It is only necessary to avoid crashes * when module-tunnel-sink-new is used together with module-loopback or * module-combine-sink. Both modules base their asyncmsq on the rtpoll provided @@ -605,18 +676,11 @@ int pa__init(pa_module *m) { goto fail; } - if (create_sink(u) < 0) { - pa_log("Failed to create sink."); - goto fail; - } - if (!(u->thread = pa_thread_new("tunnel-sink", thread_func, u))) { pa_log("Failed to create thread."); goto fail; } - pa_sink_put(u->sink); - pa_modargs_free(ma); pa_xfree(default_sink_name); @@ -642,6 +706,8 @@ void pa__done(pa_module *m) { if (!(u = m->userdata)) return; + u->shutting_down = true; + if (u->sink) pa_sink_unlink(u->sink); @@ -679,5 +745,7 @@ void pa__done(pa_module *m) { if (u->sink_name) pa_xfree(u->sink_name); + pa_xfree(u->msg); + pa_xfree(u); } -- cgit v1.2.1