mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-27 20:21:24 +00:00
Code Issues 1 Projects Releases Wiki Activity

gst-libs/gst/rtp/gstbasertpdepayload.*: Deprecate the queue handling thread thing and remove the code.

Browse source 
Original commit message from CVS:
* gst-libs/gst/rtp/gstbasertpdepayload.c:
(gst_base_rtp_depayload_class_init),
(gst_base_rtp_depayload_finalize),
(gst_base_rtp_depayload_setcaps), (gst_base_rtp_depayload_chain),
(gst_base_rtp_depayload_handle_sink_event),
(gst_base_rtp_depayload_set_gst_timestamp),
(gst_base_rtp_depayload_change_state):
* gst-libs/gst/rtp/gstbasertpdepayload.h:
Deprecate the queue handling thread thing and remove the code.
Use new method to calculate the extended timestamp.
This commit is contained in:
Wim Taymans 2007-08-31 15:58:30 +00:00
parent 27ea51ec37
commit 827967c8e8
3 changed files with 52 additions and 295 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
ChangeLog
View file

@ -1,3 +1,16 @@
2007-08-31 Wim Taymans <wim.taymans@gmail.com>
* gst-libs/gst/rtp/gstbasertpdepayload.c:
(gst_base_rtp_depayload_class_init),
(gst_base_rtp_depayload_finalize),
(gst_base_rtp_depayload_setcaps), (gst_base_rtp_depayload_chain),
(gst_base_rtp_depayload_handle_sink_event),
(gst_base_rtp_depayload_set_gst_timestamp),
(gst_base_rtp_depayload_change_state):
* gst-libs/gst/rtp/gstbasertpdepayload.h:
Deprecate the queue handling thread thing and remove the code.
Use new method to calculate the extended timestamp.
2007-08-31 Wim Taymans <wim.taymans@gmail.com> 2007-08-31 Wim Taymans <wim.taymans@gmail.com>
* gst-libs/gst/rtp/gstrtcpbuffer.c: * gst-libs/gst/rtp/gstrtcpbuffer.c:

323
gst-libs/gst/rtp/gstbasertpdepayload.c
View file

@ -55,8 +55,7 @@ struct _GstBaseRTPDepayloadPrivate
gdouble play_speed; gdouble play_speed;
gdouble play_scale; gdouble play_scale;
GstClockTime ts_wraparound; GstClockTime exttimestamp;
GstClockTime prev_timestamp;
}; };
/* Filter signals and args */ /* Filter signals and args */
@ -89,14 +88,8 @@ static gboolean gst_base_rtp_depayload_handle_sink_event (GstPad * pad,
static GstStateChangeReturn gst_base_rtp_depayload_change_state (GstElement * static GstStateChangeReturn gst_base_rtp_depayload_change_state (GstElement *
element, GstStateChange transition); element, GstStateChange transition);
static GstFlowReturn gst_base_rtp_depayload_add_to_queue (GstBaseRTPDepayload *
filter, GstBuffer * in);
static GstFlowReturn gst_base_rtp_depayload_process (GstBaseRTPDepayload *
filter, GstBuffer * rtp_buf);
static void gst_base_rtp_depayload_set_gst_timestamp static void gst_base_rtp_depayload_set_gst_timestamp
(GstBaseRTPDepayload * filter, guint32 timestamp, GstBuffer * buf); (GstBaseRTPDepayload * filter, guint32 timestamp, GstBuffer * buf);
static void gst_base_rtp_depayload_wait (GstBaseRTPDepayload * filter,
GstClockTime time);
GST_BOILERPLATE (GstBaseRTPDepayload, gst_base_rtp_depayload, GstElement, GST_BOILERPLATE (GstBaseRTPDepayload, gst_base_rtp_depayload, GstElement,
GST_TYPE_ELEMENT); GST_TYPE_ELEMENT);
@ -125,12 +118,11 @@ gst_base_rtp_depayload_class_init (GstBaseRTPDepayloadClass * klass)
g_object_class_install_property (gobject_class, PROP_QUEUE_DELAY, g_object_class_install_property (gobject_class, PROP_QUEUE_DELAY,
g_param_spec_uint ("queue_delay", "Queue Delay", g_param_spec_uint ("queue_delay", "Queue Delay",
"Amount of ms to queue/buffer", 0, G_MAXUINT, DEFAULT_QUEUE_DELAY, "Amount of ms to queue/buffer, deprecated", 0, G_MAXUINT,
G_PARAM_READWRITE)); DEFAULT_QUEUE_DELAY, G_PARAM_READWRITE));
gstelement_class->change_state = gst_base_rtp_depayload_change_state; gstelement_class->change_state = gst_base_rtp_depayload_change_state;
klass->add_to_queue = gst_base_rtp_depayload_add_to_queue;
klass->set_gst_timestamp = gst_base_rtp_depayload_set_gst_timestamp; klass->set_gst_timestamp = gst_base_rtp_depayload_set_gst_timestamp;
GST_DEBUG_CATEGORY_INIT (basertpdepayload_debug, "basertpdepayload", 0, GST_DEBUG_CATEGORY_INIT (basertpdepayload_debug, "basertpdepayload", 0,
@ -174,11 +166,8 @@ gst_base_rtp_depayload_init (GstBaseRTPDepayload * filter,
static void static void
gst_base_rtp_depayload_finalize (GObject * object) gst_base_rtp_depayload_finalize (GObject * object)
{ {
GstBuffer *buf;
GstBaseRTPDepayload *filter = GST_BASE_RTP_DEPAYLOAD (object); GstBaseRTPDepayload *filter = GST_BASE_RTP_DEPAYLOAD (object);
while ((buf = g_queue_pop_head (filter->queue)))
gst_buffer_unref (buf);
g_queue_free (filter->queue); g_queue_free (filter->queue);
G_OBJECT_CLASS (parent_class)->finalize (object); G_OBJECT_CLASS (parent_class)->finalize (object);
@ -235,7 +224,7 @@ gst_base_rtp_depayload_setcaps (GstPad * pad, GstCaps * caps)
else else
priv->play_scale = 1.0; priv->play_scale = 1.0;
priv->prev_timestamp = -1; priv->exttimestamp = -1;
if (bclass->set_caps) if (bclass->set_caps)
res = bclass->set_caps (filter, caps); res = bclass->set_caps (filter, caps);
@ -253,6 +242,7 @@ gst_base_rtp_depayload_chain (GstPad * pad, GstBuffer * in)
GstBaseRTPDepayload *filter; GstBaseRTPDepayload *filter;
GstBaseRTPDepayloadClass *bclass; GstBaseRTPDepayloadClass *bclass;
GstFlowReturn ret = GST_FLOW_OK; GstFlowReturn ret = GST_FLOW_OK;
GstBuffer *out_buf;
filter = GST_BASE_RTP_DEPAYLOAD (GST_OBJECT_PARENT (pad)); filter = GST_BASE_RTP_DEPAYLOAD (GST_OBJECT_PARENT (pad));
@ -261,15 +251,23 @@ gst_base_rtp_depayload_chain (GstPad * pad, GstBuffer * in)
bclass = GST_BASE_RTP_DEPAYLOAD_GET_CLASS (filter); bclass = GST_BASE_RTP_DEPAYLOAD_GET_CLASS (filter);
if (filter->queue_delay == 0) { /* let's send it out to processing */
GST_DEBUG_OBJECT (filter, "Pushing directly!"); out_buf = bclass->process (filter, in);
ret = gst_base_rtp_depayload_process (filter, in); if (out_buf) {
} else { guint32 timestamp;
if (bclass->add_to_queue)
ret = bclass->add_to_queue (filter, in); timestamp = gst_rtp_buffer_get_timestamp (in);
else
goto no_delay; /* push buffer with timestamp
* We are assuming here that the timestamp of the last RTP buffer
* is the same as the timestamp wanted on the collector. If this is not a
* desired result, the process function should push itself with another
* timestamp and return NULL.
*/
ret = gst_base_rtp_depayload_push_ts (filter, timestamp, out_buf);
} }
gst_buffer_unref (in);
return ret; return ret;
/* ERRORS */ /* ERRORS */
@ -280,13 +278,6 @@ not_configured:
gst_buffer_unref (in); gst_buffer_unref (in);
return GST_FLOW_NOT_NEGOTIATED; return GST_FLOW_NOT_NEGOTIATED;
} }
no_delay:
{
GST_ELEMENT_ERROR (filter, STREAM, NOT_IMPLEMENTED,
(NULL), ("This element cannot operate with delay"));
gst_buffer_unref (in);
return GST_FLOW_NOT_SUPPORTED;
}
} }
static gboolean static gboolean
@ -299,28 +290,15 @@ gst_base_rtp_depayload_handle_sink_event (GstPad * pad, GstEvent * event)
switch (GST_EVENT_TYPE (event)) { switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_NEWSEGMENT: case GST_EVENT_NEWSEGMENT:
{ {
/* intercept NEWSEGMENT events only if the packet scheduler thread GstFormat format;
is active */
if (filter->thread) {
GST_DEBUG_OBJECT (filter,
"Upstream sent a NEWSEGMENT, handle in worker thread.");
/* the worker thread will assign a new RTP-TS<->GST-TS mapping
* based on the next processed RTP packet */
filter->need_newsegment = TRUE;
gst_event_unref (event);
break;
} else {
GstFormat format;
gst_event_parse_new_segment (event, NULL, NULL, &format, NULL, NULL, gst_event_parse_new_segment (event, NULL, NULL, &format, NULL, NULL,
NULL); NULL);
if (format != GST_FORMAT_TIME) if (format != GST_FORMAT_TIME)
goto wrong_format; goto wrong_format;
GST_DEBUG_OBJECT (filter, GST_DEBUG_OBJECT (filter, "Upstream sent a NEWSEGMENT, passing through.");
"Upstream sent a NEWSEGMENT, passing through."); /* fallthrough */
}
/* note: pass through to default if no thread running */
} }
default: default:
/* pass other events forward */ /* pass other events forward */
@ -339,55 +317,6 @@ wrong_format:
} }
} }
static GstFlowReturn
gst_base_rtp_depayload_add_to_queue (GstBaseRTPDepayload * filter,
GstBuffer * in)
{
GQueue *queue = filter->queue;
int i;
/* our first packet, just push it */
QUEUE_LOCK (filter);
if (g_queue_is_empty (queue)) {
g_queue_push_tail (queue, in);
QUEUE_UNLOCK (filter);
} else {
guint16 seqnum, queueseq;
guint32 timestamp;
seqnum = gst_rtp_buffer_get_seq (in);
queueseq = gst_rtp_buffer_get_seq (GST_BUFFER (g_queue_peek_head (queue)));
/* look for right place to insert it */
i = 0;
/* Check for seqnum wraparound.
* Seqnums in the lowest quadrant of the 0-65535 space are considered to
* be greater than seqnums in the highest quadrant of this space. */
while (seqnum > queueseq || (seqnum < 16384 && queueseq > 49150)) {
gpointer data;
i++;
data = g_queue_peek_nth (queue, i);
if (!data)
break;
queueseq = gst_rtp_buffer_get_seq (GST_BUFFER (data));
}
/* now insert it at that place */
g_queue_push_nth (queue, in, i);
QUEUE_UNLOCK (filter);
timestamp = gst_rtp_buffer_get_timestamp (in);
GST_DEBUG_OBJECT (filter,
"Packet added to queue %d at pos %d timestamp %u sn %d",
g_queue_get_length (queue), i, timestamp, seqnum);
}
return GST_FLOW_OK;
}
static GstFlowReturn static GstFlowReturn
gst_base_rtp_depayload_push_full (GstBaseRTPDepayload * filter, gst_base_rtp_depayload_push_full (GstBaseRTPDepayload * filter,
gboolean do_ts, guint32 timestamp, GstBuffer * out_buf) gboolean do_ts, guint32 timestamp, GstBuffer * out_buf)
@ -458,41 +387,12 @@ gst_base_rtp_depayload_push (GstBaseRTPDepayload * filter, GstBuffer * out_buf)
return gst_base_rtp_depayload_push_full (filter, FALSE, 0, out_buf); return gst_base_rtp_depayload_push_full (filter, FALSE, 0, out_buf);
} }
static GstFlowReturn
gst_base_rtp_depayload_process (GstBaseRTPDepayload * filter,
GstBuffer * rtp_buf)
{
GstBaseRTPDepayloadClass *bclass;
GstBuffer *out_buf;
GstFlowReturn ret = GST_FLOW_OK;
bclass = GST_BASE_RTP_DEPAYLOAD_GET_CLASS (filter);
/* let's send it out to processing */
out_buf = bclass->process (filter, rtp_buf);
if (out_buf) {
guint32 timestamp = gst_rtp_buffer_get_timestamp (rtp_buf);
/* push buffer with timestamp
* We are assuming here that the timestamp of the last RTP buffer
* is the same as the timestamp wanted on the collector. If this is not a
* desired result, the process function should push itself with another
* timestamp and return NULL.
*/
ret = gst_base_rtp_depayload_push_ts (filter, timestamp, out_buf);
}
gst_buffer_unref (rtp_buf);
return ret;
}
static void static void
gst_base_rtp_depayload_set_gst_timestamp (GstBaseRTPDepayload * filter, gst_base_rtp_depayload_set_gst_timestamp (GstBaseRTPDepayload * filter,
guint32 timestamp, GstBuffer * buf) guint32 timestamp, GstBuffer * buf)
{ {
GstClockTime ts, adjusted, exttimestamp; GstClockTime ts, adjusted, exttimestamp;
GstBaseRTPDepayloadPrivate *priv; GstBaseRTPDepayloadPrivate *priv;
guint64 diff;
priv = filter->priv; priv = filter->priv;
@ -500,34 +400,16 @@ gst_base_rtp_depayload_set_gst_timestamp (GstBaseRTPDepayload * filter,
if (priv->clock_base == -1) if (priv->clock_base == -1)
priv->clock_base = timestamp; priv->clock_base = timestamp;
if (priv->prev_timestamp == -1) { /* get extended timestamp */
priv->prev_timestamp = timestamp; exttimestamp = gst_rtp_buffer_ext_timestamp (&priv->exttimestamp, timestamp);
priv->ts_wraparound = 0;
}
/* check for timestamp wraparound */
exttimestamp = timestamp + priv->ts_wraparound;
if (exttimestamp < priv->prev_timestamp)
diff = priv->prev_timestamp - exttimestamp;
else
diff = exttimestamp - priv->prev_timestamp;
if (diff > G_MAXINT32) {
/* timestamp went backwards more than allowed, we wrap around and get
* updated extended timestamp. */
priv->ts_wraparound += (G_GINT64_CONSTANT (1) << 32);
exttimestamp = timestamp + priv->ts_wraparound;
}
priv->prev_timestamp = exttimestamp;
/* rtp timestamps are based on the clock_rate /* rtp timestamps are based on the clock_rate
* gst timesamps are in nanoseconds */ * gst timesamps are in nanoseconds */
ts = gst_util_uint64_scale_int (exttimestamp, GST_SECOND, filter->clock_rate); ts = gst_util_uint64_scale_int (exttimestamp, GST_SECOND, filter->clock_rate);
GST_DEBUG_OBJECT (filter, GST_DEBUG_OBJECT (filter,
"timestamp: %u, wrap %" G_GUINT64_FORMAT ", clockrate : %u", timestamp, "timestamp: %u, exttimestamp %" G_GUINT64_FORMAT ", clockrate : %u",
priv->ts_wraparound, filter->clock_rate); timestamp, exttimestamp, filter->clock_rate);
/* add delay to timestamp */ /* add delay to timestamp */
adjusted = ts + (filter->queue_delay * GST_MSECOND); adjusted = ts + (filter->queue_delay * GST_MSECOND);
@ -563,125 +445,6 @@ gst_base_rtp_depayload_set_gst_timestamp (GstBaseRTPDepayload * filter,
} }
} }
static void
gst_base_rtp_depayload_queue_release (GstBaseRTPDepayload * filter)
{
GQueue *queue = filter->queue;
guint32 headts, tailts;
GstBaseRTPDepayloadClass *bclass;
gfloat q_size_secs;
guint maxtsunits;
if (g_queue_is_empty (queue))
return;
/* if our queue is getting to big (more than RTP_QUEUEDELAY ms of data)
* release heading buffers
*/
/*GST_DEBUG_OBJECT (filter, "clockrate %d, queue_delay %d", filter->clock_rate,
filter->queue_delay); */
q_size_secs = (gfloat) filter->queue_delay / 1000;
maxtsunits = (gfloat) filter->clock_rate * q_size_secs;
QUEUE_LOCK (filter);
headts =
gst_rtp_buffer_get_timestamp (GST_BUFFER (g_queue_peek_head (queue)));
tailts =
gst_rtp_buffer_get_timestamp (GST_BUFFER (g_queue_peek_tail (queue)));
bclass = GST_BASE_RTP_DEPAYLOAD_GET_CLASS (filter);
/*GST_DEBUG("maxtsunit is %u %u %u %u", maxtsunits, headts, tailts, headts - tailts); */
while (headts - tailts > maxtsunits) {
GST_DEBUG_OBJECT (filter, "Poping packet from queue");
if (bclass->process) {
GstBuffer *in = g_queue_pop_head (queue);
gst_base_rtp_depayload_process (filter, in);
}
headts =
gst_rtp_buffer_get_timestamp (GST_BUFFER (g_queue_peek_head (queue)));
}
QUEUE_UNLOCK (filter);
}
static gpointer
gst_base_rtp_depayload_thread (GstBaseRTPDepayload * filter)
{
while (filter->thread_running) {
gst_base_rtp_depayload_queue_release (filter);
/* sleep for 5msec (XXX: 5msec is a value that works for audio and video,
* should be adjusted based on frequency of incoming packet,
* or by data comsumption rate of the sink (depends on how
* clock-drift compensation is implemented) */
gst_base_rtp_depayload_wait (filter, GST_MSECOND * 5);
}
return NULL;
}
static gboolean
gst_base_rtp_depayload_start_thread (GstBaseRTPDepayload * filter)
{
/* only launch the thread if processing is needed */
if (filter->queue_delay) {
GST_DEBUG_OBJECT (filter, "Starting queue release thread");
QUEUE_LOCK_INIT (filter);
filter->thread_running = TRUE;
filter->thread =
g_thread_create ((GThreadFunc) gst_base_rtp_depayload_thread, filter,
TRUE, NULL);
GST_DEBUG_OBJECT (filter, "Started queue release thread");
}
return TRUE;
}
static gboolean
gst_base_rtp_depayload_stop_thread (GstBaseRTPDepayload * filter)
{
filter->thread_running = FALSE;
if (filter->thread) {
g_thread_join (filter->thread);
filter->thread = NULL;
}
QUEUE_LOCK_FREE (filter);
return TRUE;
}
static void
gst_base_rtp_depayload_wait (GstBaseRTPDepayload * filter, GstClockTime time)
{
GstClockID id;
GstClock *clock;
GstClockTime base;
g_return_if_fail (GST_CLOCK_TIME_IS_VALID (time));
GST_OBJECT_LOCK (filter);
if ((clock = GST_ELEMENT_CLOCK (filter)) == NULL)
goto no_clock;
gst_object_ref (clock);
GST_OBJECT_UNLOCK (filter);
base = gst_clock_get_time (clock);
id = gst_clock_new_single_shot_id (clock, base + time);
gst_object_unref (clock);
gst_clock_id_wait (id, NULL);
gst_clock_id_unref (id);
return;
no_clock:
{
GST_DEBUG_OBJECT (filter, "No clock given yet");
GST_OBJECT_UNLOCK (filter);
return;
}
}
static GstStateChangeReturn static GstStateChangeReturn
gst_base_rtp_depayload_change_state (GstElement * element, gst_base_rtp_depayload_change_state (GstElement * element,
GstStateChange transition) GstStateChange transition)
@ -691,20 +454,14 @@ gst_base_rtp_depayload_change_state (GstElement * element,
filter = GST_BASE_RTP_DEPAYLOAD (element); filter = GST_BASE_RTP_DEPAYLOAD (element);
/* we disallow changing the state from the thread */
if (g_thread_self () == filter->thread)
goto wrong_thread;
switch (transition) { switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY: case GST_STATE_CHANGE_NULL_TO_READY:
if (!gst_base_rtp_depayload_start_thread (filter))
goto start_failed;
break; break;
case GST_STATE_CHANGE_READY_TO_PAUSED: case GST_STATE_CHANGE_READY_TO_PAUSED:
/* clock_rate needs to be overwritten by child */ /* clock_rate needs to be overwritten by child */
filter->clock_rate = 0; filter->clock_rate = 0;
filter->priv->clock_base = -1; filter->priv->clock_base = -1;
filter->priv->ts_wraparound = 0; filter->priv->exttimestamp = -1;
filter->need_newsegment = TRUE; filter->need_newsegment = TRUE;
break; break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING: case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
@ -721,25 +478,11 @@ gst_base_rtp_depayload_change_state (GstElement * element,
case GST_STATE_CHANGE_PAUSED_TO_READY: case GST_STATE_CHANGE_PAUSED_TO_READY:
break; break;
case GST_STATE_CHANGE_READY_TO_NULL: case GST_STATE_CHANGE_READY_TO_NULL:
gst_base_rtp_depayload_stop_thread (filter);
break; break;
default: default:
break; break;
} }
return ret; return ret;
/* ERRORS */
wrong_thread:
{
GST_ELEMENT_ERROR (filter, CORE, STATE_CHANGE,
(NULL), ("cannot perform a state change from this thread"));
return GST_STATE_CHANGE_FAILURE;
}
start_failed:
{
/* start method should have posted an error message */
return GST_STATE_CHANGE_FAILURE;
}
} }
static void static void

11
gst-libs/gst/rtp/gstbasertpdepayload.h
View file

@ -59,23 +59,24 @@ struct _GstBaseRTPDepayload
GstPad *sinkpad, *srcpad; GstPad *sinkpad, *srcpad;
/* lock to protect the queue */ /* lock to protect the queue, deprecated */
GStaticRecMutex queuelock; GStaticRecMutex queuelock;
/* deprecated */
gboolean thread_running; gboolean thread_running;
/* the releaser thread */ /* the releaser thread, deprecated */
GThread *thread; GThread *thread;
/* this attribute must be set by the child */ /* this attribute must be set by the child */
guint clock_rate; guint clock_rate;
/* this value can be modified by the child if needed */ /* this value can be modified by the child if needed, deprecated */
guint queue_delay; guint queue_delay;
/* we will queue up to RTP_QUEUEDELAY ms of packets, /* we will queue up to RTP_QUEUEDELAY ms of packets,
* reordering them if necessary * reordering them if necessary
* dropping any packets that are more than * dropping any packets that are more than
* RTP_QUEUEDELAY ms late */ * RTP_QUEUEDELAY ms late, deprecated */
GQueue *queue; GQueue *queue;
GstSegment segment; GstSegment segment;
@ -95,7 +96,7 @@ struct _GstBaseRTPDepayloadClass
gboolean (*set_caps) (GstBaseRTPDepayload *filter, GstCaps *caps); gboolean (*set_caps) (GstBaseRTPDepayload *filter, GstCaps *caps);
/* non-pure function, default implementation in base class /* non-pure function, default implementation in base class
* this does buffering, reordering and dropping */ * this does buffering, reordering and dropping, deprecated */
GstFlowReturn (*add_to_queue) (GstBaseRTPDepayload *filter, GstBuffer *in); GstFlowReturn (*add_to_queue) (GstBaseRTPDepayload *filter, GstBuffer *in);
/* pure virtual function, child must use this to process incoming /* pure virtual function, child must use this to process incoming