res_stasis: Add control ref to playback and recording structs.

The stasis_app_playback and stasis_app_recording structs need to have a
struct stasis_app_control ref.  Other threads can get a reference to the
playback and recording structs from their respective global container.
These other threads can then use the control pointer they contain after
the control struct has gone.

* Add control ref to stasis_app_playback and stasis_app_recording structs.

With the refs added, the control command queue can now have a circular
control reference which will cause the control struct to never get
released if the control's command queue is not flushed when the channel
leaves the Stasis application.  Also the command queue needs better
protection from adding commands if the control->is_done flag is set.

* Flush the control command queue on exit.

ASTERISK-25882 #close

Change-Id: I3cf1fb59cbe6f50f20d9e35a2c07ac07d7f4320d

diff --git a/include/asterisk/stasis_app.h b/include/asterisk/stasis_app.h
index f2b07e0bfebb93eba2150023dc92cb49fd2780bc..90ef82ebf435ab0efb41cbcf417fe3783d76d7b5 100644 (file)
--- a/include/asterisk/stasis_app.h
+++ b/include/asterisk/stasis_app.h
@@ -439,6 +439,16 @@ void stasis_app_control_execute_until_exhausted(
 int stasis_app_control_is_done(
        struct stasis_app_control *control);
 
+/*!
+ * \brief Flush the control command queue.
+ * \since 13.9.0
+ *
+ * \param control Control object to flush command queue.
+ *
+ * \return Nothing
+ */
+void stasis_app_control_flush_queue(struct stasis_app_control *control);
+
 /*!
  * \brief Returns the uniqueid of the channel associated with this control
  *

diff --git a/res/ari/resource_bridges.c b/res/ari/resource_bridges.c
index 7b9b94665f3cff40b01af87ae7f1e0307faad911..93bdc62171695acc52629d76a4e03e7bd9264e61 100644 (file)
--- a/res/ari/resource_bridges.c
+++ b/res/ari/resource_bridges.c
@@ -296,10 +296,11 @@ static void *bridge_channel_control_thread(void *data)
        thread_data = NULL;
 
        stasis_app_control_execute_until_exhausted(bridge_channel, control);
+       stasis_app_control_flush_queue(control);
 
-       ast_hangup(bridge_channel);
-       ao2_cleanup(control);
        stasis_forward_cancel(forward);
+       ao2_cleanup(control);
+       ast_hangup(bridge_channel);
        return NULL;
 }
 

diff --git a/res/res_stasis.c b/res/res_stasis.c
index dc469c8cc40e00a97bd061c6b9def63232892f22..5afb90e3c3f2f00a1ab0d1ec713bf94791d40b8c 100644 (file)
--- a/res/res_stasis.c
+++ b/res/res_stasis.c
@@ -1179,6 +1179,11 @@ int stasis_app_control_is_done(struct stasis_app_control *control)
        return control_is_done(control);
 }
 
+void stasis_app_control_flush_queue(struct stasis_app_control *control)
+{
+       control_flush_queue(control);
+}
+
 struct ast_datastore_info set_end_published_info = {
        .type = "stasis_end_published",
 };
@@ -1371,6 +1376,8 @@ int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
                remove_stasis_end_published(chan);
        }
 
+       control_flush_queue(control);
+
        /* Stop any lingering silence generator */
        control_silence_stop_now(control);
 

diff --git a/res/res_stasis_playback.c b/res/res_stasis_playback.c
index 74336abdcf25a0c933c04bd84171445492cefa44..97191c26dd1153d4fdb19227447b9c798681314a 100644 (file)
--- a/res/res_stasis_playback.c
+++ b/res/res_stasis_playback.c
@@ -118,6 +118,7 @@ static void playback_dtor(void *obj)
 {
        struct stasis_app_playback *playback = obj;
 
+       ao2_cleanup(playback->control);
        ast_string_field_free_memory(playback);
 }
 
@@ -143,6 +144,7 @@ static struct stasis_app_playback *playback_create(
                ast_string_field_set(playback, id, uuid);
        }
 
+       ao2_ref(control, +1);
        playback->control = control;
 
        ao2_ref(playback, +1);

diff --git a/res/res_stasis_recording.c b/res/res_stasis_recording.c
index dcabfa699509d0f5af8845a73523323523e5e0ab..c6f9002e3219c0ac4af1ca03dc37c76263f0c5b9 100644 (file)
--- a/res/res_stasis_recording.c
+++ b/res/res_stasis_recording.c
@@ -358,6 +358,7 @@ static void recording_dtor(void *obj)
        struct stasis_app_recording *recording = obj;
 
        ast_free(recording->absolute_name);
+       ao2_cleanup(recording->control);
        ao2_cleanup(recording->options);
 }
 
@@ -413,6 +414,7 @@ struct stasis_app_recording *stasis_app_control_record(
 
        ao2_ref(options, +1);
        recording->options = options;
+       ao2_ref(control, +1);
        recording->control = control;
        recording->state = STASIS_APP_RECORDING_STATE_QUEUED;
 

diff --git a/res/stasis/control.c b/res/stasis/control.c
index a1004248f3569b2212906158abe7b10409f05efc..86f94423de3828b473bb0578707c8ba6f5caefb6 100644 (file)
--- a/res/stasis/control.c
+++ b/res/stasis/control.c
@@ -251,6 +251,11 @@ static struct stasis_app_command *exec_command_on_condition(
        }
 
        ao2_lock(control->command_queue);
+       if (control->is_done) {
+               ao2_unlock(control->command_queue);
+               ao2_ref(command, -1);
+               return NULL;
+       }
        if (can_exec_fn && (retval = can_exec_fn(control))) {
                ao2_unlock(control->command_queue);
                command_complete(command, retval);
@@ -402,7 +407,10 @@ int control_is_done(struct stasis_app_control *control)
 
 void control_mark_done(struct stasis_app_control *control)
 {
+       /* Locking necessary to sync with other threads adding commands to the queue. */
+       ao2_lock(control->command_queue);
        control->is_done = 1;
+       ao2_unlock(control->command_queue);
 }
 
 struct stasis_app_control_continue_data {
@@ -427,7 +435,7 @@ static int app_control_continue(struct stasis_app_control *control,
        /* Called from stasis_app_exec thread; no lock needed */
        ast_explicit_goto(control->channel, continue_data->context, continue_data->extension, continue_data->priority);
 
-       control->is_done = 1;
+       control_mark_done(control);
 
        return 0;
 }
@@ -1115,24 +1123,36 @@ int stasis_app_control_queue_control(struct stasis_app_control *control,
        return ast_queue_control(control->channel, frame_type);
 }
 
+void control_flush_queue(struct stasis_app_control *control)
+{
+       struct ao2_iterator iter;
+       struct stasis_app_command *command;
+
+       iter = ao2_iterator_init(control->command_queue, AO2_ITERATOR_UNLINK);
+       while ((command = ao2_iterator_next(&iter))) {
+               command_complete(command, -1);
+               ao2_ref(command, -1);
+       }
+       ao2_iterator_destroy(&iter);
+}
+
 int control_dispatch_all(struct stasis_app_control *control,
        struct ast_channel *chan)
 {
        int count = 0;
-       struct ao2_iterator i;
-       void *obj;
+       struct ao2_iterator iter;
+       struct stasis_app_command *command;
 
        ast_assert(control->channel == chan);
 
-       i = ao2_iterator_init(control->command_queue, AO2_ITERATOR_UNLINK);
-
-       while ((obj = ao2_iterator_next(&i))) {
-               RAII_VAR(struct stasis_app_command *, command, obj, ao2_cleanup);
+       iter = ao2_iterator_init(control->command_queue, AO2_ITERATOR_UNLINK);
+       while ((command = ao2_iterator_next(&iter))) {
                command_invoke(command, control, chan);
+               ao2_ref(command, -1);
                ++count;
        }
+       ao2_iterator_destroy(&iter);
 
-       ao2_iterator_destroy(&i);
        return count;
 }
 

diff --git a/res/stasis/control.h b/res/stasis/control.h
index d053a35f7c3aa98a44817841d841c43658833cda..1d37a494aec9f49fcef7deda13d5ff9dccfdb62e 100644 (file)
--- a/res/stasis/control.h
+++ b/res/stasis/control.h
@@ -40,6 +40,16 @@
  */
 struct stasis_app_control *control_create(struct ast_channel *channel, struct stasis_app *app);
 
+/*!
+ * \brief Flush the control command queue.
+ * \since 13.9.0
+ *
+ * \param control Control object to flush command queue.
+ *
+ * \return Nothing
+ */
+void control_flush_queue(struct stasis_app_control *control);
+
 /*!
  * \brief Dispatch all commands enqueued to this control.
  *