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

[Tzu-Jung Lee]

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

---

Add congestion control to avoid running out memory

 modules/access/udp.c | 91 ++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 74 insertions(+), 17 deletions(-)

diff --git a/modules/access/udp.c b/modules/access/udp.c
index 6613ebe..16ff1a3 100644
--- a/modules/access/udp.c
+++ b/modules/access/udp.c
@@ -40,6 +40,7 @@
 #include <vlc_plugin.h>
 #include <vlc_access.h>
 #include <vlc_network.h>
+#include <vlc_block.h>
 
 #define MTU 65535
 
@@ -49,6 +50,11 @@
 static int  Open ( vlc_object_t * );
 static void Close( vlc_object_t * );
 
+#define BUFFER_TEXT N_("UDP buffer size")
+#define BUFFER_LONGTEXT N_("The UDP access module pulls recieved packets " \
+                           "from socket buffer into a fifo to avoid or " \
+                           "reduce packet loss due to socket buffer " \
+			   "overflow. The default size is 4M Bytes" )
 vlc_module_begin ()
     set_shortname( N_("UDP" ) )
     set_description( N_("UDP input") )
@@ -56,6 +62,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 +70,22 @@ vlc_module_begin ()
     set_callbacks( Open, Close )
 vlc_module_end ()
 
+struct access_sys_t
+{
+    int          i_handle;
+    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 +93,18 @@ 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_handle;
+
+    if( !( p_sys = malloc ( sizeof( *p_sys ) ) ) )
+        return VLC_ENOMEM;
+
+    p_access->p_sys = p_sys;
 
     /* Set up p_access */
     access_InitFields( p_access );
@@ -128,15 +152,28 @@ 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,
+    i_handle = net_OpenDgram( p_access, psz_bind_addr, i_bind_port,
                         psz_server_addr, i_server_port, IPPROTO_UDP );
     free (psz_name);
-    if( fd == -1 )
+    if( i_handle == -1 )
     {
         msg_Err( p_access, "cannot open socket" );
         return VLC_EGENERIC;
     }
-    p_access->p_sys = (void *)(intptr_t)fd;
+
+    p_sys->i_handle = i_handle;
+    p_sys->p_fifo = block_FifoNew();
+    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_handle);
+        free (p_sys);
+        return VLC_EGENERIC;
+    }
 
     return VLC_SUCCESS;
 }
@@ -147,8 +184,11 @@ 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 );
+    block_FifoRelease( p_sys->p_fifo );
+    net_Close( p_sys->i_handle );
+    free (p_sys);
 }
 
 /*****************************************************************************
@@ -198,20 +238,37 @@ 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);
 
-    return block_Realloc( p_block, 0, len );
+        block_t *p_pk = block_Alloc( MTU );
+        if( unlikely(p_pk == NULL) )
+            return NULL;
+
+        ssize_t len = net_Read( p_access, p_sys->i_handle, NULL,
+                                p_pk->p_buffer, MTU, false );
+        if( len < 0 )
+        {
+            block_Release( p_pk );
+            return NULL;
+        }
+
+        p_pk = block_Realloc( p_pk, 0, len );
+        block_FifoPut( p_sys->p_fifo, p_pk );
+    }
+    return NULL;
 }
-- 
1.8.3.2