]> git.sesse.net Git - vlc/blobdiff - modules/access/udp.c
udp: fix locking (fixes #14234) and cancellation
[vlc] / modules / access / udp.c
index 6818ef3682e8c158d40db21fc07e4df32fd94bad..c61bf8a8906b63a96418cf5b2c65c6d39d89c901 100644 (file)
 # include "config.h"
 #endif
 
+#include <errno.h>
 #include <vlc_common.h>
 #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 +61,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 +69,21 @@ vlc_module_begin ()
     set_callbacks( Open, Close )
 vlc_module_end ()
 
+struct access_sys_t
+{
+    int fd;
+    bool running;
+    size_t fifo_size;
+    block_fifo_t *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,16 +91,23 @@ 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 *sys = malloc( sizeof( *sys ) );
+    if( unlikely( sys == NULL ) )
+        return VLC_ENOMEM;
+
+    p_access->p_sys = sys;
+
+    /* Set up p_access */
+    access_InitFields( p_access );
+    ACCESS_SET_CALLBACKS( NULL, BlockUDP, Control, NULL );
 
     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;
 
-    /* Set up p_access */
-    access_InitFields( p_access );
-    ACCESS_SET_CALLBACKS( NULL, BlockUDP, Control, NULL );
+    if( unlikely(psz_name == NULL) )
+        goto error;
 
     /* Parse psz_name syntax :
      * [serveraddr[:serverport]][@[bindaddr]:[bindport]] */
@@ -128,15 +151,34 @@ 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 )
+    sys->fd = net_OpenDgram( p_access, psz_bind_addr, i_bind_port,
+                             psz_server_addr, i_server_port, IPPROTO_UDP );
+    free( psz_name );
+    if( sys->fd == -1 )
     {
         msg_Err( p_access, "cannot open socket" );
+        goto error;
+    }
+
+    sys->fifo = block_FifoNew();
+    if( unlikely( sys->fifo == NULL ) )
+    {
+        net_Close( sys->fd );
+        goto error;
+    }
+
+    sys->running = true;
+    sys->fifo_size = var_InheritInteger( p_access, "udp-buffer");
+
+    if( vlc_clone( &sys->thread, ThreadRead, p_access,
+                   VLC_THREAD_PRIORITY_INPUT ) )
+    {
+        block_FifoRelease( sys->fifo );
+        net_Close( sys->fd );
+error:
+        free( sys );
         return VLC_EGENERIC;
     }
-    p_access->p_sys = (void *)(intptr_t)fd;
 
     return VLC_SUCCESS;
 }
@@ -147,8 +189,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 *sys = p_access->p_sys;
 
-    net_Close( (intptr_t)p_access->p_sys );
+    vlc_cancel( sys->thread );
+    vlc_join( sys->thread, NULL );
+    block_FifoRelease( sys->fifo );
+    net_Close( sys->fd );
+    free( sys );
 }
 
 /*****************************************************************************
@@ -161,7 +208,6 @@ static int Control( access_t *p_access, int i_query, va_list args )
 
     switch( i_query )
     {
-        /* */
         case ACCESS_CAN_SEEK:
         case ACCESS_CAN_FASTSEEK:
         case ACCESS_CAN_PAUSE:
@@ -169,26 +215,15 @@ static int Control( access_t *p_access, int i_query, va_list args )
             pb_bool = (bool*)va_arg( args, bool* );
             *pb_bool = false;
             break;
-        /* */
+
         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");
             break;
 
-        /* */
-        case ACCESS_SET_PAUSE_STATE:
-        case ACCESS_GET_TITLE_INFO:
-        case ACCESS_SET_TITLE:
-        case ACCESS_SET_SEEKPOINT:
-        case ACCESS_SET_PRIVATE_ID_STATE:
-        case ACCESS_GET_CONTENT_TYPE:
-            return VLC_EGENERIC;
-
         default:
-            msg_Warn( p_access, "unimplemented query in control" );
             return VLC_EGENERIC;
-
     }
     return VLC_SUCCESS;
 }
@@ -198,22 +233,69 @@ static int Control( access_t *p_access, int i_query, va_list args )
  *****************************************************************************/
 static block_t *BlockUDP( access_t *p_access )
 {
-    access_sys_t *p_sys = p_access->p_sys;
-    block_t      *p_block;
-    ssize_t len;
+    access_sys_t *sys = p_access->p_sys;
+    block_t *block;
 
-    if( p_access->info.b_eof )
+    if (p_access->info.b_eof)
         return NULL;
 
-    /* Read data */
-    p_block = block_Alloc( MTU );
-    len = net_Read( p_access, (intptr_t)p_sys, NULL,
-                    p_block->p_buffer, MTU, false );
-    if( len < 0 )
+    vlc_fifo_Lock(sys->fifo);
+    while (vlc_fifo_IsEmpty(sys->fifo) && sys->running)
+       vlc_fifo_Wait(sys->fifo);
+
+    block = vlc_fifo_DequeueUnlocked(sys->fifo);
+    p_access->info.b_eof = !sys->running;
+    vlc_fifo_Unlock(sys->fifo);
+
+    return block;
+}
+
+/*****************************************************************************
+ * ThreadRead: Pull packets from socket as soon as possible.
+ *****************************************************************************/
+static void* ThreadRead( void *data )
+{
+    access_t *access = data;
+    access_sys_t *sys = access->p_sys;
+
+    for(;;)
     {
-        block_Release( p_block );
-        return NULL;
+        block_t *pkt = block_Alloc(MTU);
+        if (unlikely(pkt == NULL))
+            break;
+
+        ssize_t len;
+
+        block_cleanup_push(pkt);
+        do
+            len = net_Read(access, sys->fd, NULL, pkt->p_buffer, MTU, false);
+        while (len == -1 && errno != EINTR);
+        vlc_cleanup_pop();
+
+        if (len == -1)
+        {
+            block_Release(pkt);
+            break;
+        }
+
+        pkt->i_buffer = len;
+
+        vlc_fifo_Lock(sys->fifo);
+        /* Discard old buffers on overflow */
+        while (vlc_fifo_GetBytes(sys->fifo) + len > sys->fifo_size)
+        {
+            int canc = vlc_savecancel();
+            block_Release(vlc_fifo_DequeueUnlocked(sys->fifo));
+            vlc_restorecancel(canc);
+        }
+
+        vlc_fifo_QueueUnlocked(sys->fifo, pkt);
+        vlc_fifo_Unlock(sys->fifo);
     }
 
-    return block_Realloc( p_block, 0, len );
+    vlc_fifo_Lock(sys->fifo);
+    sys->running = false;
+    vlc_fifo_Signal(sys->fifo);
+    vlc_fifo_Unlock(sys->fifo);
+    return NULL;
 }