mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-04-26 06:54:49 +00:00
Code Issues 1 Projects Releases Wiki Activity

rtprtxsend: retransmit packets in the same order as the rtx requests

Browse source
This commit is contained in:
George Kiagiadakis 2013-11-04 18:38:24 +02:00 committed by Wim Taymans
parent 3e818e218b
commit 487fa8c989
3 changed files with 33 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
gst/rtpmanager/gstrtprtxsend.c
View file

@ -167,9 +167,8 @@ gst_rtp_rtx_send_reset (GstRtpRtxSend * rtx, gboolean full)
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
g_queue_foreach (rtx->queue, (GFunc) buffer_queue_item_free, NULL); g_queue_foreach (rtx->queue, (GFunc) buffer_queue_item_free, NULL);
g_queue_clear (rtx->queue); g_queue_clear (rtx->queue);
g_list_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL); g_queue_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL);
g_list_free (rtx->pending); g_queue_clear (rtx->pending);
rtx->pending = NULL;
rtx->master_ssrc = 0; rtx->master_ssrc = 0;
rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16); rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16);
rtx->rtx_ssrc = g_random_int (); rtx->rtx_ssrc = g_random_int ();
@ -185,6 +184,7 @@ gst_rtp_rtx_send_finalize (GObject * object)
gst_rtp_rtx_send_reset (rtx, TRUE); gst_rtp_rtx_send_reset (rtx, TRUE);
g_queue_free (rtx->queue); g_queue_free (rtx->queue);
g_queue_free (rtx->pending);
g_mutex_clear (&rtx->lock); g_mutex_clear (&rtx->lock);
G_OBJECT_CLASS (gst_rtp_rtx_send_parent_class)->finalize (object); G_OBJECT_CLASS (gst_rtp_rtx_send_parent_class)->finalize (object);
@ -216,7 +216,7 @@ gst_rtp_rtx_send_init (GstRtpRtxSend * rtx)
gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad); gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad);
rtx->queue = g_queue_new (); rtx->queue = g_queue_new ();
rtx->pending = NULL; rtx->pending = g_queue_new ();
g_mutex_init (&rtx->lock); g_mutex_init (&rtx->lock);
rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16); rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16);
@ -262,7 +262,7 @@ push_seqnum (BufferQueueItem * item, RTXData * data)
if (item->seqnum == data->seqnum) { if (item->seqnum == data->seqnum) {
data->found = TRUE; data->found = TRUE;
GST_DEBUG_OBJECT (rtx, "found %" G_GUINT16_FORMAT, item->seqnum); GST_DEBUG_OBJECT (rtx, "found %" G_GUINT16_FORMAT, item->seqnum);
rtx->pending = g_list_prepend (rtx->pending, gst_buffer_ref (item->buffer)); g_queue_push_tail (rtx->pending, gst_buffer_ref (item->buffer));
} }
} }
@ -330,9 +330,8 @@ gst_rtp_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
g_queue_clear (rtx->queue); g_queue_clear (rtx->queue);
/* clear buffers that are about to be retransmited */ /* clear buffers that are about to be retransmited */
g_list_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL); g_queue_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL);
g_list_free (rtx->pending); g_queue_clear (rtx->pending);
rtx->pending = NULL;
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
@ -475,7 +474,7 @@ _gst_rtp_rtx_buffer_new (GstBuffer * buffer, guint32 ssrc, guint16 seqnum,
return new_buffer; return new_buffer;
} }
/* psuh pending retransmission packet. /* push pending retransmission packet.
* it constructs rtx packet from original paclets */ * it constructs rtx packet from original paclets */
static void static void
do_push (GstBuffer * buffer, GstRtpRtxSend * rtx) do_push (GstBuffer * buffer, GstRtpRtxSend * rtx)
@ -494,7 +493,7 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{ {
GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (parent); GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (parent);
GstFlowReturn ret = GST_FLOW_ERROR; GstFlowReturn ret = GST_FLOW_ERROR;
GList *pending = NULL; GQueue *pending = NULL;
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
BufferQueueItem *item; BufferQueueItem *item;
guint16 seqnum; guint16 seqnum;
@ -535,11 +534,13 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
} }
/* within lock, get packets that have to be retransmited */ /* within lock, get packets that have to be retransmited */
pending = rtx->pending; if (g_queue_get_length (rtx->pending) > 0) {
rtx->pending = NULL; pending = rtx->pending;
rtx->pending = g_queue_new ();
/* update statistics - assume we will succeed to retransmit those packets */ /* update statistics - assume we will succeed to retransmit those packets */
rtx->num_rtx_packets += g_list_length (pending); rtx->num_rtx_packets += g_queue_get_length (pending);
}
/* transfer payload type while holding the lock */ /* transfer payload type while holding the lock */
rtx->rtx_payload_type = rtx->rtx_payload_type_pending; rtx->rtx_payload_type = rtx->rtx_payload_type_pending;
@ -548,9 +549,10 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
/* retransmit requested packets */ /* retransmit requested packets */
g_list_foreach (pending, (GFunc) do_push, rtx); if (pending) {
g_list_foreach (pending, (GFunc) gst_buffer_unref, NULL); g_queue_foreach (pending, (GFunc) do_push, rtx);
g_list_free (pending); g_queue_free_full (pending, (GDestroyNotify) gst_buffer_unref);
}
GST_LOG_OBJECT (rtx, GST_LOG_OBJECT (rtx,
"push seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT, seqnum, "push seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT, seqnum,

2
gst/rtpmanager/gstrtprtxsend.h
View file

@ -50,7 +50,7 @@ struct _GstRtpRtxSend
/* history of rtp packets */ /* history of rtp packets */
GQueue *queue; GQueue *queue;
/* rtp packets that will be pushed upon next buffer */ /* rtp packets that will be pushed upon next buffer */
GList *pending; GQueue *pending;
guint32 master_ssrc; guint32 master_ssrc;
guint32 rtx_ssrc; guint32 rtx_ssrc;

30
tests/check/elements/rtprtx.c
View file

@ -1235,7 +1235,7 @@ test_rtxsender_packet_retention (gboolean test_with_time)
} }
/* verify the result. buffers should be in this order (numbers are seqnums): /* verify the result. buffers should be in this order (numbers are seqnums):
* 1, 1rtx, 2, 2rtx, 1rtx, 3, ... , 9, 9rtx, 8rtx, 7rtx, 6rtx, 5rtx, 10 */ * 1, 1rtx, 2, 1rtx, 2rtx, 3, ... , 9, 5rtx, 6rtx, 7rtx, 8rtx, 9rtx, 10 */
{ {
GstRTPBuffer orig_rtp = GST_RTP_BUFFER_INIT; GstRTPBuffer orig_rtp = GST_RTP_BUFFER_INIT;
gint expected_rtx_requests, expected_rtx_packets; gint expected_rtx_requests, expected_rtx_packets;
@ -1258,22 +1258,8 @@ test_rtxsender_packet_retention (gboolean test_with_time)
node = buffers; node = buffers;
for (i = 1; i <= num_buffers; i++) { for (i = 1; i <= num_buffers; i++) {
/* verify the normal rtp flow packet */ /* verify the retransmission packets */
res = gst_rtp_buffer_map (GST_BUFFER (node->data), GST_MAP_READ, &rtp); for (j = MAX (i - half_buffers, 1); j < i; j++) {
fail_unless_equals_int (res, TRUE);
fail_unless_equals_int (gst_rtp_buffer_get_ssrc (&rtp), ssrc);
fail_unless_equals_int (gst_rtp_buffer_get_payload_type (&rtp),
payload_type);
fail_unless_equals_int (gst_rtp_buffer_get_seq (&rtp), i);
gst_rtp_buffer_unmap (&rtp);
node = g_list_next (node);
/* there are no rtx packets after the last normal one */
if (i == num_buffers)
break;
/* now verify the retransmission packets */
for (j = i; j > MAX (i - half_buffers, 0); j--) {
GST_INFO ("checking %d, %d", i, j); GST_INFO ("checking %d, %d", i, j);
res = gst_rtp_buffer_map (GST_BUFFER (node->data), GST_MAP_READ, &rtp); res = gst_rtp_buffer_map (GST_BUFFER (node->data), GST_MAP_READ, &rtp);
@ -1295,6 +1281,16 @@ test_rtxsender_packet_retention (gboolean test_with_time)
gst_rtp_buffer_unmap (&rtp); gst_rtp_buffer_unmap (&rtp);
node = g_list_next (node); node = g_list_next (node);
} }
/* verify the normal rtp flow packet */
res = gst_rtp_buffer_map (GST_BUFFER (node->data), GST_MAP_READ, &rtp);
fail_unless_equals_int (res, TRUE);
fail_unless_equals_int (gst_rtp_buffer_get_ssrc (&rtp), ssrc);
fail_unless_equals_int (gst_rtp_buffer_get_payload_type (&rtp),
payload_type);
fail_unless_equals_int (gst_rtp_buffer_get_seq (&rtp), i);
gst_rtp_buffer_unmap (&rtp);
node = g_list_next (node);
} }
} }