[vlc-devel] [PATCH v5] udp: pull and buffer received packets in udp access module

Tzu-Jung Lee roylee17 at gmail.com
Wed Aug 21 07:09:32 CEST 2013


Signed-off-by: Tzu-Jung Lee <tjlee at ambarella.com>
---

It turns out the control freezing in earlier revision was due to the
blocking in the block_FifoGet(). This revision fixes this by invoking
block_FifoWake() in case the net_Read() returns less or euqal to zero
due to termination of connection os errors.

Note:
  This patch requeires the symbol of block_FifoWake to be exported.

 modules/access/udp.c | 114 +++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 92 insertions(+), 22 deletions(-)

diff --git a/modules/access/udp.c b/modules/access/udp.c
index 6613ebe..28b0f36 100644
--- a/modules/access/udp.c
+++ b/modules/access/udp.c
@@ -40,15 +40,19 @@
 #include <vlc_plugin.h>
 #include <vlc_access.h>
 #include <vlc_network.h>
+#include <vlc_block.h>
 
 #define MTU 65535
 
 /*****************************************************************************
  * Module descriptor
  *****************************************************************************/
-static int  Open ( vlc_object_t * );
+static int  Open( vlc_object_t * );
 static void Close( vlc_object_t * );
 
+#define BUFFER_TEXT N_("Receive buffer")
+#define BUFFER_LONGTEXT N_("UDP receive buffer size (bytes)" )
+
 vlc_module_begin ()
     set_shortname( N_("UDP" ) )
     set_description( N_("UDP input") )
@@ -56,6 +60,7 @@ vlc_module_begin ()
     set_subcategory( SUBCAT_INPUT_ACCESS )
 
     add_obsolete_integer( "server-port" ) /* since 2.0.0 */
+    add_integer( "udp-buffer", 0x400000, BUFFER_TEXT, BUFFER_LONGTEXT, true )
 
     set_capability( "access", 0 )
     add_shortcut( "udp", "udpstream", "udp4", "udp6" )
@@ -63,11 +68,22 @@ vlc_module_begin ()
     set_callbacks( Open, Close )
 vlc_module_end ()
 
+struct access_sys_t
+{
+    int          i_fd;
+    int          i_fifo_size;
+
+    block_fifo_t *p_fifo;
+
+    vlc_thread_t thread;
+};
+
 /*****************************************************************************
  * Local prototypes
  *****************************************************************************/
 static block_t *BlockUDP( access_t * );
 static int Control( access_t *, int, va_list );
+static void* ThreadRead( void *data );
 
 /*****************************************************************************
  * Open: open the socket
@@ -75,12 +91,19 @@ static int Control( access_t *, int, va_list );
 static int Open( vlc_object_t *p_this )
 {
     access_t     *p_access = (access_t*)p_this;
+    access_sys_t *p_sys;
 
     char *psz_name = strdup( p_access->psz_location );
     char *psz_parser;
     const char *psz_server_addr, *psz_bind_addr = "";
     int  i_bind_port = 1234, i_server_port = 0;
-    int fd;
+    int  i_fd;
+
+    p_sys = malloc( sizeof( *p_sys ) );
+    if( unlikely( p_sys == NULL ) )
+        return VLC_ENOMEM;
+
+    p_access->p_sys = p_sys;
 
     /* Set up p_access */
     access_InitFields( p_access );
@@ -128,15 +151,33 @@ static int Open( vlc_object_t *p_this )
     msg_Dbg( p_access, "opening server=%s:%d local=%s:%d",
              psz_server_addr, i_server_port, psz_bind_addr, i_bind_port );
 
-    fd = net_OpenDgram( p_access, psz_bind_addr, i_bind_port,
-                        psz_server_addr, i_server_port, IPPROTO_UDP );
-    free (psz_name);
-    if( fd == -1 )
+    i_fd = net_OpenDgram( p_access, psz_bind_addr, i_bind_port,
+                          psz_server_addr, i_server_port, IPPROTO_UDP );
+    free( psz_name );
+    if( i_fd == -1 )
     {
         msg_Err( p_access, "cannot open socket" );
+        net_Close( i_fd );
+        free( p_sys );
+        return VLC_EGENERIC;
+    }
+
+    p_sys->i_fd = i_fd;
+    p_sys->p_fifo = block_FifoNew();
+    if( unlikely( p_sys->p_fifo == NULL ) )
+        return VLC_ENOMEM;
+
+    p_sys->i_fifo_size = var_InheritInteger( p_access, "udp-buffer");
+
+    if( vlc_clone( &p_sys->thread, ThreadRead, p_access,
+                           VLC_THREAD_PRIORITY_HIGHEST ) )
+    {
+        msg_Err( p_access, "cannot spawn UDP access thread" );
+        block_FifoRelease( p_sys->p_fifo );
+        net_Close( i_fd );
+        free( p_sys );
         return VLC_EGENERIC;
     }
-    p_access->p_sys = (void *)(intptr_t)fd;
 
     return VLC_SUCCESS;
 }
@@ -147,8 +188,13 @@ static int Open( vlc_object_t *p_this )
 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;
 
-    net_Close( (intptr_t)p_access->p_sys );
+    vlc_cancel( p_sys->thread );
+    vlc_join( p_sys->thread, NULL );
+    block_FifoRelease( p_sys->p_fifo );
+    net_Close( p_sys->i_fd );
+    free( p_sys );
 }
 
 /*****************************************************************************
@@ -172,8 +218,8 @@ static int Control( access_t *p_access, int i_query, va_list args )
         /* */
         case ACCESS_GET_PTS_DELAY:
             pi_64 = (int64_t*)va_arg( args, int64_t * );
-            *pi_64 = INT64_C(1000)
-                   * var_InheritInteger(p_access, "network-caching");
+            *pi_64 = INT64_C( 1000 )
+                   * var_InheritInteger( p_access, "network-caching" );
             break;
 
         /* */
@@ -198,20 +244,44 @@ static int Control( access_t *p_access, int i_query, va_list args )
  *****************************************************************************/
 static block_t *BlockUDP( access_t *p_access )
 {
-    int fd = (intptr_t)p_access->p_sys;
+    access_sys_t *p_sys = p_access->p_sys;
 
-    /* Read data */
-    block_t *p_block = block_Alloc( MTU );
-    if( unlikely(p_block == NULL) )
-        return NULL;
+    return block_FifoGet( p_sys->p_fifo );
+}
+/*****************************************************************************
+ * ThreadRead: Pull packets from socket as soon as possible.
+ *****************************************************************************/
+static void* ThreadRead( void *data )
+{
+    access_t     *p_access = (access_t*)data;
+    access_sys_t *p_sys = p_access->p_sys;
+    int          i_fifo_size = p_sys->i_fifo_size;
 
-    ssize_t len = net_Read( p_access, fd, NULL,
-                            p_block->p_buffer, MTU, false );
-    if( len < 0 )
+    for( ;; )
     {
-        block_Release( p_block );
-        return NULL;
-    }
+        block_FifoPace( p_sys->p_fifo, SIZE_MAX, i_fifo_size );
+
+        block_t *p_pk = block_Alloc( MTU );
+        if( unlikely( p_pk == NULL ) )
+            return NULL;
 
-    return block_Realloc( p_block, 0, len );
+        ssize_t len = net_Read( p_access, p_sys->i_fd, NULL,
+                                p_pk->p_buffer, MTU, false );
+        if( unlikely( len <= 0 ) )
+        {
+            /* Preserve the errno before subsequent calls */
+            bool retry = ( len < 0 && errno == EINTR );
+
+            block_Release( p_pk );
+            block_FifoWake( p_sys->p_fifo );
+            if( retry )
+                continue;
+
+            return NULL;
+        }
+
+        p_pk = block_Realloc( p_pk, 0, len );
+        block_FifoPut( p_sys->p_fifo, p_pk );
+    }
+    return NULL;
 }
-- 
1.8.3.2




More information about the vlc-devel mailing list