modules/access_output/udp.c

   1 /*****************************************************************************
   2  * udp.c
   3  *****************************************************************************
   4  * Copyright (C) 2001-2007 VLC authors and VideoLAN
   5  * $Id$
   6  *
   7  * Authors: Laurent Aimar <fenrir@via.ecp.fr>
   8  *          Eric Petit <titer@videolan.org>
   9  *
  10  * This program is free software; you can redistribute it and/or modify it
  11  * under the terms of the GNU Lesser General Public License as published by
  12  * the Free Software Foundation; either version 2.1 of the License, or
  13  * (at your option) any later version.
  14  *
  15  * This program is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  18  * GNU Lesser General Public License for more details.
  19  *
  20  * You should have received a copy of the GNU Lesser General Public License
  21  * along with this program; if not, write to the Free Software Foundation,
  22  * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
  23  *****************************************************************************/
  24
  25 /*****************************************************************************
  26  * Preamble
  27  *****************************************************************************/
  28 #ifdef HAVE_CONFIG_H
  29 # include "config.h"
  30 #endif
  31
  32 #include <vlc_common.h>
  33 #include <vlc_plugin.h>
  34
  35 #include <sys/types.h>
  36 #include <unistd.h>
  37 #include <assert.h>
  38 #include <errno.h>
  39
  40 #include <vlc_sout.h>
  41 #include <vlc_block.h>
  42
  43 #ifdef _WIN32
  44 #   include <winsock2.h>
  45 #   include <ws2tcpip.h>
  46 #else
  47 #   include <sys/socket.h>
  48 #endif
  49
  50 #include <vlc_network.h>
  51
  52 #define MAX_EMPTY_BLOCKS 200
  53
  54 /*****************************************************************************
  55  * Module descriptor
  56  *****************************************************************************/
  57 static int  Open ( vlc_object_t * );
  58 static void Close( vlc_object_t * );
  59
  60 #define SOUT_CFG_PREFIX "sout-udp-"
  61
  62 #define CACHING_TEXT N_("Caching value (ms)")
  63 #define CACHING_LONGTEXT N_( \
  64     "Default caching value for outbound UDP streams. This " \
  65     "value should be set in milliseconds." )
  66
  67 #define GROUP_TEXT N_("Group packets")
  68 #define GROUP_LONGTEXT N_("Packets can be sent one by one at the right time " \
  69                           "or by groups. You can choose the number " \
  70                           "of packets that will be sent at a time. It " \
  71                           "helps reducing the scheduling load on " \
  72                           "heavily-loaded systems." )
  73
  74 vlc_module_begin ()
  75     set_description( N_("UDP stream output") )
  76     set_shortname( "UDP" )
  77     set_category( CAT_SOUT )
  78     set_subcategory( SUBCAT_SOUT_ACO )
  79     add_integer( SOUT_CFG_PREFIX "caching", DEFAULT_PTS_DELAY / 1000, CACHING_TEXT, CACHING_LONGTEXT, true )
  80     add_integer( SOUT_CFG_PREFIX "group", 1, GROUP_TEXT, GROUP_LONGTEXT,
  81                                  true )
  82
  83     set_capability( "sout access", 0 )
  84     add_shortcut( "udp" )
  85     set_callbacks( Open, Close )
  86 vlc_module_end ()
  87
  88 /*****************************************************************************
  89  * Exported prototypes
  90  *****************************************************************************/
  91
  92 static const char *const ppsz_sout_options[] = {
  93     "caching",
  94     "group",
  95     NULL
  96 };
  97
  98 /* Options handled by the libvlc network core */
  99 static const char *const ppsz_core_options[] = {
 100     "dscp",
 101     "ttl",
 102     "miface",
 103     NULL
 104 };
 105
 106 static ssize_t Write   ( sout_access_out_t *, block_t * );
 107 static int Control( sout_access_out_t *, int, va_list );
 108
 109 static void* ThreadWrite( void * );
 110 static block_t *NewUDPPacket( sout_access_out_t *, mtime_t );
 111
 112 struct sout_access_out_sys_t
 113 {
 114     mtime_t       i_caching;
 115     int           i_handle;
 116     bool          b_mtu_warning;
 117     size_t        i_mtu;
 118
 119     block_fifo_t *p_fifo;
 120     block_fifo_t *p_empty_blocks;
 121     block_t      *p_buffer;
 122
 123     vlc_thread_t  thread;
 124 };
 125
 126 #define DEFAULT_PORT 1234
 127
 128 /*****************************************************************************
 129  * Open: open the file
 130  *****************************************************************************/
 131 static int Open( vlc_object_t *p_this )
 132 {
 133     sout_access_out_t       *p_access = (sout_access_out_t*)p_this;
 134     sout_access_out_sys_t   *p_sys;
 135
 136     char                *psz_dst_addr = NULL;
 137     int                 i_dst_port;
 138
 139     int                 i_handle;
 140
 141     config_ChainParse( p_access, SOUT_CFG_PREFIX,
 142                        ppsz_sout_options, p_access->p_cfg );
 143     config_ChainParse( p_access, "",
 144                        ppsz_core_options, p_access->p_cfg );
 145
 146     if (var_Create (p_access, "dst-port", VLC_VAR_INTEGER)
 147      || var_Create (p_access, "src-port", VLC_VAR_INTEGER)
 148      || var_Create (p_access, "dst-addr", VLC_VAR_STRING)
 149      || var_Create (p_access, "src-addr", VLC_VAR_STRING))
 150     {
 151         return VLC_ENOMEM;
 152     }
 153
 154     if( !( p_sys = malloc ( sizeof( *p_sys ) ) ) )
 155         return VLC_ENOMEM;
 156     p_access->p_sys = p_sys;
 157
 158     i_dst_port = DEFAULT_PORT;
 159     char *psz_parser = psz_dst_addr = strdup( p_access->psz_path );
 160     if( !psz_dst_addr )
 161     {
 162         free( p_sys );
 163         return VLC_ENOMEM;
 164     }
 165
 166     if (psz_parser[0] == '[')
 167         psz_parser = strchr (psz_parser, ']');
 168
 169     psz_parser = strchr (psz_parser ? psz_parser : psz_dst_addr, ':');
 170     if (psz_parser != NULL)
 171     {
 172         *psz_parser++ = '\0';
 173         i_dst_port = atoi (psz_parser);
 174     }
 175
 176     i_handle = net_ConnectDgram( p_this, psz_dst_addr, i_dst_port, -1,
 177                                  IPPROTO_UDP );
 178     free (psz_dst_addr);
 179
 180     if( i_handle == -1 )
 181     {
 182          msg_Err( p_access, "failed to create raw UDP socket" );
 183          free (p_sys);
 184          return VLC_EGENERIC;
 185     }
 186     else
 187     {
 188         char addr[NI_MAXNUMERICHOST];
 189         int port;
 190
 191         if (net_GetSockAddress (i_handle, addr, &port) == 0)
 192         {
 193             msg_Dbg (p_access, "source: %s port %d", addr, port);
 194             var_SetString (p_access, "src-addr", addr);
 195             var_SetInteger (p_access, "src-port", port);
 196         }
 197
 198         if (net_GetPeerAddress (i_handle, addr, &port) == 0)
 199         {
 200             msg_Dbg (p_access, "destination: %s port %d", addr, port);
 201             var_SetString (p_access, "dst-addr", addr);
 202             var_SetInteger (p_access, "dst-port", port);
 203         }
 204     }
 205     shutdown( i_handle, SHUT_RD );
 206
 207     p_sys->i_caching = UINT64_C(1000)
 208                      * var_GetInteger( p_access, SOUT_CFG_PREFIX "caching");
 209     p_sys->i_handle = i_handle;
 210     p_sys->i_mtu = var_CreateGetInteger( p_this, "mtu" );
 211     p_sys->b_mtu_warning = false;
 212     p_sys->p_fifo = block_FifoNew();
 213     p_sys->p_empty_blocks = block_FifoNew();
 214     p_sys->p_buffer = NULL;
 215
 216     if( vlc_clone( &p_sys->thread, ThreadWrite, p_access,
 217                            VLC_THREAD_PRIORITY_HIGHEST ) )
 218     {
 219         msg_Err( p_access, "cannot spawn sout access thread" );
 220         block_FifoRelease( p_sys->p_fifo );
 221         block_FifoRelease( p_sys->p_empty_blocks );
 222         net_Close (i_handle);
 223         free (p_sys);
 224         return VLC_EGENERIC;
 225     }
 226
 227     p_access->pf_write = Write;
 228     p_access->pf_control = Control;
 229
 230     return VLC_SUCCESS;
 231 }
 232
 233 /*****************************************************************************
 234  * Close: close the target
 235  *****************************************************************************/
 236 static void Close( vlc_object_t * p_this )
 237 {
 238     sout_access_out_t     *p_access = (sout_access_out_t*)p_this;
 239     sout_access_out_sys_t *p_sys = p_access->p_sys;
 240
 241     vlc_cancel( p_sys->thread );
 242     vlc_join( p_sys->thread, NULL );
 243     block_FifoRelease( p_sys->p_fifo );
 244     block_FifoRelease( p_sys->p_empty_blocks );
 245
 246     if( p_sys->p_buffer ) block_Release( p_sys->p_buffer );
 247
 248     net_Close( p_sys->i_handle );
 249     free( p_sys );
 250 }
 251
 252 static int Control( sout_access_out_t *p_access, int i_query, va_list args )
 253 {
 254     (void)p_access;
 255
 256     switch( i_query )
 257     {
 258         case ACCESS_OUT_CONTROLS_PACE:
 259             *va_arg( args, bool * ) = false;
 260             break;
 261
 262         default:
 263             return VLC_EGENERIC;
 264     }
 265     return VLC_SUCCESS;
 266 }
 267
 268 /*****************************************************************************
 269  * Write: standard write on a file descriptor.
 270  *****************************************************************************/
 271 static ssize_t Write( sout_access_out_t *p_access, block_t *p_buffer )
 272 {
 273     sout_access_out_sys_t *p_sys = p_access->p_sys;
 274     int i_len = 0;
 275
 276     while( p_buffer )
 277     {
 278         block_t *p_next;
 279         int i_packets = 0;
 280         mtime_t now = mdate();
 281
 282         if( !p_sys->b_mtu_warning && p_buffer->i_buffer > p_sys->i_mtu )
 283         {
 284             msg_Warn( p_access, "packet size > MTU, you should probably "
 285                       "increase the MTU" );
 286             p_sys->b_mtu_warning = true;
 287         }
 288
 289         /* Check if there is enough space in the buffer */
 290         if( p_sys->p_buffer &&
 291             p_sys->p_buffer->i_buffer + p_buffer->i_buffer > p_sys->i_mtu )
 292         {
 293             if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
 294             {
 295                 msg_Dbg( p_access, "late packet for UDP input (%"PRId64 ")",
 296                          now - p_sys->p_buffer->i_dts
 297                           - p_sys->i_caching );
 298             }
 299             block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
 300             p_sys->p_buffer = NULL;
 301         }
 302
 303         i_len += p_buffer->i_buffer;
 304         while( p_buffer->i_buffer )
 305         {
 306             size_t i_payload_size = p_sys->i_mtu;
 307             size_t i_write = __MIN( p_buffer->i_buffer, i_payload_size );
 308
 309             i_packets++;
 310
 311             if( !p_sys->p_buffer )
 312             {
 313                 p_sys->p_buffer = NewUDPPacket( p_access, p_buffer->i_dts );
 314                 if( !p_sys->p_buffer ) break;
 315             }
 316
 317             memcpy( p_sys->p_buffer->p_buffer + p_sys->p_buffer->i_buffer,
 318                     p_buffer->p_buffer, i_write );
 319
 320             p_sys->p_buffer->i_buffer += i_write;
 321             p_buffer->p_buffer += i_write;
 322             p_buffer->i_buffer -= i_write;
 323             if ( p_buffer->i_flags & BLOCK_FLAG_CLOCK )
 324             {
 325                 if ( p_sys->p_buffer->i_flags & BLOCK_FLAG_CLOCK )
 326                     msg_Warn( p_access, "putting two PCRs at once" );
 327                 p_sys->p_buffer->i_flags |= BLOCK_FLAG_CLOCK;
 328             }
 329
 330             if( p_sys->p_buffer->i_buffer == p_sys->i_mtu || i_packets > 1 )
 331             {
 332                 /* Flush */
 333                 if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
 334                 {
 335                     msg_Dbg( p_access, "late packet for udp input (%"PRId64 ")",
 336                              mdate() - p_sys->p_buffer->i_dts
 337                               - p_sys->i_caching );
 338                 }
 339                 block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
 340                 p_sys->p_buffer = NULL;
 341             }
 342         }
 343
 344         p_next = p_buffer->p_next;
 345         block_Release( p_buffer );
 346         p_buffer = p_next;
 347     }
 348
 349     return i_len;
 350 }
 351
 352 /*****************************************************************************
 353  * NewUDPPacket: allocate a new UDP packet of size p_sys->i_mtu
 354  *****************************************************************************/
 355 static block_t *NewUDPPacket( sout_access_out_t *p_access, mtime_t i_dts)
 356 {
 357     sout_access_out_sys_t *p_sys = p_access->p_sys;
 358     block_t *p_buffer;
 359
 360     while ( block_FifoCount( p_sys->p_empty_blocks ) > MAX_EMPTY_BLOCKS )
 361     {
 362         p_buffer = block_FifoGet( p_sys->p_empty_blocks );
 363         block_Release( p_buffer );
 364     }
 365
 366     if( block_FifoCount( p_sys->p_empty_blocks ) == 0 )
 367     {
 368         p_buffer = block_Alloc( p_sys->i_mtu );
 369     }
 370     else
 371     {
 372         p_buffer = block_FifoGet(p_sys->p_empty_blocks );
 373         p_buffer->i_flags = 0;
 374         p_buffer = block_Realloc( p_buffer, 0, p_sys->i_mtu );
 375     }
 376
 377     p_buffer->i_dts = i_dts;
 378     p_buffer->i_buffer = 0;
 379
 380     return p_buffer;
 381 }
 382
 383 /*****************************************************************************
 384  * ThreadWrite: Write a packet on the network at the good time.
 385  *****************************************************************************/
 386 static void* ThreadWrite( void *data )
 387 {
 388     sout_access_out_t *p_access = data;
 389     sout_access_out_sys_t *p_sys = p_access->p_sys;
 390     mtime_t i_date_last = -1;
 391     const unsigned i_group = var_GetInteger( p_access,
 392                                              SOUT_CFG_PREFIX "group" );
 393     mtime_t i_to_send = i_group;
 394     unsigned i_dropped_packets = 0;
 395
 396     for (;;)
 397     {
 398         block_t *p_pk = block_FifoGet( p_sys->p_fifo );
 399         mtime_t       i_date, i_sent;
 400
 401         i_date = p_sys->i_caching + p_pk->i_dts;
 402         if( i_date_last > 0 )
 403         {
 404             if( i_date - i_date_last > 2000000 )
 405             {
 406                 if( !i_dropped_packets )
 407                     msg_Dbg( p_access, "mmh, hole (%"PRId64" > 2s) -> drop",
 408                              i_date - i_date_last );
 409
 410                 block_FifoPut( p_sys->p_empty_blocks, p_pk );
 411
 412                 i_date_last = i_date;
 413                 i_dropped_packets++;
 414                 continue;
 415             }
 416             else if( i_date - i_date_last < -1000 )
 417             {
 418                 if( !i_dropped_packets )
 419                     msg_Dbg( p_access, "mmh, packets in the past (%"PRId64")",
 420                              i_date_last - i_date );
 421             }
 422         }
 423
 424         block_cleanup_push( p_pk );
 425         i_to_send--;
 426         if( !i_to_send || (p_pk->i_flags & BLOCK_FLAG_CLOCK) )
 427         {
 428             mwait( i_date );
 429             i_to_send = i_group;
 430         }
 431         if ( send( p_sys->i_handle, p_pk->p_buffer, p_pk->i_buffer, 0 ) == -1 )
 432             msg_Warn( p_access, "send error: %s", vlc_strerror_c(errno) );
 433         vlc_cleanup_pop();
 434
 435         if( i_dropped_packets )
 436         {
 437             msg_Dbg( p_access, "dropped %i packets", i_dropped_packets );
 438             i_dropped_packets = 0;
 439         }
 440
 441 #if 1
 442         i_sent = mdate();
 443         if ( i_sent > i_date + 20000 )
 444         {
 445             msg_Dbg( p_access, "packet has been sent too late (%"PRId64 ")",
 446                      i_sent - i_date );
 447         }
 448 #endif
 449
 450         block_FifoPut( p_sys->p_empty_blocks, p_pk );
 451
 452         i_date_last = i_date;
 453     }
 454     return NULL;
 455 }