modules/access/srt.c

   1 /*****************************************************************************
   2  * srt.c: SRT (Secure Reliable Transport) access module
   3  *****************************************************************************
   4  * Copyright (C) 2017-2018, Collabora Ltd.
   5  * Copyright (C) 2018, Haivision Systems Inc.
   6  *
   7  * Authors: Justin Kim <justin.kim@collabora.com>
   8  *          Roman Diouskine <rdiouskine@haivision.com>
   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 #include "srt_common.h"
  25
  26 #include <vlc_fs.h>
  27 #include <vlc_plugin.h>
  28 #include <vlc_access.h>
  29 #include <vlc_interrupt.h>
  30
  31 #include <vlc_network.h>
  32 #include <vlc_url.h>
  33
  34
  35
  36 typedef struct
  37 {
  38     SRTSOCKET   sock;
  39     int         i_poll_id;
  40     vlc_mutex_t lock;
  41     bool        b_interrupted;
  42     char       *psz_host;
  43     int         i_port;
  44     int         i_chunks; /* Number of chunks to allocate in the next read */
  45 } stream_sys_t;
  46
  47
  48
  49 static void srt_wait_interrupted(void *p_data)
  50 {
  51     stream_t *p_stream = p_data;
  52     stream_sys_t *p_sys = p_stream->p_sys;
  53
  54     vlc_mutex_lock( &p_sys->lock );
  55     if ( p_sys->i_poll_id >= 0 &&  p_sys->sock != SRT_INVALID_SOCK )
  56     {
  57         p_sys->b_interrupted = true;
  58
  59         msg_Dbg( p_stream, "Waking up srt_epoll_wait");
  60
  61         /* Removing all socket descriptors from the monitoring list
  62          * wakes up SRT's threads. We only have one to remove. */
  63         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
  64     }
  65     vlc_mutex_unlock( &p_sys->lock );
  66 }
  67
  68 static int Control(stream_t *p_stream, int i_query, va_list args)
  69 {
  70     int i_ret = VLC_SUCCESS;
  71
  72     switch( i_query )
  73     {
  74         case STREAM_CAN_SEEK:
  75         case STREAM_CAN_FASTSEEK:
  76         case STREAM_CAN_PAUSE:
  77         case STREAM_CAN_CONTROL_PACE:
  78             *va_arg( args, bool * ) = false;
  79             break;
  80         case STREAM_GET_PTS_DELAY:
  81             *va_arg( args, vlc_tick_t * ) = VLC_TICK_FROM_MS(
  82                    var_InheritInteger(p_stream, "network-caching") );
  83             break;
  84         default:
  85             i_ret = VLC_EGENERIC;
  86             break;
  87     }
  88
  89     return i_ret;
  90 }
  91
  92 static bool srt_schedule_reconnect(stream_t *p_stream)
  93 {
  94     vlc_object_t *strm_obj = (vlc_object_t *) p_stream;
  95     int i_latency=var_InheritInteger( p_stream, SRT_PARAM_LATENCY );
  96     int i_payload_size = var_InheritInteger( p_stream, SRT_PARAM_PAYLOAD_SIZE );
  97     int stat;
  98     char *psz_passphrase = var_InheritString( p_stream, SRT_PARAM_PASSPHRASE );
  99     bool passphrase_needs_free = true;
 100     char *url = NULL;
 101     srt_params_t params;
 102     struct addrinfo hints = {
 103         .ai_socktype = SOCK_DGRAM,
 104     }, *res = NULL;
 105
 106     stream_sys_t *p_sys = p_stream->p_sys;
 107     bool failed = false;
 108
 109     stat = vlc_getaddrinfo( p_sys->psz_host, p_sys->i_port, &hints, &res );
 110     if ( stat )
 111     {
 112         msg_Err( p_stream, "Cannot resolve [%s]:%d (reason: %s)",
 113                  p_sys->psz_host,
 114                  p_sys->i_port,
 115                  gai_strerror( stat ) );
 116
 117         failed = true;
 118         goto out;
 119     }
 120
 121     /* Always start with a fresh socket */
 122     if (p_sys->sock != SRT_INVALID_SOCK)
 123     {
 124         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 125         srt_close( p_sys->sock );
 126     }
 127
 128     p_sys->sock = srt_socket( res->ai_family, SOCK_DGRAM, 0 );
 129     if ( p_sys->sock == SRT_INVALID_SOCK )
 130     {
 131         msg_Err( p_stream, "Failed to open socket." );
 132         failed = true;
 133         goto out;
 134     }
 135
 136     if (p_stream->psz_url) {
 137         url = strdup( p_stream->psz_url );
 138         if (srt_parse_url( url, &params )) {
 139             if (params.latency != -1)
 140                 i_latency = params.latency;
 141             if (params.payload_size != -1)
 142                 i_payload_size = params.payload_size;
 143             if (params.passphrase != NULL) {
 144                 free( psz_passphrase );
 145                 passphrase_needs_free = false;
 146                 psz_passphrase = (char *) params.passphrase;
 147             }
 148         }
 149     }
 150
 151     /* Make SRT non-blocking */
 152     srt_setsockopt( p_sys->sock, 0, SRTO_SNDSYN,
 153         &(bool) { false }, sizeof( bool ) );
 154     srt_setsockopt( p_sys->sock, 0, SRTO_RCVSYN,
 155         &(bool) { false }, sizeof( bool ) );
 156
 157     /* Make sure TSBPD mode is enable (SRT mode) */
 158     srt_setsockopt( p_sys->sock, 0, SRTO_TSBPDMODE,
 159         &(int) { 1 }, sizeof( int ) );
 160
 161     /* This is an access module so it is always a receiver */
 162     srt_setsockopt( p_sys->sock, 0, SRTO_SENDER,
 163         &(int) { 0 }, sizeof( int ) );
 164
 165     /* Set latency */
 166     srt_set_socket_option( strm_obj, SRT_PARAM_LATENCY, p_sys->sock,
 167             SRTO_TSBPDDELAY, &i_latency, sizeof(i_latency) );
 168
 169     /* set passphrase */
 170     if (psz_passphrase != NULL && psz_passphrase[0] != '\0') {
 171         int i_key_length = var_InheritInteger( p_stream, SRT_PARAM_KEY_LENGTH );
 172
 173         srt_set_socket_option( strm_obj, SRT_PARAM_KEY_LENGTH, p_sys->sock,
 174                 SRTO_PBKEYLEN, &i_key_length, sizeof(i_key_length) );
 175
 176         srt_set_socket_option( strm_obj, SRT_PARAM_PASSPHRASE, p_sys->sock,
 177                 SRTO_PASSPHRASE, psz_passphrase, strlen(psz_passphrase) );
 178     }
 179
 180     /* set maximum payload size */
 181     srt_set_socket_option( strm_obj, SRT_PARAM_PAYLOAD_SIZE, p_sys->sock,
 182             SRTO_PAYLOADSIZE, &i_payload_size, sizeof(i_payload_size) );
 183
 184
 185     srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
 186         &(int) { SRT_EPOLL_ERR | SRT_EPOLL_IN });
 187
 188     /* Schedule a connect */
 189     msg_Dbg( p_stream, "Schedule SRT connect (dest addresss: %s, port: %d).",
 190         p_sys->psz_host, p_sys->i_port);
 191
 192     stat = srt_connect( p_sys->sock, res->ai_addr, res->ai_addrlen );
 193     if (stat == SRT_ERROR) {
 194         msg_Err( p_stream, "Failed to connect to server (reason: %s)",
 195                 srt_getlasterror_str() );
 196         failed = true;
 197     }
 198
 199     /* Reset the number of chunks to allocate as the bitrate of
 200      * the stream may have changed.
 201      */
 202     p_sys->i_chunks = SRT_MIN_CHUNKS_TRYREAD;
 203
 204 out:
 205     if (failed && p_sys->sock != SRT_INVALID_SOCK)
 206     {
 207         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 208         srt_close(p_sys->sock);
 209         p_sys->sock = SRT_INVALID_SOCK;
 210     }
 211
 212     if (passphrase_needs_free)
 213         free( psz_passphrase );
 214     freeaddrinfo( res );
 215     free( url );
 216
 217     return !failed;
 218 }
 219
 220 static block_t *BlockSRT(stream_t *p_stream, bool *restrict eof)
 221 {
 222     stream_sys_t *p_sys = p_stream->p_sys;
 223     int i_chunk_size = var_InheritInteger( p_stream, SRT_PARAM_CHUNK_SIZE );
 224     int i_poll_timeout = var_InheritInteger( p_stream, SRT_PARAM_POLL_TIMEOUT );
 225     /* SRT doesn't have a concept of EOF for live streams. */
 226     VLC_UNUSED(eof);
 227
 228     if ( vlc_killed() )
 229     {
 230         /* We are told to stop. Stop. */
 231         return NULL;
 232     }
 233
 234     if ( p_sys->i_chunks == 0 )
 235         p_sys->i_chunks = SRT_MIN_CHUNKS_TRYREAD;
 236
 237     size_t i_chunk_size_actual = ( i_chunk_size > 0 )
 238         ? i_chunk_size : SRT_DEFAULT_CHUNK_SIZE;
 239     size_t bufsize = i_chunk_size_actual * p_sys->i_chunks;
 240     block_t *pkt = block_Alloc( bufsize );
 241     if ( unlikely( pkt == NULL ) )
 242     {
 243         return NULL;
 244     }
 245
 246     vlc_interrupt_register( srt_wait_interrupted, p_stream);
 247
 248     SRTSOCKET ready[1];
 249     int readycnt = 1;
 250     while ( srt_epoll_wait( p_sys->i_poll_id,
 251         ready, &readycnt, 0, 0,
 252         i_poll_timeout, NULL, 0, NULL, 0 ) >= 0)
 253     {
 254         if ( readycnt < 0  || ready[0] != p_sys->sock )
 255         {
 256             /* should never happen, force recovery */
 257             srt_close(p_sys->sock);
 258             p_sys->sock = SRT_INVALID_SOCK;
 259         }
 260
 261         switch( srt_getsockstate( p_sys->sock ) )
 262         {
 263             case SRTS_CONNECTED:
 264                 /* Good to go */
 265                 break;
 266             case SRTS_BROKEN:
 267             case SRTS_NONEXIST:
 268             case SRTS_CLOSED:
 269                 /* Failed. Schedule recovery. */
 270                 if ( !srt_schedule_reconnect( p_stream ) )
 271                     msg_Err( p_stream, "Failed to schedule connect" );
 272                 /* Fall-through */
 273             default:
 274                 /* Not ready */
 275                 continue;
 276         }
 277
 278         /* Try to get as much data as possible out of the lib, if there
 279          * is still some left, increase the number of chunks to read so that
 280          * it will read faster on the next iteration. This way the buffer will
 281          * grow until it reads fast enough to keep the library empty after
 282          * each iteration.
 283          */
 284         pkt->i_buffer = 0;
 285         while ( ( bufsize - pkt->i_buffer ) >= i_chunk_size_actual )
 286         {
 287             int stat = srt_recvmsg( p_sys->sock,
 288                 (char *)( pkt->p_buffer + pkt->i_buffer ),
 289                 bufsize - pkt->i_buffer );
 290             if ( stat <= 0 )
 291             {
 292                 break;
 293             }
 294             pkt->i_buffer += (size_t)stat;
 295         }
 296
 297         msg_Dbg ( p_stream, "Read %zu bytes out of a max of %zu"
 298             " (%d chunks of %zu bytes)", pkt->i_buffer,
 299             p_sys->i_chunks * i_chunk_size_actual, p_sys->i_chunks,
 300                  i_chunk_size_actual );
 301
 302
 303         /* Gradually adjust number of chunks we read at a time
 304         * up to a predefined maximum. The actual number we might
 305         * settle on depends on stream's bit rate.
 306         */
 307         size_t rem = bufsize - pkt->i_buffer;
 308         if ( rem < i_chunk_size_actual )
 309         {
 310             if ( p_sys->i_chunks < SRT_MAX_CHUNKS_TRYREAD )
 311             {
 312                 p_sys->i_chunks++;
 313             }
 314         }
 315
 316         goto out;
 317     }
 318
 319     /* if the poll reports errors for any reason at all,
 320      * including a timeout, we skip the turn.
 321      */
 322     pkt->i_buffer = 0;
 323
 324 out:
 325     if (pkt->i_buffer == 0) {
 326       block_Release(pkt);
 327       pkt = NULL;
 328     }
 329
 330     vlc_interrupt_unregister();
 331
 332     /* Re-add the socket to the poll if we were interrupted */
 333     vlc_mutex_lock( &p_sys->lock );
 334     if ( p_sys->b_interrupted )
 335     {
 336         srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
 337             &(int) { SRT_EPOLL_ERR | SRT_EPOLL_IN } );
 338         p_sys->b_interrupted = false;
 339     }
 340     vlc_mutex_unlock( &p_sys->lock );
 341
 342     return pkt;
 343 }
 344
 345 static int Open(vlc_object_t *p_this)
 346 {
 347     stream_t     *p_stream = (stream_t*)p_this;
 348     stream_sys_t *p_sys = NULL;
 349     vlc_url_t     parsed_url = { 0 };
 350
 351     p_sys = vlc_obj_calloc( p_this, 1, sizeof( *p_sys ) );
 352     if( unlikely( p_sys == NULL ) )
 353         return VLC_ENOMEM;
 354
 355     srt_startup();
 356
 357     vlc_mutex_init( &p_sys->lock );
 358
 359     p_stream->p_sys = p_sys;
 360
 361     if ( vlc_UrlParse( &parsed_url, p_stream->psz_url ) == -1 )
 362     {
 363         msg_Err( p_stream, "Failed to parse input URL (%s)",
 364             p_stream->psz_url );
 365         goto failed;
 366     }
 367
 368     p_sys->psz_host = vlc_obj_strdup( p_this, parsed_url.psz_host );
 369     p_sys->i_port = parsed_url.i_port;
 370
 371     vlc_UrlClean( &parsed_url );
 372
 373     p_sys->i_poll_id = srt_epoll_create();
 374     if ( p_sys->i_poll_id == -1 )
 375     {
 376         msg_Err( p_stream, "Failed to create poll id for SRT socket." );
 377         goto failed;
 378     }
 379
 380     if ( !srt_schedule_reconnect( p_stream ) )
 381     {
 382         msg_Err( p_stream, "Failed to schedule connect");
 383
 384         goto failed;
 385     }
 386
 387     p_stream->pf_block = BlockSRT;
 388     p_stream->pf_control = Control;
 389
 390     return VLC_SUCCESS;
 391
 392 failed:
 393     if ( p_sys->sock != -1 ) srt_close( p_sys->sock );
 394     if ( p_sys->i_poll_id != -1 ) srt_epoll_release( p_sys->i_poll_id );
 395
 396     return VLC_EGENERIC;
 397 }
 398
 399 static void Close(vlc_object_t *p_this)
 400 {
 401     stream_t     *p_stream = (stream_t*)p_this;
 402     stream_sys_t *p_sys = p_stream->p_sys;
 403
 404     srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 405     srt_close( p_sys->sock );
 406     srt_epoll_release( p_sys->i_poll_id );
 407
 408     srt_cleanup();
 409 }
 410
 411 /* Module descriptor */
 412 vlc_module_begin ()
 413     set_shortname( N_( "SRT" ) )
 414     set_description( N_( "SRT input" ) )
 415     set_category( CAT_INPUT )
 416     set_subcategory( SUBCAT_INPUT_ACCESS )
 417
 418     add_integer( SRT_PARAM_CHUNK_SIZE, SRT_DEFAULT_CHUNK_SIZE,
 419             N_( "SRT chunk size (bytes)" ), NULL, true )
 420     add_integer( SRT_PARAM_POLL_TIMEOUT, SRT_DEFAULT_POLL_TIMEOUT,
 421             N_( "Return poll wait after timeout milliseconds (-1 = infinite)" ),
 422             NULL, true )
 423     add_integer( SRT_PARAM_LATENCY, SRT_DEFAULT_LATENCY,
 424             N_( "SRT latency (ms)" ), NULL, true )
 425     add_password( SRT_PARAM_PASSPHRASE, "",
 426             N_( "Password for stream encryption" ), NULL )
 427     add_integer( SRT_PARAM_PAYLOAD_SIZE, SRT_DEFAULT_PAYLOAD_SIZE,
 428             N_( "SRT maximum payload size (bytes)" ), NULL, true )
 429     add_integer( SRT_PARAM_KEY_LENGTH, SRT_DEFAULT_KEY_LENGTH,
 430             SRT_KEY_LENGTH_TEXT, SRT_KEY_LENGTH_TEXT, false )
 431     change_integer_list( srt_key_lengths, srt_key_length_names )
 432
 433     set_capability("access", 0)
 434     add_shortcut("srt")
 435
 436     set_callbacks(Open, Close)
 437 vlc_module_end ()