]> git.sesse.net Git - vlc/blobdiff - modules/access/udp.c
modifications for RTP reordering:
[vlc] / modules / access / udp.c
index 399cea0527356338a78f9be6d8054e49b02585d4..4189a3454180b30c40f4aa9bece1a187ac51e681 100644 (file)
@@ -1,12 +1,13 @@
 /*****************************************************************************
  * udp.c: raw UDP & RTP input module
  *****************************************************************************
- * Copyright (C) 2001-2004 VideoLAN
+ * Copyright (C) 2001-2005 the VideoLAN team
  * $Id$
  *
  * Authors: Christophe Massiot <massiot@via.ecp.fr>
  *          Tristan Leteurtre <tooney@via.ecp.fr>
  *          Laurent Aimar <fenrir@via.ecp.fr>
+ *          Jean-Paul Saman <jpsaman #_at_# m2x dot nl>
  *
  * Reviewed: 23 October 2003, Jean-Paul Saman <jpsaman@wxs.nl>
  *
 #define AUTO_MTU_LONGTEXT N_( \
     "Allows growing the MTU if truncated packets are found" )
 
+#define RTP_LATE_TEXT N_("RTP reordering timeout in ms")
+#define RTP_LATE_LONGTEXT N_( \
+    "Allows you to modify the RTP reordering behaviour. " \
+    "RTP input will wait for late packets upto " \
+    "the specified timeout in milisecond units." )
+
 static int  Open ( vlc_object_t * );
 static void Close( vlc_object_t * );
 
@@ -58,6 +65,8 @@ vlc_module_begin();
 
     add_integer( "udp-caching", DEFAULT_PTS_DELAY / 1000, NULL, CACHING_TEXT,
                  CACHING_LONGTEXT, VLC_TRUE );
+    add_integer( "rtp-late", 100, NULL, RTP_LATE_TEXT, RTP_LATE_LONGTEXT, VLC_TRUE );
+
     add_bool( "udp-auto-mtu", 1, NULL,
               AUTO_MTU_TEXT, AUTO_MTU_LONGTEXT, VLC_TRUE );
 
@@ -88,6 +97,12 @@ struct access_sys_t
 
     int i_mtu;
     vlc_bool_t b_auto_mtu;
+
+    /* reorder rtp packets when out-of-sequence */
+    int64_t i_rtp_late;
+    uint16_t i_last_seqno;
+    block_t *p_list;
+    block_t *p_end;
 };
 
 /*****************************************************************************
@@ -99,14 +114,8 @@ static int Open( vlc_object_t *p_this )
     access_sys_t *p_sys;
 
     char *psz_name = strdup( p_access->psz_path );
-    char *psz_parser = psz_name;
-    char *psz_server_addr = "";
-    char *psz_server_port = "";
-    char *psz_bind_addr = "";
-    char *psz_bind_port = "";
-    int  i_bind_port = 0;
-    int  i_server_port = 0;
-
+    char *psz_parser, *psz_server_addr, *psz_bind_addr = "";
+    int  i_bind_port, i_server_port = 0;
 
     /* First set ipv4/ipv6 */
     var_Create( p_access, "ipv4", VLC_VAR_BOOL | VLC_VAR_DOINHERIT );
@@ -136,77 +145,47 @@ static int Open( vlc_object_t *p_this )
         }
     }
 
+    i_bind_port = var_CreateGetInteger( p_access, "server-port" );
+
     /* Parse psz_name syntax :
      * [serveraddr[:serverport]][@[bindaddr]:[bindport]] */
-    if( *psz_parser && *psz_parser != '@' )
+    psz_parser = strchr( psz_name, '@' );
+    if( psz_parser != NULL )
     {
-        /* Found server */
-        psz_server_addr = psz_parser;
+        /* Found bind address and/or bind port */
+        *psz_parser++ = '\0';
+        psz_bind_addr = psz_parser;
 
-        while( *psz_parser && *psz_parser != ':' && *psz_parser != '@' )
-        {
-            if( *psz_parser == '[' )
-            {
-                /* IPv6 address */
-                while( *psz_parser && *psz_parser != ']' )
-                {
-                    psz_parser++;
-                }
-            }
-            psz_parser++;
-        }
+        if( *psz_parser == '[' )
+            /* skips bracket'd IPv6 address */
+            psz_parser = strchr( psz_parser, ']' );
 
-        if( *psz_parser == ':' )
+        if( psz_parser != NULL )
         {
-            /* Found server port */
-            *psz_parser++ = '\0'; /* Terminate server name */
-            psz_server_port = psz_parser;
-
-            while( *psz_parser && *psz_parser != '@' )
+            psz_parser = strchr( psz_parser, ':' );
+            if( psz_parser != NULL )
             {
-                psz_parser++;
+                *psz_parser++ = '\0';
+                i_bind_port = atoi( psz_parser );
             }
         }
     }
 
-    if( *psz_parser == '@' )
-    {
-        /* Found bind address or bind port */
-        *psz_parser++ = '\0'; /* Terminate server port or name if necessary */
+    psz_server_addr = psz_name;
+    if( *psz_server_addr == '[' )
+        /* skips bracket'd IPv6 address */
+        psz_parser = strchr( psz_name, ']' );
 
-        if( *psz_parser && *psz_parser != ':' )
-        {
-            /* Found bind address */
-            psz_bind_addr = psz_parser;
-
-            while( *psz_parser && *psz_parser != ':' )
-            {
-                if( *psz_parser == '[' )
-                {
-                    /* IPv6 address */
-                    while( *psz_parser && *psz_parser != ']' )
-                    {
-                        psz_parser++;
-                    }
-                }
-                psz_parser++;
-            }
-        }
-
-        if( *psz_parser == ':' )
+    if( psz_parser != NULL )
+    {
+        psz_parser = strchr( psz_parser, ':' );
+        if( psz_parser != NULL )
         {
-            /* Found bind port */
-            *psz_parser++ = '\0'; /* Terminate bind address if necessary */
-            psz_bind_port = psz_parser;
+            *psz_parser++ = '\0';
+            i_server_port = atoi( psz_parser );
         }
     }
 
-    i_server_port = strtol( psz_server_port, NULL, 10 );
-    if( ( i_bind_port   = strtol( psz_bind_port,   NULL, 10 ) ) == 0 )
-    {
-        i_bind_port = var_CreateGetInteger( p_access, "server-port" );
-    }
-
     msg_Dbg( p_access, "opening server=%s:%d local=%s:%d",
              psz_server_addr, i_server_port, psz_bind_addr, i_bind_port );
 
@@ -228,6 +207,7 @@ static int Open( vlc_object_t *p_this )
     p_access->info.i_size = 0;
     p_access->info.i_pos = 0;
     p_access->info.b_eof = VLC_FALSE;
+    p_access->info.b_prebuffered = VLC_FALSE;
     p_access->info.i_title = 0;
     p_access->info.i_seekpoint = 0;
 
@@ -243,6 +223,8 @@ static int Open( vlc_object_t *p_this )
     }
     free( psz_name );
 
+    net_StopSend( p_sys->fd );
+
     /* FIXME */
     p_sys->i_mtu = var_CreateGetInteger( p_access, "mtu" );
     if( p_sys->i_mtu <= 1 )
@@ -253,6 +235,12 @@ static int Open( vlc_object_t *p_this )
     /* Update default_pts to a suitable value for udp access */
     var_Create( p_access, "udp-caching", VLC_VAR_INTEGER | VLC_VAR_DOINHERIT );
 
+
+    /* RTP reordering for out-of-sequence packets */
+    p_sys->i_rtp_late = var_CreateGetInteger( p_access, "rtp-late" ) * 1000;
+    p_sys->i_last_seqno = 0;
+    p_sys->p_list = NULL;
+    p_sys->p_end = NULL;
     return VLC_SUCCESS;
 }
 
@@ -264,6 +252,7 @@ static void Close( vlc_object_t *p_this )
     access_t     *p_access = (access_t*)p_this;
     access_sys_t *p_sys = p_access->p_sys;
 
+    block_ChainRelease( p_sys->p_list );
     net_Close( p_sys->fd );
     free( p_sys );
 }
@@ -334,7 +323,7 @@ static block_t *BlockUDP( access_t *p_access )
         return NULL;
     }
 
-    if( p_block->i_buffer >= p_sys->i_mtu && p_sys->b_auto_mtu &&
+    if( (p_block->i_buffer >= p_sys->i_mtu) && p_sys->b_auto_mtu &&
         p_sys->i_mtu < 32767 )
     {
         /* Increase by 100% */
@@ -345,59 +334,215 @@ static block_t *BlockUDP( access_t *p_access )
     return p_block;
 }
 
+/*
+ * rtp_ChainInsert - insert a p_block in the chain and
+ * look at the sequence numbers.
+ */
+static inline vlc_bool_t rtp_ChainInsert( access_t *p_access, block_t *p_block )
+{
+    access_sys_t *p_sys = (access_sys_t *) p_access->p_sys;
+    block_t *p_prev = NULL;
+    block_t *p = p_sys->p_end;
+    uint16_t i_new = p_block->i_seqno;
+    uint16_t i_tmp = 0;
+
+    if( !p_sys->p_list )
+    {
+        p_sys->p_list = p_block;
+        p_sys->p_end = p_block;
+        return VLC_TRUE;
+    }
+    /* walk through the queue from top down since the new packet is in 
+    most cases just appended to the end */
+
+    for( ;; )
+    {
+        i_tmp = i_new - p->i_seqno;
+
+        if( !i_tmp )   /* trash duplicate */
+            break; 
+
+        if ( i_tmp < 32768 )
+        {   /* insert after this block ( i_new > p->i_seqno ) */
+            p_block->p_next = p->p_next;
+            p->p_next = p_block;
+            p_block->p_prev = p;
+            if (p_prev)
+            {
+                p_prev->p_prev = p_block;
+                msg_Dbg(p_access, "RTP reordering: insert after %d, new %d", 
+                    p->i_seqno, i_new );
+            }
+            else 
+            {
+                p_sys->p_end = p_block;
+            }
+            return VLC_TRUE;
+        }
+        if( p == p_sys->p_list )
+        {   /* we've reached bottom of chain */
+            i_tmp = p_sys->i_last_seqno - i_new;
+            if( !p_access->info.b_prebuffered || (i_tmp > 32767) )
+            {
+                msg_Dbg(p_access, "RTP reordering: prepend %d before %d", 
+                        i_new, p->i_seqno );
+                p_block->p_next = p;
+                p->p_prev = p_block;
+                p_sys->p_list = p_block;
+                return VLC_TRUE;
+            }
+
+            if( !i_tmp )   /* trash duplicate */
+                break;    
+
+            /* reordering failed - append the packet to the end of queue */
+            msg_Dbg(p_access, "RTP: sequence changed (or buffer too small) "
+                "new: %d, buffer %d...%d", i_new, p->i_seqno, 
+                p_sys->p_end->i_seqno);
+            p_sys->p_end->p_next = p_block;
+            p_block->p_prev = p_sys->p_end;
+            p_sys->p_end = p_block;
+            return VLC_TRUE;
+        }
+        p_prev = p;
+        p = p->p_prev;
+    }
+    block_Release( p_block );
+    return VLC_FALSE;
+}
+
 /*****************************************************************************
  * BlockParseRTP/BlockRTP:
  *****************************************************************************/
 static block_t *BlockParseRTP( access_t *p_access, block_t *p_block )
 {
-    int     i_rtp_version;
-    int     i_CSRC_count;
-    int     i_payload_type;
-    int     i_skip = 0;
+    access_sys_t *p_sys = (access_sys_t *) p_access->p_sys;
+    int      i_rtp_version;
+    int      i_CSRC_count;
+    int      i_payload_type;
+    int      i_skip = 0;
+    int      i_extension_flag = 0;
+    int      i_extension_length = 0;
+    uint16_t i_sequence_number = 0;
+
+    if( p_block == NULL )
+        return NULL;
 
     if( p_block->i_buffer < RTP_HEADER_LEN )
         goto trash;
 
     /* Parse the header and make some verifications.
-     * See RFC 1889 & RFC 2250. */
-    i_rtp_version  = ( p_block->p_buffer[0] & 0xC0 ) >> 6;
-    i_CSRC_count   = ( p_block->p_buffer[0] & 0x0F );
-    i_payload_type = ( p_block->p_buffer[1] & 0x7F );
+     * See RFC 3550. */
+    i_rtp_version     = ( p_block->p_buffer[0] & 0xC0 ) >> 6;
+    i_CSRC_count      = p_block->p_buffer[0] & 0x0F;
+    i_extension_flag  = p_block->p_buffer[0] & 0x10;
+    i_payload_type    = p_block->p_buffer[1] & 0x7F;
+    i_sequence_number = (p_block->p_buffer[2] << 8 ) + p_block->p_buffer[3];
 
-    if ( i_rtp_version != 2 )
+    if( i_rtp_version != 2 )
         msg_Dbg( p_access, "RTP version is %u, should be 2", i_rtp_version );
 
     if( i_payload_type == 14 )
         i_skip = 4;
     else if( i_payload_type !=  33 && i_payload_type != 32 )
         msg_Dbg( p_access, "unsupported RTP payload type (%u)", i_payload_type );
+    if( i_extension_flag )
+        i_extension_length = 4 +
+            4 * ( (p_block->p_buffer[14] << 8) + p_block->p_buffer[15] );
 
-    i_skip += RTP_HEADER_LEN + 4*i_CSRC_count;
+    /* Skip header + CSRC extension field n*(32 bits) + extension */
+    i_skip += RTP_HEADER_LEN + 4*i_CSRC_count + i_extension_length;
 
-    /* A CSRC extension field is 32 bits in size (4 bytes) */
     if( i_skip >= p_block->i_buffer )
         goto trash;
 
-    /* Return the packet without the RTP header. */
+    /* Return the packet without the RTP header, remember seqno */
     p_block->i_buffer -= i_skip;
     p_block->p_buffer += i_skip;
+    p_block->i_seqno = i_sequence_number;
+
+#if 0
+    /* Emulate packet loss */
+    if ( (i_sequence_number % 4000) == 0)
+    {
+        msg_Warn( p_access, "Emulating packet drop" );
+        block_Release( p_block );
+        return NULL;
+    }
+#endif
 
     return p_block;
 
+
 trash:
     msg_Warn( p_access, "received a too short packet for RTP" );
     block_Release( p_block );
     return NULL;
 }
 
+static block_t *BlockPrebufferRTP( access_t *p_access, block_t *p_block )
+{
+    access_sys_t *p_sys = p_access->p_sys;
+    int64_t   i_first = mdate();
+    int       i_count = 0;
+    block_t   *p = p_block;
+
+    for( ;; )
+    {
+        int64_t i_date = mdate();
+
+        if( p && rtp_ChainInsert( p_access, p ))
+            i_count++;
+
+        /* Require at least 3 packets in the buffer */
+        if( i_count > 3 && (i_date - i_first) > p_sys->i_rtp_late )
+            break;
+
+        p = BlockParseRTP( p_access, BlockUDP( p_access ));
+        if( !p && (i_date - i_first) > p_sys->i_rtp_late ) 
+        {
+            msg_Err( p_access, "Error in RTP prebuffering!" );
+            break;
+        }
+    }
+
+    msg_Dbg( p_access, "RTP: prebuffered %d packets", i_count - 1 );
+    p_access->info.b_prebuffered = VLC_TRUE;
+    p = p_sys->p_list;
+    p_sys->p_list = p_sys->p_list->p_next;
+    p_sys->i_last_seqno = p->i_seqno;
+    p->p_next = NULL;
+    return p;
+}
+
 static block_t *BlockRTP( access_t *p_access )
 {
-    block_t *p_block = BlockUDP( p_access );
+    access_sys_t *p_sys = p_access->p_sys;
+    block_t *p;
 
-    if ( p_block != NULL )
-        return BlockParseRTP( p_access, p_block );
-    else
+again:
+    p = BlockParseRTP( p_access, BlockUDP( p_access ));
+
+    if ( !p ) 
         return NULL;
+
+    if ( !p_access->info.b_prebuffered )
+        return BlockPrebufferRTP( p_access, p );
+
+    if( !rtp_ChainInsert( p_access, p ))
+        goto again;
+
+    p = p_sys->p_list;
+    p_sys->p_list = p_sys->p_list->p_next;
+    p_sys->i_last_seqno++;
+    if( p_sys->i_last_seqno != p->i_seqno )
+    {
+        msg_Dbg( p_access, "RTP: packet(s) lost, expected %d, got %d",
+                 p_sys->i_last_seqno, p->i_seqno );
+        p_sys->i_last_seqno = p->i_seqno;
+    }
+    p->p_next = NULL;
+    return p;
 }
 
 /*****************************************************************************
@@ -424,7 +569,7 @@ static block_t *BlockChoose( access_t *p_access )
         return p_block;
 
     /* Parse the header and make some verifications.
-     * See RFC 1889 & RFC 2250. */
+     * See RFC 3550. */
 
     i_rtp_version  = ( p_block->p_buffer[0] & 0xC0 ) >> 6;
     i_CSRC_count   = ( p_block->p_buffer[0] & 0x0F );
@@ -460,7 +605,9 @@ static block_t *BlockChoose( access_t *p_access )
             return p_block;
     }
 
+    if( !BlockParseRTP( p_access, p_block )) return NULL;
+
     p_access->pf_block = BlockRTP;
 
-    return BlockParseRTP( p_access, p_block );
+    return BlockPrebufferRTP( p_access, p_block );
 }