elements: Fix pad callbacks so they handle when parent goes away

1) We need to lock and get a strong ref to the parent, if still there.
2) If it has gone away, we need to handle that gracefully.

This is necessary in order to safely modify a running pipeline. Has been
observed when a streaming thread is doing a buffer_alloc() while an
application thread sends an event on a pad further downstream, and from
within a pad probe (holding STREAM_LOCK) carries out the pipeline plumbing
while the streaming thread has its buffer_alloc() in progress.

plugins/elements/gstfunnel.c

@@ -191,7 +191,10 @@ gst_funnel_sink_buffer_alloc (GstPad * pad, guint64 offset, guint size,
     GstCaps * caps, GstBuffer ** buf)
 {
   GstFunnel *funnel = GST_FUNNEL (gst_pad_get_parent_element (pad));
-  GstFlowReturn ret = GST_FLOW_OK;
+  GstFlowReturn ret;
+
+  if (G_UNLIKELY (funnel == NULL))
+    return GST_FLOW_WRONG_STATE;
 
   ret = gst_pad_alloc_buffer (funnel->srcpad, offset, size, caps, buf);
 
@@ -246,6 +249,9 @@ gst_funnel_sink_getcaps (GstPad * pad)
   GstFunnel *funnel = GST_FUNNEL (gst_pad_get_parent (pad));
   GstCaps *caps;
 
+  if (G_UNLIKELY (funnel == NULL))
+    return gst_caps_new_any ();
+
   caps = gst_pad_peer_get_caps_reffed (funnel->srcpad);
   if (caps == NULL)
     caps = gst_caps_copy (gst_pad_get_pad_template_caps (pad));
@@ -327,6 +333,11 @@ gst_funnel_sink_event (GstPad * pad, GstEvent * event)
   gboolean forward = TRUE;
   gboolean res = TRUE;
 
+  if (G_UNLIKELY (funnel == NULL)) {
+    gst_event_unref (event);
+    return FALSE;
+  }
+
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_NEWSEGMENT:
     {
@@ -346,7 +357,6 @@ gst_funnel_sink_event (GstPad * pad, GstEvent * event)
       GST_OBJECT_UNLOCK (funnel);
 
       forward = FALSE;
-      gst_event_unref (event);
     }
       break;
     case GST_EVENT_FLUSH_STOP:
@@ -363,6 +373,8 @@ gst_funnel_sink_event (GstPad * pad, GstEvent * event)
 
   if (forward)
     res = gst_pad_push_event (funnel->srcpad, event);
+  else
+    gst_event_unref (event);
 
   gst_object_unref (funnel);
 
@@ -379,7 +391,10 @@ gst_funnel_src_event (GstPad * pad, GstEvent * event)
   gboolean done = FALSE;
 
   funnel = gst_pad_get_parent_element (pad);
-  g_return_val_if_fail (funnel != NULL, FALSE);
+  if (G_UNLIKELY (funnel == NULL)) {
+    gst_event_unref (event);
+    return FALSE;
+  }
 
   iter = gst_element_iterate_sink_pads (funnel);
 

plugins/elements/gstinputselector.c

@@ -386,6 +386,10 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
   GstPad *active_sinkpad;
 
   sel = GST_INPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL)) {
+    gst_event_unref (event);
+    return FALSE;
+  }
   selpad = GST_SELECTOR_PAD_CAST (pad);
 
   GST_INPUT_SELECTOR_LOCK (sel);
@@ -513,6 +517,8 @@ gst_selector_pad_getcaps (GstPad * pad)
   GstCaps *caps;
 
   sel = GST_INPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL))
+    return gst_caps_new_any ();
 
   GST_DEBUG_OBJECT (sel, "Getting caps of srcpad peer");
   caps = gst_pad_peer_get_caps_reffed (sel->srcpad);
@@ -531,6 +537,8 @@ gst_selector_pad_acceptcaps (GstPad * pad, GstCaps * caps)
   gboolean res;
 
   sel = GST_INPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL))
+    return FALSE;
 
   GST_DEBUG_OBJECT (sel, "Checking acceptcaps of srcpad peer");
   res = gst_pad_peer_accept_caps (sel->srcpad, caps);
@@ -550,6 +558,9 @@ gst_selector_pad_bufferalloc (GstPad * pad, guint64 offset,
   GstSelectorPad *selpad;
 
   sel = GST_INPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL))
+    return GST_FLOW_WRONG_STATE;
+
   selpad = GST_SELECTOR_PAD_CAST (pad);
 
   GST_LOG_OBJECT (pad, "received alloc");

plugins/elements/gstoutputselector.c

@@ -294,10 +294,13 @@ gst_output_selector_get_property (GObject * object, guint prop_id,
 static GstCaps *
 gst_output_selector_sink_getcaps (GstPad * pad)
 {
-  GstOutputSelector *sel = GST_OUTPUT_SELECTOR (GST_PAD_PARENT (pad));
+  GstOutputSelector *sel = GST_OUTPUT_SELECTOR (gst_pad_get_parent (pad));
   GstPad *active = NULL;
   GstCaps *caps = NULL;
 
+  if (G_UNLIKELY (sel == NULL))
+    return gst_caps_new_any ();
+
   GST_OBJECT_LOCK (sel);
   if (sel->pending_srcpad)
     active = gst_object_ref (sel->pending_srcpad);
@@ -312,6 +315,8 @@ gst_output_selector_sink_getcaps (GstPad * pad)
   if (caps == NULL) {
     caps = gst_caps_new_any ();
   }
+
+  gst_object_unref (sel);
   return caps;
 }
 
@@ -363,7 +368,9 @@ gst_output_selector_buffer_alloc (GstPad * pad, guint64 offset, guint size,
   GstFlowReturn res;
   GstPad *allocpad;
 
-  sel = GST_OUTPUT_SELECTOR (GST_PAD_PARENT (pad));
+  sel = GST_OUTPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL))
+    return GST_FLOW_WRONG_STATE;
   res = GST_FLOW_NOT_LINKED;
 
   GST_OBJECT_LOCK (sel);
@@ -390,6 +397,7 @@ gst_output_selector_buffer_alloc (GstPad * pad, guint64 offset, guint size,
 
   GST_DEBUG_OBJECT (sel, "buffer alloc finished: %s", gst_flow_get_name (res));
 
+  gst_object_unref (sel);
   return res;
 }
 
@@ -579,6 +587,10 @@ gst_output_selector_handle_sink_event (GstPad * pad, GstEvent * event)
   GstPad *output_pad = NULL;
 
   sel = GST_OUTPUT_SELECTOR (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (sel == NULL)) {
+    gst_event_unref (event);
+    return FALSE;
+  }
 
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_NEWSEGMENT:

plugins/elements/gstqueue.c

@@ -472,11 +472,15 @@ gst_queue_acceptcaps (GstPad * pad, GstCaps * caps)
   GstQueue *queue;
   GstPad *otherpad;
 
-  queue = GST_QUEUE (GST_PAD_PARENT (pad));
+  queue = GST_QUEUE (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (queue == NULL))
+    return FALSE;
 
   otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
   result = gst_pad_peer_accept_caps (otherpad, caps);
 
+  gst_object_unref (queue);
+
   return result;
 }
 
@@ -545,11 +549,14 @@ gst_queue_bufferalloc (GstPad * pad, guint64 offset, guint size, GstCaps * caps,
   GstQueue *queue;
   GstFlowReturn result;
 
-  queue = GST_QUEUE (GST_PAD_PARENT (pad));
+  queue = GST_QUEUE (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (queue == NULL))
+    return GST_FLOW_WRONG_STATE;
 
   /* Forward to src pad, without setting caps on the src pad */
   result = gst_pad_alloc_buffer (queue->srcpad, offset, size, caps, buf);
 
+  gst_object_unref (queue);
   return result;
 }
 
@@ -804,7 +811,11 @@ gst_queue_handle_sink_event (GstPad * pad, GstEvent * event)
 {
   GstQueue *queue;
 
-  queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
+  queue = GST_QUEUE (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (queue == NULL)) {
+    gst_event_unref (event);
+    return FALSE;
+  }
 
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_FLUSH_START:
@@ -865,6 +876,7 @@ gst_queue_handle_sink_event (GstPad * pad, GstEvent * event)
       break;
   }
 done:
+  gst_object_unref (queue);
   return TRUE;
 
   /* ERRORS */
@@ -873,6 +885,7 @@ out_flushing:
     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
         "refusing event, we are flushing");
     GST_QUEUE_MUTEX_UNLOCK (queue);
+    gst_object_unref (queue);
     gst_event_unref (event);
     return FALSE;
   }
@@ -880,6 +893,7 @@ out_eos:
   {
     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "refusing event, we are EOS");
     GST_QUEUE_MUTEX_UNLOCK (queue);
+    gst_object_unref (queue);
     gst_event_unref (event);
     return FALSE;
   }

plugins/elements/gsttee.c

@@ -551,7 +551,9 @@ gst_tee_buffer_alloc (GstPad * pad, guint64 offset, guint size,
   GstFlowReturn res;
   GstPad *allocpad;
 
-  tee = GST_TEE_CAST (GST_PAD_PARENT (pad));
+  tee = GST_TEE_CAST (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (tee == NULL))
+    return GST_FLOW_WRONG_STATE;
 
   res = GST_FLOW_NOT_LINKED;
 
@@ -587,6 +589,7 @@ gst_tee_buffer_alloc (GstPad * pad, guint64 offset, guint size,
   }
   GST_OBJECT_UNLOCK (tee);
 
+  gst_object_unref (tee);
   return res;
 }