• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 /* GStreamer
2  * Copyright (C) 2013 Rdio <ingestions@rdio.com>
3  * Copyright (C) 2013 David Schleef <ds@schleef.org>
4  *
5  * This library is free software; you can redistribute it and/or
6  * modify it under the terms of the GNU Library General Public
7  * License as published by the Free Software Foundation; either
8  * version 2 of the License, or (at your option) any later version.
9  *
10  * This library is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * Library General Public License for more details.
14  *
15  * You should have received a copy of the GNU Library General Public
16  * License along with this library; if not, write to the
17  * Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
18  * Boston, MA 02110-1335, USA.
19  */
20 /**
21  * SECTION:element-watchdog
22  * @title: watchdog
23  *
24  * The watchdog element watches buffers and events flowing through
25  * a pipeline.  If no buffers are seen for a configurable amount of
26  * time, a error message is sent to the bus.
27  *
28  * To use this element, insert it into a pipeline as you would an
29  * identity element.  Once activated, any pause in the flow of
30  * buffers through the element will cause an element error.  The
31  * maximum allowed pause is determined by the timeout property.
32  *
33  * This element is currently intended for transcoding pipelines,
34  * although may be useful in other contexts.
35  *
36  * ## Example launch line
37  * |[
38  * gst-launch-1.0 -v fakesrc ! watchdog ! fakesink
39  * ]|
40  *
41  */
42 
43 #ifdef HAVE_CONFIG_H
44 #include "config.h"
45 #endif
46 
47 #include <gst/gst.h>
48 #include <gst/base/gstbasetransform.h>
49 #include "gstdebugutilsbadelements.h"
50 #include "gstwatchdog.h"
51 
52 GST_DEBUG_CATEGORY_STATIC (gst_watchdog_debug_category);
53 #define GST_CAT_DEFAULT gst_watchdog_debug_category
54 
55 /* prototypes */
56 
57 static void gst_watchdog_set_property (GObject * object,
58     guint property_id, const GValue * value, GParamSpec * pspec);
59 static void gst_watchdog_get_property (GObject * object,
60     guint property_id, GValue * value, GParamSpec * pspec);
61 
62 static gboolean gst_watchdog_start (GstBaseTransform * trans);
63 static gboolean gst_watchdog_stop (GstBaseTransform * trans);
64 static gboolean gst_watchdog_sink_event (GstBaseTransform * trans,
65     GstEvent * event);
66 static gboolean gst_watchdog_src_event (GstBaseTransform * trans,
67     GstEvent * event);
68 static GstFlowReturn gst_watchdog_transform_ip (GstBaseTransform * trans,
69     GstBuffer * buf);
70 static void gst_watchdog_feed (GstWatchdog * watchdog, gpointer mini_object,
71     gboolean force);
72 
73 static GstStateChangeReturn
74 gst_watchdog_change_state (GstElement * element, GstStateChange transition);
75 
76 enum
77 {
78   PROP_0,
79   PROP_TIMEOUT
80 };
81 
82 /* class initialization */
83 
84 G_DEFINE_TYPE_WITH_CODE (GstWatchdog, gst_watchdog, GST_TYPE_BASE_TRANSFORM,
85     GST_DEBUG_CATEGORY_INIT (gst_watchdog_debug_category, "watchdog", 0,
86         "debug category for watchdog element"));
87 GST_ELEMENT_REGISTER_DEFINE (watchdog, "watchdog", GST_RANK_NONE,
88     gst_watchdog_get_type ());
89 
90 static void
gst_watchdog_class_init(GstWatchdogClass * klass)91 gst_watchdog_class_init (GstWatchdogClass * klass)
92 {
93   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
94   GstBaseTransformClass *base_transform_class =
95       GST_BASE_TRANSFORM_CLASS (klass);
96 
97   GstElementClass *gstelement_klass = (GstElementClass *) klass;
98 
99   gst_element_class_add_pad_template (GST_ELEMENT_CLASS (klass),
100       gst_pad_template_new ("src", GST_PAD_SRC, GST_PAD_ALWAYS,
101           gst_caps_new_any ()));
102   gst_element_class_add_pad_template (GST_ELEMENT_CLASS (klass),
103       gst_pad_template_new ("sink", GST_PAD_SINK, GST_PAD_ALWAYS,
104           gst_caps_new_any ()));
105 
106   gst_element_class_set_static_metadata (GST_ELEMENT_CLASS (klass),
107       "Watchdog", "Generic", "Watches for pauses in stream buffers",
108       "David Schleef <ds@schleef.org>");
109 
110   gstelement_klass->change_state =
111       GST_DEBUG_FUNCPTR (gst_watchdog_change_state);
112   gobject_class->set_property = gst_watchdog_set_property;
113   gobject_class->get_property = gst_watchdog_get_property;
114   base_transform_class->start = GST_DEBUG_FUNCPTR (gst_watchdog_start);
115   base_transform_class->stop = GST_DEBUG_FUNCPTR (gst_watchdog_stop);
116   base_transform_class->sink_event =
117       GST_DEBUG_FUNCPTR (gst_watchdog_sink_event);
118   base_transform_class->src_event = GST_DEBUG_FUNCPTR (gst_watchdog_src_event);
119   base_transform_class->transform_ip =
120       GST_DEBUG_FUNCPTR (gst_watchdog_transform_ip);
121 
122   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
123       g_param_spec_int ("timeout", "Timeout", "Timeout (in ms) after "
124           "which an element error is sent to the bus if no buffers are "
125           "received. 0 means disabled.", 0, G_MAXINT, 1000,
126           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
127 
128 }
129 
130 static void
gst_watchdog_init(GstWatchdog * watchdog)131 gst_watchdog_init (GstWatchdog * watchdog)
132 {
133 }
134 
135 static void
gst_watchdog_set_property(GObject * object,guint property_id,const GValue * value,GParamSpec * pspec)136 gst_watchdog_set_property (GObject * object, guint property_id,
137     const GValue * value, GParamSpec * pspec)
138 {
139   GstWatchdog *watchdog = GST_WATCHDOG (object);
140 
141   GST_DEBUG_OBJECT (watchdog, "set_property");
142 
143   switch (property_id) {
144     case PROP_TIMEOUT:
145       GST_OBJECT_LOCK (watchdog);
146       watchdog->timeout = g_value_get_int (value);
147       gst_watchdog_feed (watchdog, NULL, FALSE);
148       GST_OBJECT_UNLOCK (watchdog);
149       break;
150     default:
151       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
152       break;
153   }
154 }
155 
156 static void
gst_watchdog_get_property(GObject * object,guint property_id,GValue * value,GParamSpec * pspec)157 gst_watchdog_get_property (GObject * object, guint property_id,
158     GValue * value, GParamSpec * pspec)
159 {
160   GstWatchdog *watchdog = GST_WATCHDOG (object);
161 
162   GST_DEBUG_OBJECT (watchdog, "get_property");
163 
164   switch (property_id) {
165     case PROP_TIMEOUT:
166       g_value_set_int (value, watchdog->timeout);
167       break;
168     default:
169       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
170       break;
171   }
172 }
173 
174 static gpointer
gst_watchdog_thread(gpointer user_data)175 gst_watchdog_thread (gpointer user_data)
176 {
177   GstWatchdog *watchdog = GST_WATCHDOG (user_data);
178 
179   GST_DEBUG_OBJECT (watchdog, "thread starting");
180 
181   g_main_loop_run (watchdog->main_loop);
182 
183   GST_DEBUG_OBJECT (watchdog, "thread exiting");
184 
185   return NULL;
186 }
187 
188 static gboolean
gst_watchdog_trigger(gpointer ptr)189 gst_watchdog_trigger (gpointer ptr)
190 {
191   GstWatchdog *watchdog = GST_WATCHDOG (ptr);
192 
193   GST_DEBUG_OBJECT (watchdog, "watchdog triggered");
194 
195   GST_ELEMENT_ERROR (watchdog, STREAM, FAILED, ("Watchdog triggered"),
196       ("Watchdog triggered"));
197 
198   return FALSE;
199 }
200 
201 static gboolean
gst_watchdog_quit_mainloop(gpointer ptr)202 gst_watchdog_quit_mainloop (gpointer ptr)
203 {
204   GstWatchdog *watchdog = GST_WATCHDOG (ptr);
205 
206   GST_DEBUG_OBJECT (watchdog, "watchdog quit");
207 
208   g_main_loop_quit (watchdog->main_loop);
209 
210   return FALSE;
211 }
212 
213 /*  Call with OBJECT_LOCK taken */
214 static void
gst_watchdog_feed(GstWatchdog * watchdog,gpointer mini_object,gboolean force)215 gst_watchdog_feed (GstWatchdog * watchdog, gpointer mini_object, gboolean force)
216 {
217   if (watchdog->source) {
218     if (watchdog->waiting_for_flush_start) {
219       if (mini_object && GST_IS_EVENT (mini_object) &&
220           GST_EVENT_TYPE (mini_object) == GST_EVENT_FLUSH_START) {
221         watchdog->waiting_for_flush_start = FALSE;
222         watchdog->waiting_for_flush_stop = TRUE;
223       }
224 
225       force = TRUE;
226     } else if (watchdog->waiting_for_flush_stop) {
227       if (mini_object && GST_IS_EVENT (mini_object) &&
228           GST_EVENT_TYPE (mini_object) == GST_EVENT_FLUSH_STOP) {
229         watchdog->waiting_for_flush_stop = FALSE;
230         watchdog->waiting_for_a_buffer = TRUE;
231       }
232 
233       force = TRUE;
234     } else if (watchdog->waiting_for_a_buffer) {
235       if (mini_object && GST_IS_BUFFER (mini_object)) {
236         watchdog->waiting_for_a_buffer = FALSE;
237         GST_DEBUG_OBJECT (watchdog, "Got a buffer \\o/");
238       } else {
239         GST_DEBUG_OBJECT (watchdog, "Waiting for a buffer and did not get it,"
240             " keep trying even in PAUSED state");
241         force = TRUE;
242       }
243     }
244     g_source_destroy (watchdog->source);
245     g_source_unref (watchdog->source);
246     watchdog->source = NULL;
247 
248   }
249 
250   if (watchdog->timeout == 0) {
251     GST_LOG_OBJECT (watchdog, "Timeout is 0 => nothing to do");
252   } else if (watchdog->main_context == NULL) {
253     GST_LOG_OBJECT (watchdog, "No maincontext => nothing to do");
254   } else if ((GST_STATE (watchdog) != GST_STATE_PLAYING) && force == FALSE) {
255     GST_LOG_OBJECT (watchdog,
256         "Not in playing and force is FALSE => Nothing to do");
257   } else {
258     watchdog->source = g_timeout_source_new (watchdog->timeout);
259     g_source_set_callback (watchdog->source, gst_watchdog_trigger,
260         gst_object_ref (watchdog), gst_object_unref);
261     g_source_attach (watchdog->source, watchdog->main_context);
262   }
263 }
264 
265 static gboolean
gst_watchdog_start(GstBaseTransform * trans)266 gst_watchdog_start (GstBaseTransform * trans)
267 {
268   GstWatchdog *watchdog = GST_WATCHDOG (trans);
269 
270   GST_DEBUG_OBJECT (watchdog, "start");
271   GST_OBJECT_LOCK (watchdog);
272 
273   watchdog->main_context = g_main_context_new ();
274   watchdog->main_loop = g_main_loop_new (watchdog->main_context, TRUE);
275   watchdog->thread = g_thread_new ("watchdog", gst_watchdog_thread, watchdog);
276 
277   GST_OBJECT_UNLOCK (watchdog);
278   return TRUE;
279 }
280 
281 static gboolean
gst_watchdog_stop(GstBaseTransform * trans)282 gst_watchdog_stop (GstBaseTransform * trans)
283 {
284   GstWatchdog *watchdog = GST_WATCHDOG (trans);
285   GSource *quit_source;
286 
287   GST_DEBUG_OBJECT (watchdog, "stop");
288   GST_OBJECT_LOCK (watchdog);
289 
290   if (watchdog->source) {
291     g_source_destroy (watchdog->source);
292     g_source_unref (watchdog->source);
293     watchdog->source = NULL;
294   }
295 
296   /* dispatch an idle event that trigger g_main_loop_quit to avoid race
297    * between g_main_loop_run and g_main_loop_quit */
298   quit_source = g_idle_source_new ();
299   g_source_set_callback (quit_source, gst_watchdog_quit_mainloop, watchdog,
300       NULL);
301   g_source_attach (quit_source, watchdog->main_context);
302   g_source_unref (quit_source);
303 
304   g_thread_join (watchdog->thread);
305   watchdog->thread = NULL;
306 
307   g_main_loop_unref (watchdog->main_loop);
308   watchdog->main_loop = NULL;
309 
310   g_main_context_unref (watchdog->main_context);
311   watchdog->main_context = NULL;
312 
313   GST_OBJECT_UNLOCK (watchdog);
314   return TRUE;
315 }
316 
317 static gboolean
gst_watchdog_sink_event(GstBaseTransform * trans,GstEvent * event)318 gst_watchdog_sink_event (GstBaseTransform * trans, GstEvent * event)
319 {
320   GstWatchdog *watchdog = GST_WATCHDOG (trans);
321 
322   GST_DEBUG_OBJECT (watchdog, "sink_event");
323 
324   GST_OBJECT_LOCK (watchdog);
325   gst_watchdog_feed (watchdog, event, FALSE);
326   GST_OBJECT_UNLOCK (watchdog);
327 
328   return
329       GST_BASE_TRANSFORM_CLASS (gst_watchdog_parent_class)->sink_event (trans,
330       event);
331 }
332 
333 static gboolean
gst_watchdog_src_event(GstBaseTransform * trans,GstEvent * event)334 gst_watchdog_src_event (GstBaseTransform * trans, GstEvent * event)
335 {
336   gboolean force = FALSE;
337   GstWatchdog *watchdog = GST_WATCHDOG (trans);
338 
339   GST_DEBUG_OBJECT (watchdog, "src_event");
340 
341   GST_OBJECT_LOCK (watchdog);
342   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK) {
343     GstSeekFlags flags;
344 
345     gst_event_parse_seek (event, NULL, NULL, &flags, NULL, NULL, NULL, NULL);
346 
347     if (flags & GST_SEEK_FLAG_FLUSH) {
348       force = TRUE;
349       GST_DEBUG_OBJECT (watchdog, "Got a FLUSHING seek, we need a buffer now!");
350       watchdog->waiting_for_flush_start = TRUE;
351     }
352   }
353 
354   gst_watchdog_feed (watchdog, event, force);
355   GST_OBJECT_UNLOCK (watchdog);
356 
357   return GST_BASE_TRANSFORM_CLASS (gst_watchdog_parent_class)->src_event (trans,
358       event);
359 }
360 
361 static GstFlowReturn
gst_watchdog_transform_ip(GstBaseTransform * trans,GstBuffer * buf)362 gst_watchdog_transform_ip (GstBaseTransform * trans, GstBuffer * buf)
363 {
364   GstWatchdog *watchdog = GST_WATCHDOG (trans);
365 
366   GST_DEBUG_OBJECT (watchdog, "transform_ip");
367 
368   GST_OBJECT_LOCK (watchdog);
369   gst_watchdog_feed (watchdog, buf, FALSE);
370   GST_OBJECT_UNLOCK (watchdog);
371 
372   return GST_FLOW_OK;
373 }
374 
375 /*
376  * Change state handler for the element.
377  */
378 static GstStateChangeReturn
gst_watchdog_change_state(GstElement * element,GstStateChange transition)379 gst_watchdog_change_state (GstElement * element, GstStateChange transition)
380 {
381   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
382   GstWatchdog *watchdog = GST_WATCHDOG (element);
383 
384   GST_DEBUG_OBJECT (watchdog, "gst_watchdog_change_state");
385 
386   switch (transition) {
387     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
388       /* Activate timer */
389       GST_OBJECT_LOCK (watchdog);
390       gst_watchdog_feed (watchdog, NULL, FALSE);
391       GST_OBJECT_UNLOCK (watchdog);
392       break;
393     default:
394       break;
395   }
396 
397   ret =
398       GST_ELEMENT_CLASS (gst_watchdog_parent_class)->change_state (element,
399       transition);
400 
401   switch (transition) {
402     case GST_STATE_CHANGE_READY_TO_PAUSED:
403       GST_OBJECT_LOCK (watchdog);
404       watchdog->waiting_for_a_buffer = TRUE;
405       gst_watchdog_feed (watchdog, NULL, TRUE);
406       GST_OBJECT_UNLOCK (watchdog);
407       break;
408     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
409       /* Disable the timer */
410       GST_OBJECT_LOCK (watchdog);
411       if (watchdog->source) {
412         g_source_destroy (watchdog->source);
413         g_source_unref (watchdog->source);
414         watchdog->source = NULL;
415       }
416       GST_OBJECT_UNLOCK (watchdog);
417       break;
418     default:
419       break;
420   }
421 
422   return ret;
423 }
424