refactor stream attaching/detaching

Move repetitive code into convenience functions. No changes in
behaviour.

src/pulsecore/sink-input.c

@@ -2298,6 +2298,30 @@ int pa_sink_input_update_rate(pa_sink_input *i) {
     return 0;
 }
 
+/* Called from the IO thread. */
+void pa_sink_input_attach(pa_sink_input *i) {
+    pa_assert(i);
+    pa_assert(!i->thread_info.attached);
+
+    i->thread_info.attached = true;
+
+    if (i->attach)
+        i->attach(i);
+}
+
+/* Called from the IO thread. */
+void pa_sink_input_detach(pa_sink_input *i) {
+    pa_assert(i);
+
+    if (!i->thread_info.attached)
+        return;
+
+    i->thread_info.attached = false;
+
+    if (i->detach)
+        i->detach(i);
+}
+
 /* Called from the main thread. */
 void pa_sink_input_set_volume_direct(pa_sink_input *i, const pa_cvolume *volume) {
     pa_cvolume old_volume;

src/pulsecore/sink-input.h

@@ -413,6 +413,31 @@ bool pa_sink_input_process_underrun(pa_sink_input *i);
 
 pa_memchunk* pa_sink_input_get_silence(pa_sink_input *i, pa_memchunk *ret);
 
+/* Calls the attach() callback if it's set. The input must be in detached
+ * state. */
+void pa_sink_input_attach(pa_sink_input *i);
+
+/* Calls the detach() callback if it's set and the input is attached. The input
+ * is allowed to be already detached, in which case this does nothing.
+ *
+ * The reason why this can be called for already-detached inputs is that when
+ * a filter sink's input is detached, it has to detach also all inputs
+ * connected to the filter sink. In case the filter sink's input was detached
+ * because the filter sink is being removed, those other inputs will be moved
+ * to another sink or removed, and moving and removing involve detaching the
+ * inputs, but the inputs at that point are already detached.
+ *
+ * XXX: Moving or removing an input also involves sending messages to the
+ * input's sink. If the input's sink is a detached filter sink, shouldn't
+ * sending messages to it be prohibited? The messages are processed in the
+ * root sink's IO thread, and when the filter sink is detached, it would seem
+ * logical to prohibit any interaction with the IO thread that isn't any more
+ * associated with the filter sink. Currently sending messages to detached
+ * filter sinks mostly works, because the filter sinks don't update their
+ * asyncmsgq pointer when detaching, so messages still find their way to the
+ * old IO thread. */
+void pa_sink_input_detach(pa_sink_input *i);
+
 /* Called from the main thread, from sink.c only. The normal way to set the
  * sink input volume is to call pa_sink_input_set_volume(), but the flat volume
  * logic in sink.c needs also a function that doesn't do all the extra stuff

src/pulsecore/sink.c

@@ -2490,11 +2490,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
                 i->thread_info.sync_next->thread_info.sync_prev = i;
             }
 
-            pa_assert(!i->thread_info.attached);
+            pa_sink_input_attach(i);
-            i->thread_info.attached = true;
-
-            if (i->attach)
-                i->attach(i);
 
             pa_sink_input_set_state_within_thread(i, i->state);
 
@@ -2536,12 +2532,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
              * sink input handling a few lines down at
              * PA_SINK_MESSAGE_START_MOVE, too. */
 
-            if (i->thread_info.attached) {
+            pa_sink_input_detach(i);
-                i->thread_info.attached = false;
-
-                if (i->detach)
-                    i->detach(i);
-            }
 
             pa_sink_input_set_state_within_thread(i, i->state);
 
@@ -2636,12 +2627,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
                 }
             }
 
-            if (i->thread_info.attached) {
+            pa_sink_input_detach(i);
-                i->thread_info.attached = false;
-
-                if (i->detach)
-                    i->detach(i);
-            }
 
             /* Let's remove the sink input ...*/
             pa_hashmap_remove_and_free(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index));
@@ -2667,11 +2653,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
 
             pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i));
 
-            pa_assert(!i->thread_info.attached);
+            pa_sink_input_attach(i);
-            i->thread_info.attached = true;
-
-            if (i->attach)
-                i->attach(i);
 
             if (i->thread_info.state != PA_SINK_INPUT_CORKED) {
                 pa_usec_t usec = 0;
@@ -2933,14 +2915,8 @@ void pa_sink_detach_within_thread(pa_sink *s) {
     pa_sink_assert_io_context(s);
     pa_assert(PA_SINK_IS_LINKED(s->thread_info.state));
 
-    PA_HASHMAP_FOREACH(i, s->thread_info.inputs, state) {
+    PA_HASHMAP_FOREACH(i, s->thread_info.inputs, state)
-        if (i->thread_info.attached) {
+        pa_sink_input_detach(i);
-            i->thread_info.attached = false;
-
-            if (i->detach)
-                i->detach(i);
-        }
-    }
 
     if (s->monitor_source)
         pa_source_detach_within_thread(s->monitor_source);
@@ -2955,13 +2931,8 @@ void pa_sink_attach_within_thread(pa_sink *s) {
     pa_sink_assert_io_context(s);
     pa_assert(PA_SINK_IS_LINKED(s->thread_info.state));
 
-    PA_HASHMAP_FOREACH(i, s->thread_info.inputs, state) {
+    PA_HASHMAP_FOREACH(i, s->thread_info.inputs, state)
-        pa_assert(!i->thread_info.attached);
+        pa_sink_input_attach(i);
-        i->thread_info.attached = true;
-
-        if (i->attach)
-            i->attach(i);
-    }
 
     if (s->monitor_source)
         pa_source_attach_within_thread(s->monitor_source);

src/pulsecore/source-output.c

@@ -1753,6 +1753,30 @@ int pa_source_output_update_rate(pa_source_output *o) {
     return 0;
 }
 
+/* Called from the IO thread. */
+void pa_source_output_attach(pa_source_output *o) {
+    pa_assert(o);
+    pa_assert(!o->thread_info.attached);
+
+    o->thread_info.attached = true;
+
+    if (o->attach)
+        o->attach(o);
+}
+
+/* Called from the IO thread. */
+void pa_source_output_detach(pa_source_output *o) {
+    pa_assert(o);
+
+    if (!o->thread_info.attached)
+        return;
+
+    o->thread_info.attached = false;
+
+    if (o->detach)
+        o->detach(o);
+}
+
 /* Called from the main thread. */
 void pa_source_output_set_volume_direct(pa_source_output *o, const pa_cvolume *volume) {
     pa_cvolume old_volume;

src/pulsecore/source-output.h

@@ -349,6 +349,31 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int
 
 pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec);
 
+/* Calls the attach() callback if it's set. The output must be in detached
+ * state. */
+void pa_source_output_attach(pa_source_output *o);
+
+/* Calls the detach() callback if it's set and the output is attached. The
+ * output is allowed to be already detached, in which case this does nothing.
+ *
+ * The reason why this can be called for already-detached outputs is that when
+ * a filter source's output is detached, it has to detach also all outputs
+ * connected to the filter source. In case the filter source's output was
+ * detached because the filter source is being removed, those other outputs
+ * will be moved to another source or removed, and moving and removing involve
+ * detaching the outputs, but the outputs at that point are already detached.
+ *
+ * XXX: Moving or removing an output also involves sending messages to the
+ * output's source. If the output's source is a detached filter source,
+ * shouldn't sending messages to it be prohibited? The messages are processed
+ * in the root source's IO thread, and when the filter source is detached, it
+ * would seem logical to prohibit any interaction with the IO thread that isn't
+ * any more associated with the filter source. Currently sending messages to
+ * detached filter sources mostly works, because the filter sources don't
+ * update their asyncmsgq pointer when detaching, so messages still find their
+ * way to the old IO thread. */
+void pa_source_output_detach(pa_source_output *o);
+
 /* Called from the main thread, from source.c only. The normal way to set the
  * source output volume is to call pa_source_output_set_volume(), but the flat
  * volume logic in source.c needs also a function that doesn't do all the extra

src/pulsecore/source.c

@@ -2034,11 +2034,7 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
                 pa_hashmap_put(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index), o);
             }
 
-            pa_assert(!o->thread_info.attached);
+            pa_source_output_attach(o);
-            o->thread_info.attached = true;
-
-            if (o->attach)
-                o->attach(o);
 
             pa_source_output_set_state_within_thread(o, o->state);
 
@@ -2062,12 +2058,7 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
 
             pa_source_output_set_state_within_thread(o, o->state);
 
-            if (o->thread_info.attached) {
+            pa_source_output_detach(o);
-                o->thread_info.attached = false;
-
-                if (o->detach)
-                    o->detach(o);
-            }
 
             if (o->thread_info.direct_on_input) {
                 pa_hashmap_remove(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index));
@@ -2289,14 +2280,8 @@ void pa_source_detach_within_thread(pa_source *s) {
     pa_source_assert_io_context(s);
     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 
-    PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state) {
+    PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
-        if (o->thread_info.attached) {
+        pa_source_output_detach(o);
-            o->thread_info.attached = false;
-
-            if (o->detach)
-                o->detach(o);
-        }
-    }
 }
 
 /* Called from IO thread */
@@ -2308,13 +2293,8 @@ void pa_source_attach_within_thread(pa_source *s) {
     pa_source_assert_io_context(s);
     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 
-    PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state) {
+    PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
-        pa_assert(!o->thread_info.attached);
+        pa_source_output_attach(o);
-        o->thread_info.attached = true;
-
-        if (o->attach)
-            o->attach(o);
-    }
 }
 
 /* Called from IO thread */