stream-manager: handle ramp finished task only at the main thread

Sometimes sd handle is accessed simultaneously in different thread, which may leads to memory corruption.

This patch makes to handle ramp finish task directly only if inside of main thread.
Otherwise, request a message to make it handle from the main thread.
To avoid possible deadlock, request by posting(async) instead of send(sync).

[Version] 13.0.52
[Issue Type] Bug fix

Change-Id: I178dbc5b29adccc6f74e41b7e105681c3fd570d0

diff --git a/packaging/pulseaudio-modules-tizen.spec b/packaging/pulseaudio-modules-tizen.spec
index 112f948..e5e4575 100644 (file)
--- a/packaging/pulseaudio-modules-tizen.spec
+++ b/packaging/pulseaudio-modules-tizen.spec
@@ -1,6 +1,6 @@
 Name:             pulseaudio-modules-tizen
 Summary:          Pulseaudio modules for Tizen
-Version:          13.0.51
+Version:          13.0.52
 Release:          0
 Group:            Multimedia/Audio
 License:          LGPL-2.1+

diff --git a/src/stream-manager-priv.h b/src/stream-manager-priv.h
index 390b156..4a8b7be 100644 (file)
--- a/src/stream-manager-priv.h
+++ b/src/stream-manager-priv.h
@@ -233,6 +233,10 @@ typedef struct _filter_info {
     uint32_t n_controls;
 } filter_info;
 
+typedef struct _stream_manager_msg {
+    pa_msgobject parent;
+} stream_manager_msg;
+
 struct _stream_manager {
     PA_REFCNT_DECLARE;
 
@@ -255,6 +259,10 @@ struct _stream_manager {
     pa_time_event *time_event_for_unmute;
     pa_hashmap *filter_infos;
 
+    pa_thread_mq thread_mq;
+    pa_rtpoll *rtpoll;
+    stream_manager_msg *msg;
+
     pa_hook_slot
         *sink_input_new_slot,
         *sink_input_put_slot,

diff --git a/src/stream-manager.c b/src/stream-manager.c
index 076b842..2b6d489 100644 (file)
--- a/src/stream-manager.c
+++ b/src/stream-manager.c
@@ -129,6 +129,62 @@ static const char* notify_command_type_str[] = {
 #define STREAM_MAP_STREAM_AVAIL_OUT_DEVICES "avail-out-devices"
 #define STREAM_MAP_STREAM_AVAIL_FRAMEWORKS "avail-frameworks"
 
+PA_DEFINE_PRIVATE_CLASS(stream_manager_msg, pa_msgobject);
+
+enum {
+    MESSAGE_RAMP_FINISHED,
+};
+
+struct stream_manager_param {
+    pa_stream_manager *m;
+    pa_sink_input *sink_input;
+    uint32_t index;
+};
+
+/* Called from main context */
+static void process_ramp_finish(struct stream_manager_param *param) {
+    stream_ducking *sd;
+    void *state;
+
+    /* Find a context id from all the ducked stream list by this stream index.
+    * Check the number of managed streams of the context id, if it is the last one
+    * then broadcast a signal with context id.*/
+    PA_HASHMAP_FOREACH(sd, param->m->stream_duckings, state) {
+        if (!pa_idxset_get_by_data(sd->idx_ducking_streams, param->sink_input, NULL)) {
+            pa_log_debug("not found matched stream(%p, index:%u) in sd(%p)",
+                        param->sink_input, param->index, sd);
+            continue;
+        }
+
+        pa_log_info("found matched stream(%p, index:%u) in sd(%p, ducking_stream_count:%d, state:%u)",
+            param->sink_input, param->index, sd, sd->ducking_stream_count, sd->state);
+
+        if (sd->ducking_stream_count <= 0)
+            continue;
+
+        /* Remove trigger when unducked */
+        if (sd->state == STREAM_DUCKING_STATE_UNDUCKING)
+            pa_idxset_remove_by_data(sd->idx_ducking_streams, (void *)param->sink_input, NULL);
+
+        /* Send signal when all streams are ducked.
+            * Note that the condition of increasing count value below is located in
+            * handle_activate_ducking() of DBus handler. */
+        if (--sd->ducking_stream_count == 0) {
+            if (sd->state == STREAM_DUCKING_STATE_DUCKING) {
+                sd->state = STREAM_DUCKING_STATE_DUCKED;
+            } else if (sd->state == STREAM_DUCKING_STATE_UNDUCKING) {
+                sd->state = STREAM_DUCKING_STATE_UNDUCKED;
+            } else {
+                pa_log_warn("sd->state(%d), already ducked or unducked, skip sending signal", sd->state);
+                continue;
+            }
+
+            pa_log_info("send signal for ramp finished - sd(%p, state:%u)", sd, sd->state);
+            send_ducking_state_changed_signal(pa_dbus_connection_get(param->m->dbus_conn), sd->trigger_index, is_stream_ducked(sd));
+        }
+    }
+}
+
 static bool is_valid_notify_command(notify_command_type_t command) {
     return (command < sizeof(notify_command_type_str) / sizeof(char *));
 }
@@ -1944,7 +2000,7 @@ static process_stream_result_t handle_command_prepare(pa_stream_manager *m, void
     pa_assert(stream);
 
     if (type == STREAM_SINK_INPUT) {
-        /* Parse request formats for samplerate, channel, format infomation */
+        /* Parse request formats for samplerate, channel, format information */
         if (((pa_sink_input_new_data*)stream)->req_formats) {
             req_format = pa_idxset_first(((pa_sink_input_new_data*)stream)->req_formats, NULL);
             if (req_format && req_format->plist) {
@@ -2544,7 +2600,7 @@ static pa_hook_result_t sink_input_state_changed_cb(pa_core *core, pa_sink_input
     pa_assert(i);
     pa_assert(m);
 
-    pa_log_debug("sink-input(%p, index:%u), state(%d)", i, i->index, i->state);
+    pa_log_debug("sink-input(%p, index:%u, state:%d)", i, i->index, i->state);
 
     switch (i->state) {
     case PA_SINK_INPUT_CORKED:
@@ -2592,9 +2648,15 @@ static pa_hook_result_t sink_input_move_finish_cb(pa_core *core, pa_sink_input *
     return PA_HOOK_OK;
 }
 
+static bool is_in_main_thread()
+{
+    return (getpid() == gettid());
+}
+
+/* Called from either IO thread context or main context */
+/* FIXME : make this callback be invoked from the main context only */
 static pa_hook_result_t sink_input_ramp_finish_cb(pa_core *core, pa_sink_input *i, pa_stream_manager *m) {
-    stream_ducking *sd;
-    void *state;
+    struct stream_manager_param param;
 
     pa_core_assert_ref(core);
     pa_sink_input_assert_ref(i);
@@ -2603,43 +2665,20 @@ static pa_hook_result_t sink_input_ramp_finish_cb(pa_core *core, pa_sink_input *
     if (core->state == PA_CORE_SHUTDOWN)
         return PA_HOOK_OK;
 
-    pa_log_debug("sink-input(%p, index:%u)", i, i->index);
-
-    /* Find a context id from all the ducked stream list by this stream index.
-     * Check the number of managed streams of the context id, if it is the last one
-     * then broadcast a signal with context id.*/
-    PA_HASHMAP_FOREACH(sd, m->stream_duckings, state) {
-        if (!pa_idxset_get_by_data(sd->idx_ducking_streams, i, NULL)) {
-            pa_log_debug("not found matched stream(%p, index:%u) in sd(%p)", i, i->index, sd);
-            continue;
-        }
-
-        pa_log_info("found matched stream(%p, index:%u) in sd(%p, ducking_stream_count:%d, state:%u)",
-            i, i->index, sd, sd->ducking_stream_count, sd->state);
-
-        if (sd->ducking_stream_count <= 0)
-            continue;
-
-        /* Remove trigger when unducked */
-        if (sd->state == STREAM_DUCKING_STATE_UNDUCKING)
-            pa_idxset_remove_by_data(sd->idx_ducking_streams, (void *)i, NULL);
+    param.m = m;
+    param.sink_input = i;
+    param.index = i->index;
 
-        /* Send signal when all streams are ducked.
-        * Note that the condition of increasing count value below is located in
-        * handle_activate_ducking() of DBus handler. */
-        if (--sd->ducking_stream_count == 0) {
-            if (sd->state == STREAM_DUCKING_STATE_DUCKING) {
-                sd->state = STREAM_DUCKING_STATE_DUCKED;
-            } else if (sd->state == STREAM_DUCKING_STATE_UNDUCKING) {
-                sd->state = STREAM_DUCKING_STATE_UNDUCKED;
-            } else {
-                pa_log_warn("sd->state(%d), already ducked or unducked, skip sending signal", sd->state);
-                continue;
-            }
+    pa_log_info("sink-input(%p, index:%u)", i, i->index);
 
-            pa_log_info("send signal for ramp finished - sd(%p, state:%u)", sd, sd->state);
-            send_ducking_state_changed_signal(pa_dbus_connection_get(m->dbus_conn), sd->trigger_index, is_stream_ducked(sd));
-        }
+    if (is_in_main_thread()) {
+        process_ramp_finish(&param);
+        pa_log_info("sink-input(%p, index:%u) : direct process_ramp_finish() done", i, i->index);
+    } else {
+        /* Post message to make process_ramp_finish() run from main thread */
+        pa_asyncmsgq_post(m->thread_mq.outq, PA_MSGOBJECT(m->msg), MESSAGE_RAMP_FINISHED,
+                        pa_xmemdup(&param, sizeof(struct stream_manager_param)), 0, NULL, pa_xfree);
+        pa_log_info("sink-input(%p, index:%u) : posting MESSAGE_RAMP_FINISHED done", i, i->index);
     }
 
     return PA_HOOK_OK;
@@ -3743,6 +3782,30 @@ const char* pa_stream_manager_get_volume_type(pa_stream_manager *m, stream_type_
     return s->volume_types[stream_type == STREAM_SINK_INPUT ? STREAM_DIRECTION_OUT : STREAM_DIRECTION_IN];
 }
 
+static int stream_manager_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
+    struct stream_manager_param *param = (struct stream_manager_param *)data;
+
+    pa_assert(param);
+    pa_assert(param->m);
+    pa_assert(param->m->core);
+
+    pa_log_info("code(%d), sink-input(%p, index:%u)", code, param->sink_input, param->index);
+
+    if (param->m->core->state == PA_CORE_SHUTDOWN)
+        return 0;
+
+    switch (code) {
+        case MESSAGE_RAMP_FINISHED:
+            process_ramp_finish(param);
+            break;
+
+        default:
+            pa_assert_not_reached();
+    }
+
+    return 0;
+}
+
 pa_stream_manager* pa_stream_manager_get(pa_core *c) {
     pa_stream_manager *m;
 
@@ -3757,6 +3820,14 @@ pa_stream_manager* pa_stream_manager_get(pa_core *c) {
     PA_REFCNT_INIT(m);
     m->core = c;
 
+    m->rtpoll = pa_rtpoll_new();
+    if (pa_thread_mq_init(&m->thread_mq, m->core->mainloop, m->rtpoll) < 0) {
+        pa_log("pa_thread_mq_init() failed.");
+        goto fail;
+    }
+    m->msg = pa_msgobject_new(stream_manager_msg);
+    m->msg->parent.process_msg = stream_manager_process_msg;
+
     if (!(m->hal = pa_hal_interface_get(c)))
         goto fail;
 
@@ -3818,6 +3889,11 @@ pa_stream_manager* pa_stream_manager_get(pa_core *c) {
 
 fail:
     pa_log_error("failed to initialize stream-manager");
+
+    pa_thread_mq_done(&m->thread_mq);
+    pa_rtpoll_free(m->rtpoll);
+    pa_xfree(m->msg);
+
     deinit_volumes(m);
     deinit_stream_map(m);
     deinit_filters(m);
@@ -3899,6 +3975,10 @@ void pa_stream_manager_unref(pa_stream_manager *m) {
     if (PA_REFCNT_DEC(m) > 0)
         return;
 
+    pa_thread_mq_done(&m->thread_mq);
+    pa_rtpoll_free(m->rtpoll);
+    pa_xfree(m->msg);
+
     free_hook_slots(m);
 
     if (m->comm.comm)