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  Seek    ( sout_access_out_t *, off_t  );
 108 static int Control( sout_access_out_t *, int, va_list );
 109
 110 static void* ThreadWrite( void * );
 111 static block_t *NewUDPPacket( sout_access_out_t *, mtime_t );
 112
 113 struct sout_access_out_sys_t
 114 {
 115     mtime_t       i_caching;
 116     int           i_handle;
 117     bool          b_mtu_warning;
 118     size_t        i_mtu;
 119
 120     block_fifo_t *p_fifo;
 121     block_fifo_t *p_empty_blocks;
 122     block_t      *p_buffer;
 123
 124     vlc_thread_t  thread;
 125 };
 126
 127 #define DEFAULT_PORT 1234
 128
 129 /*****************************************************************************
 130  * Open: open the file
 131  *****************************************************************************/
 132 static int Open( vlc_object_t *p_this )
 133 {
 134     sout_access_out_t       *p_access = (sout_access_out_t*)p_this;
 135     sout_access_out_sys_t   *p_sys;
 136
 137     char                *psz_dst_addr = NULL;
 138     int                 i_dst_port;
 139
 140     int                 i_handle;
 141
 142     config_ChainParse( p_access, SOUT_CFG_PREFIX,
 143                        ppsz_sout_options, p_access->p_cfg );
 144     config_ChainParse( p_access, "",
 145                        ppsz_core_options, p_access->p_cfg );
 146
 147     if (var_Create (p_access, "dst-port", VLC_VAR_INTEGER)
 148      || var_Create (p_access, "src-port", VLC_VAR_INTEGER)
 149      || var_Create (p_access, "dst-addr", VLC_VAR_STRING)
 150      || var_Create (p_access, "src-addr", VLC_VAR_STRING))
 151     {
 152         return VLC_ENOMEM;
 153     }
 154
 155     if( !( p_sys = malloc ( sizeof( *p_sys ) ) ) )
 156         return VLC_ENOMEM;
 157     p_access->p_sys = p_sys;
 158
 159     i_dst_port = DEFAULT_PORT;
 160     char *psz_parser = psz_dst_addr = strdup( p_access->psz_path );
 161     if( !psz_dst_addr )
 162     {
 163         free( p_sys );
 164         return VLC_ENOMEM;
 165     }
 166
 167     if (psz_parser[0] == '[')
 168         psz_parser = strchr (psz_parser, ']');
 169
 170     psz_parser = strchr (psz_parser ? psz_parser : psz_dst_addr, ':');
 171     if (psz_parser != NULL)
 172     {
 173         *psz_parser++ = '\0';
 174         i_dst_port = atoi (psz_parser);
 175     }
 176
 177     i_handle = net_ConnectDgram( p_this, psz_dst_addr, i_dst_port, -1,
 178                                  IPPROTO_UDP );
 179     free (psz_dst_addr);
 180
 181     if( i_handle == -1 )
 182     {
 183          msg_Err( p_access, "failed to create raw UDP socket" );
 184          free (p_sys);
 185          return VLC_EGENERIC;
 186     }
 187     else
 188     {
 189         char addr[NI_MAXNUMERICHOST];
 190         int port;
 191
 192         if (net_GetSockAddress (i_handle, addr, &port) == 0)
 193         {
 194             msg_Dbg (p_access, "source: %s port %d", addr, port);
 195             var_SetString (p_access, "src-addr", addr);
 196             var_SetInteger (p_access, "src-port", port);
 197         }
 198
 199         if (net_GetPeerAddress (i_handle, addr, &port) == 0)
 200         {
 201             msg_Dbg (p_access, "destination: %s port %d", addr, port);
 202             var_SetString (p_access, "dst-addr", addr);
 203             var_SetInteger (p_access, "dst-port", port);
 204         }
 205     }
 206     shutdown( i_handle, SHUT_RD );
 207
 208     p_sys->i_caching = UINT64_C(1000)
 209                      * var_GetInteger( p_access, SOUT_CFG_PREFIX "caching");
 210     p_sys->i_handle = i_handle;
 211     p_sys->i_mtu = var_CreateGetInteger( p_this, "mtu" );
 212     p_sys->b_mtu_warning = false;
 213     p_sys->p_fifo = block_FifoNew();
 214     p_sys->p_empty_blocks = block_FifoNew();
 215     p_sys->p_buffer = NULL;
 216
 217     if( vlc_clone( &p_sys->thread, ThreadWrite, p_access,
 218                            VLC_THREAD_PRIORITY_HIGHEST ) )
 219     {
 220         msg_Err( p_access, "cannot spawn sout access thread" );
 221         block_FifoRelease( p_sys->p_fifo );
 222         block_FifoRelease( p_sys->p_empty_blocks );
 223         net_Close (i_handle);
 224         free (p_sys);
 225         return VLC_EGENERIC;
 226     }
 227
 228     p_access->pf_write = Write;
 229     p_access->pf_seek = Seek;
 230     p_access->pf_control = Control;
 231
 232     return VLC_SUCCESS;
 233 }
 234
 235 /*****************************************************************************
 236  * Close: close the target
 237  *****************************************************************************/
 238 static void Close( vlc_object_t * p_this )
 239 {
 240     sout_access_out_t     *p_access = (sout_access_out_t*)p_this;
 241     sout_access_out_sys_t *p_sys = p_access->p_sys;
 242
 243     vlc_cancel( p_sys->thread );
 244     vlc_join( p_sys->thread, NULL );
 245     block_FifoRelease( p_sys->p_fifo );
 246     block_FifoRelease( p_sys->p_empty_blocks );
 247
 248     if( p_sys->p_buffer ) block_Release( p_sys->p_buffer );
 249
 250     net_Close( p_sys->i_handle );
 251     free( p_sys );
 252 }
 253
 254 static int Control( sout_access_out_t *p_access, int i_query, va_list args )
 255 {
 256     (void)p_access;
 257
 258     switch( i_query )
 259     {
 260         case ACCESS_OUT_CONTROLS_PACE:
 261             *va_arg( args, bool * ) = false;
 262             break;
 263
 264         default:
 265             return VLC_EGENERIC;
 266     }
 267     return VLC_SUCCESS;
 268 }
 269
 270 /*****************************************************************************
 271  * Write: standard write on a file descriptor.
 272  *****************************************************************************/
 273 static ssize_t Write( sout_access_out_t *p_access, block_t *p_buffer )
 274 {
 275     sout_access_out_sys_t *p_sys = p_access->p_sys;
 276     int i_len = 0;
 277
 278     while( p_buffer )
 279     {
 280         block_t *p_next;
 281         int i_packets = 0;
 282         mtime_t now = mdate();
 283
 284         if( !p_sys->b_mtu_warning && p_buffer->i_buffer > p_sys->i_mtu )
 285         {
 286             msg_Warn( p_access, "packet size > MTU, you should probably "
 287                       "increase the MTU" );
 288             p_sys->b_mtu_warning = true;
 289         }
 290
 291         /* Check if there is enough space in the buffer */
 292         if( p_sys->p_buffer &&
 293             p_sys->p_buffer->i_buffer + p_buffer->i_buffer > p_sys->i_mtu )
 294         {
 295             if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
 296             {
 297                 msg_Dbg( p_access, "late packet for UDP input (%"PRId64 ")",
 298                          now - p_sys->p_buffer->i_dts
 299                           - p_sys->i_caching );
 300             }
 301             block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
 302             p_sys->p_buffer = NULL;
 303         }
 304
 305         i_len += p_buffer->i_buffer;
 306         while( p_buffer->i_buffer )
 307         {
 308             size_t i_payload_size = p_sys->i_mtu;
 309             size_t i_write = __MIN( p_buffer->i_buffer, i_payload_size );
 310
 311             i_packets++;
 312
 313             if( !p_sys->p_buffer )
 314             {
 315                 p_sys->p_buffer = NewUDPPacket( p_access, p_buffer->i_dts );
 316                 if( !p_sys->p_buffer ) break;
 317             }
 318
 319             memcpy( p_sys->p_buffer->p_buffer + p_sys->p_buffer->i_buffer,
 320                     p_buffer->p_buffer, i_write );
 321
 322             p_sys->p_buffer->i_buffer += i_write;
 323             p_buffer->p_buffer += i_write;
 324             p_buffer->i_buffer -= i_write;
 325             if ( p_buffer->i_flags & BLOCK_FLAG_CLOCK )
 326             {
 327                 if ( p_sys->p_buffer->i_flags & BLOCK_FLAG_CLOCK )
 328                     msg_Warn( p_access, "putting two PCRs at once" );
 329                 p_sys->p_buffer->i_flags |= BLOCK_FLAG_CLOCK;
 330             }
 331
 332             if( p_sys->p_buffer->i_buffer == p_sys->i_mtu || i_packets > 1 )
 333             {
 334                 /* Flush */
 335                 if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
 336                 {
 337                     msg_Dbg( p_access, "late packet for udp input (%"PRId64 ")",
 338                              mdate() - p_sys->p_buffer->i_dts
 339                               - p_sys->i_caching );
 340                 }
 341                 block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
 342                 p_sys->p_buffer = NULL;
 343             }
 344         }
 345
 346         p_next = p_buffer->p_next;
 347         block_Release( p_buffer );
 348         p_buffer = p_next;
 349     }
 350
 351     return i_len;
 352 }
 353
 354 /*****************************************************************************
 355  * Seek: seek to a specific location in a file
 356  *****************************************************************************/
 357 static int Seek( sout_access_out_t *p_access, off_t i_pos )
 358 {
 359     (void) i_pos;
 360     msg_Err( p_access, "UDP sout access cannot seek" );
 361     return -1;
 362 }
 363
 364 /*****************************************************************************
 365  * NewUDPPacket: allocate a new UDP packet of size p_sys->i_mtu
 366  *****************************************************************************/
 367 static block_t *NewUDPPacket( sout_access_out_t *p_access, mtime_t i_dts)
 368 {
 369     sout_access_out_sys_t *p_sys = p_access->p_sys;
 370     block_t *p_buffer;
 371
 372     while ( block_FifoCount( p_sys->p_empty_blocks ) > MAX_EMPTY_BLOCKS )
 373     {
 374         p_buffer = block_FifoGet( p_sys->p_empty_blocks );
 375         block_Release( p_buffer );
 376     }
 377
 378     if( block_FifoCount( p_sys->p_empty_blocks ) == 0 )
 379     {
 380         p_buffer = block_Alloc( p_sys->i_mtu );
 381     }
 382     else
 383     {
 384         p_buffer = block_FifoGet(p_sys->p_empty_blocks );
 385         p_buffer->i_flags = 0;
 386         p_buffer = block_Realloc( p_buffer, 0, p_sys->i_mtu );
 387     }
 388
 389     p_buffer->i_dts = i_dts;
 390     p_buffer->i_buffer = 0;
 391
 392     return p_buffer;
 393 }
 394
 395 /*****************************************************************************
 396  * ThreadWrite: Write a packet on the network at the good time.
 397  *****************************************************************************/
 398 static void* ThreadWrite( void *data )
 399 {
 400     sout_access_out_t *p_access = data;
 401     sout_access_out_sys_t *p_sys = p_access->p_sys;
 402     mtime_t i_date_last = -1;
 403     const unsigned i_group = var_GetInteger( p_access,
 404                                              SOUT_CFG_PREFIX "group" );
 405     mtime_t i_to_send = i_group;
 406     unsigned i_dropped_packets = 0;
 407
 408     for (;;)
 409     {
 410         block_t *p_pk = block_FifoGet( p_sys->p_fifo );
 411         mtime_t       i_date, i_sent;
 412
 413         i_date = p_sys->i_caching + p_pk->i_dts;
 414         if( i_date_last > 0 )
 415         {
 416             if( i_date - i_date_last > 2000000 )
 417             {
 418                 if( !i_dropped_packets )
 419                     msg_Dbg( p_access, "mmh, hole (%"PRId64" > 2s) -> drop",
 420                              i_date - i_date_last );
 421
 422                 block_FifoPut( p_sys->p_empty_blocks, p_pk );
 423
 424                 i_date_last = i_date;
 425                 i_dropped_packets++;
 426                 continue;
 427             }
 428             else if( i_date - i_date_last < -1000 )
 429             {
 430                 if( !i_dropped_packets )
 431                     msg_Dbg( p_access, "mmh, packets in the past (%"PRId64")",
 432                              i_date_last - i_date );
 433             }
 434         }
 435
 436         block_cleanup_push( p_pk );
 437         i_to_send--;
 438         if( !i_to_send || (p_pk->i_flags & BLOCK_FLAG_CLOCK) )
 439         {
 440             mwait( i_date );
 441             i_to_send = i_group;
 442         }
 443         if ( send( p_sys->i_handle, p_pk->p_buffer, p_pk->i_buffer, 0 ) == -1 )
 444             msg_Warn( p_access, "send error: %s", vlc_strerror_c(errno) );
 445         vlc_cleanup_pop();
 446
 447         if( i_dropped_packets )
 448         {
 449             msg_Dbg( p_access, "dropped %i packets", i_dropped_packets );
 450             i_dropped_packets = 0;
 451         }
 452
 453 #if 1
 454         i_sent = mdate();
 455         if ( i_sent > i_date + 20000 )
 456         {
 457             msg_Dbg( p_access, "packet has been sent too late (%"PRId64 ")",
 458                      i_sent - i_date );
 459         }
 460 #endif
 461
 462         block_FifoPut( p_sys->p_empty_blocks, p_pk );
 463
 464         i_date_last = i_date;
 465     }
 466     return NULL;
 467 }