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 *psz_streamid = var_InheritString( p_stream, SRT_PARAM_STREAMID );
 101     bool streamid_needs_free = true;
 102     char *url = NULL;
 103     srt_params_t params;
 104     struct addrinfo hints = {
 105         .ai_socktype = SOCK_DGRAM,
 106     }, *res = NULL;
 107
 108     stream_sys_t *p_sys = p_stream->p_sys;
 109     bool failed = false;
 110
 111     stat = vlc_getaddrinfo( p_sys->psz_host, p_sys->i_port, &hints, &res );
 112     if ( stat )
 113     {
 114         msg_Err( p_stream, "Cannot resolve [%s]:%d (reason: %s)",
 115                  p_sys->psz_host,
 116                  p_sys->i_port,
 117                  gai_strerror( stat ) );
 118
 119         failed = true;
 120         goto out;
 121     }
 122
 123     /* Always start with a fresh socket */
 124     if (p_sys->sock != SRT_INVALID_SOCK)
 125     {
 126         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 127         srt_close( p_sys->sock );
 128     }
 129
 130     p_sys->sock = srt_socket( res->ai_family, SOCK_DGRAM, 0 );
 131     if ( p_sys->sock == SRT_INVALID_SOCK )
 132     {
 133         msg_Err( p_stream, "Failed to open socket." );
 134         failed = true;
 135         goto out;
 136     }
 137
 138     if (p_stream->psz_url) {
 139         url = strdup( p_stream->psz_url );
 140         if (srt_parse_url( url, &params )) {
 141             if (params.latency != -1)
 142                 i_latency = params.latency;
 143             if (params.payload_size != -1)
 144                 i_payload_size = params.payload_size;
 145             if (params.passphrase != NULL) {
 146                 free( psz_passphrase );
 147                 passphrase_needs_free = false;
 148                 psz_passphrase = (char *) params.passphrase;
 149             }
 150             if (params.streamid != NULL ) {
 151                 free( psz_streamid );
 152                 streamid_needs_free = false;
 153                 psz_streamid = (char *) params.streamid;
 154             }
 155         }
 156     }
 157
 158     /* Make SRT non-blocking */
 159     srt_setsockopt( p_sys->sock, 0, SRTO_SNDSYN,
 160         &(bool) { false }, sizeof( bool ) );
 161     srt_setsockopt( p_sys->sock, 0, SRTO_RCVSYN,
 162         &(bool) { false }, sizeof( bool ) );
 163
 164     /* Make sure TSBPD mode is enable (SRT mode) */
 165     srt_setsockopt( p_sys->sock, 0, SRTO_TSBPDMODE,
 166         &(int) { 1 }, sizeof( int ) );
 167
 168     /* This is an access module so it is always a receiver */
 169     srt_setsockopt( p_sys->sock, 0, SRTO_SENDER,
 170         &(int) { 0 }, sizeof( int ) );
 171
 172     /* Set latency */
 173     srt_set_socket_option( strm_obj, SRT_PARAM_LATENCY, p_sys->sock,
 174             SRTO_LATENCY, &i_latency, sizeof(i_latency) );
 175
 176     /* set passphrase */
 177     if (psz_passphrase != NULL && psz_passphrase[0] != '\0') {
 178         int i_key_length = var_InheritInteger( p_stream, SRT_PARAM_KEY_LENGTH );
 179
 180         srt_set_socket_option( strm_obj, SRT_PARAM_KEY_LENGTH, p_sys->sock,
 181                 SRTO_PBKEYLEN, &i_key_length, sizeof(i_key_length) );
 182
 183         srt_set_socket_option( strm_obj, SRT_PARAM_PASSPHRASE, p_sys->sock,
 184                 SRTO_PASSPHRASE, psz_passphrase, strlen(psz_passphrase) );
 185     }
 186
 187     /* set stream id */
 188     if (psz_streamid != NULL && psz_streamid[0] != '\0') {
 189         srt_set_socket_option( strm_obj, SRT_PARAM_STREAMID, p_sys->sock,
 190                 SRTO_STREAMID, psz_streamid, strlen(psz_streamid) );
 191     }
 192
 193     /* set maximum payload size */
 194     srt_set_socket_option( strm_obj, SRT_PARAM_PAYLOAD_SIZE, p_sys->sock,
 195             SRTO_PAYLOADSIZE, &i_payload_size, sizeof(i_payload_size) );
 196
 197
 198     srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
 199         &(int) { SRT_EPOLL_ERR | SRT_EPOLL_IN });
 200
 201     /* Schedule a connect */
 202     msg_Dbg( p_stream, "Schedule SRT connect (dest addresss: %s, port: %d).",
 203         p_sys->psz_host, p_sys->i_port);
 204
 205     stat = srt_connect( p_sys->sock, res->ai_addr, res->ai_addrlen );
 206     if (stat == SRT_ERROR) {
 207         msg_Err( p_stream, "Failed to connect to server (reason: %s)",
 208                 srt_getlasterror_str() );
 209         failed = true;
 210     }
 211
 212     /* Reset the number of chunks to allocate as the bitrate of
 213      * the stream may have changed.
 214      */
 215     p_sys->i_chunks = SRT_MIN_CHUNKS_TRYREAD;
 216
 217 out:
 218     if (failed && p_sys->sock != SRT_INVALID_SOCK)
 219     {
 220         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 221         srt_close(p_sys->sock);
 222         p_sys->sock = SRT_INVALID_SOCK;
 223     }
 224
 225     if (passphrase_needs_free)
 226         free( psz_passphrase );
 227     if (streamid_needs_free)
 228         free( psz_streamid );
 229     freeaddrinfo( res );
 230     free( url );
 231
 232     return !failed;
 233 }
 234
 235 static block_t *BlockSRT(stream_t *p_stream, bool *restrict eof)
 236 {
 237     stream_sys_t *p_sys = p_stream->p_sys;
 238     int i_chunk_size = var_InheritInteger( p_stream, SRT_PARAM_CHUNK_SIZE );
 239     int i_poll_timeout = var_InheritInteger( p_stream, SRT_PARAM_POLL_TIMEOUT );
 240     /* SRT doesn't have a concept of EOF for live streams. */
 241     VLC_UNUSED(eof);
 242
 243     if ( vlc_killed() )
 244     {
 245         /* We are told to stop. Stop. */
 246         return NULL;
 247     }
 248
 249     if ( p_sys->i_chunks == 0 )
 250         p_sys->i_chunks = SRT_MIN_CHUNKS_TRYREAD;
 251
 252     size_t i_chunk_size_actual = ( i_chunk_size > 0 )
 253         ? i_chunk_size : SRT_DEFAULT_CHUNK_SIZE;
 254     size_t bufsize = i_chunk_size_actual * p_sys->i_chunks;
 255     block_t *pkt = block_Alloc( bufsize );
 256     if ( unlikely( pkt == NULL ) )
 257     {
 258         return NULL;
 259     }
 260
 261     vlc_interrupt_register( srt_wait_interrupted, p_stream);
 262
 263     SRTSOCKET ready[1];
 264     int readycnt = 1;
 265     while ( srt_epoll_wait( p_sys->i_poll_id,
 266         ready, &readycnt, 0, 0,
 267         i_poll_timeout, NULL, 0, NULL, 0 ) >= 0)
 268     {
 269         if ( readycnt < 0  || ready[0] != p_sys->sock )
 270         {
 271             /* should never happen, force recovery */
 272             srt_close(p_sys->sock);
 273             p_sys->sock = SRT_INVALID_SOCK;
 274         }
 275
 276         switch( srt_getsockstate( p_sys->sock ) )
 277         {
 278             case SRTS_CONNECTED:
 279                 /* Good to go */
 280                 break;
 281             case SRTS_BROKEN:
 282             case SRTS_NONEXIST:
 283             case SRTS_CLOSED:
 284                 /* Failed. Schedule recovery. */
 285                 if ( !srt_schedule_reconnect( p_stream ) )
 286                     msg_Err( p_stream, "Failed to schedule connect" );
 287                 /* Fall-through */
 288             default:
 289                 /* Not ready */
 290                 continue;
 291         }
 292
 293         /* Try to get as much data as possible out of the lib, if there
 294          * is still some left, increase the number of chunks to read so that
 295          * it will read faster on the next iteration. This way the buffer will
 296          * grow until it reads fast enough to keep the library empty after
 297          * each iteration.
 298          */
 299         pkt->i_buffer = 0;
 300         while ( ( bufsize - pkt->i_buffer ) >= i_chunk_size_actual )
 301         {
 302             int stat = srt_recvmsg( p_sys->sock,
 303                 (char *)( pkt->p_buffer + pkt->i_buffer ),
 304                 bufsize - pkt->i_buffer );
 305             if ( stat <= 0 )
 306             {
 307                 break;
 308             }
 309             pkt->i_buffer += (size_t)stat;
 310         }
 311
 312         msg_Dbg ( p_stream, "Read %zu bytes out of a max of %zu"
 313             " (%d chunks of %zu bytes)", pkt->i_buffer,
 314             p_sys->i_chunks * i_chunk_size_actual, p_sys->i_chunks,
 315                  i_chunk_size_actual );
 316
 317
 318         /* Gradually adjust number of chunks we read at a time
 319         * up to a predefined maximum. The actual number we might
 320         * settle on depends on stream's bit rate.
 321         */
 322         size_t rem = bufsize - pkt->i_buffer;
 323         if ( rem < i_chunk_size_actual )
 324         {
 325             if ( p_sys->i_chunks < SRT_MAX_CHUNKS_TRYREAD )
 326             {
 327                 p_sys->i_chunks++;
 328             }
 329         }
 330
 331         goto out;
 332     }
 333
 334     /* if the poll reports errors for any reason at all,
 335      * including a timeout, we skip the turn.
 336      */
 337     pkt->i_buffer = 0;
 338
 339 out:
 340     if (pkt->i_buffer == 0) {
 341       block_Release(pkt);
 342       pkt = NULL;
 343     }
 344
 345     vlc_interrupt_unregister();
 346
 347     /* Re-add the socket to the poll if we were interrupted */
 348     vlc_mutex_lock( &p_sys->lock );
 349     if ( p_sys->b_interrupted )
 350     {
 351         srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
 352             &(int) { SRT_EPOLL_ERR | SRT_EPOLL_IN } );
 353         p_sys->b_interrupted = false;
 354     }
 355     vlc_mutex_unlock( &p_sys->lock );
 356
 357     return pkt;
 358 }
 359
 360 static int Open(vlc_object_t *p_this)
 361 {
 362     stream_t     *p_stream = (stream_t*)p_this;
 363     stream_sys_t *p_sys = NULL;
 364     vlc_url_t     parsed_url = { 0 };
 365
 366     p_sys = vlc_obj_calloc( p_this, 1, sizeof( *p_sys ) );
 367     if( unlikely( p_sys == NULL ) )
 368         return VLC_ENOMEM;
 369
 370     srt_startup();
 371
 372     vlc_mutex_init( &p_sys->lock );
 373
 374     p_stream->p_sys = p_sys;
 375
 376     if ( vlc_UrlParse( &parsed_url, p_stream->psz_url ) == -1 )
 377     {
 378         msg_Err( p_stream, "Failed to parse input URL (%s)",
 379             p_stream->psz_url );
 380         goto failed;
 381     }
 382
 383     p_sys->psz_host = vlc_obj_strdup( p_this, parsed_url.psz_host );
 384     p_sys->i_port = parsed_url.i_port;
 385
 386     vlc_UrlClean( &parsed_url );
 387
 388     p_sys->i_poll_id = srt_epoll_create();
 389     if ( p_sys->i_poll_id == -1 )
 390     {
 391         msg_Err( p_stream, "Failed to create poll id for SRT socket." );
 392         goto failed;
 393     }
 394
 395     if ( !srt_schedule_reconnect( p_stream ) )
 396     {
 397         msg_Err( p_stream, "Failed to schedule connect");
 398
 399         goto failed;
 400     }
 401
 402     p_stream->pf_block = BlockSRT;
 403     p_stream->pf_control = Control;
 404
 405     return VLC_SUCCESS;
 406
 407 failed:
 408     if ( p_sys->sock != -1 ) srt_close( p_sys->sock );
 409     if ( p_sys->i_poll_id != -1 ) srt_epoll_release( p_sys->i_poll_id );
 410
 411     return VLC_EGENERIC;
 412 }
 413
 414 static void Close(vlc_object_t *p_this)
 415 {
 416     stream_t     *p_stream = (stream_t*)p_this;
 417     stream_sys_t *p_sys = p_stream->p_sys;
 418
 419     srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
 420     srt_close( p_sys->sock );
 421     srt_epoll_release( p_sys->i_poll_id );
 422
 423     srt_cleanup();
 424 }
 425
 426 /* Module descriptor */
 427 vlc_module_begin ()
 428     set_shortname( N_( "SRT" ) )
 429     set_description( N_( "SRT input" ) )
 430     set_category( CAT_INPUT )
 431     set_subcategory( SUBCAT_INPUT_ACCESS )
 432
 433     add_integer( SRT_PARAM_CHUNK_SIZE, SRT_DEFAULT_CHUNK_SIZE,
 434             N_( "SRT chunk size (bytes)" ), NULL, true )
 435     add_integer( SRT_PARAM_POLL_TIMEOUT, SRT_DEFAULT_POLL_TIMEOUT,
 436             N_( "Return poll wait after timeout milliseconds (-1 = infinite)" ),
 437             NULL, true )
 438     add_integer( SRT_PARAM_LATENCY, SRT_DEFAULT_LATENCY,
 439             N_( "SRT latency (ms)" ), NULL, true )
 440     add_password( SRT_PARAM_PASSPHRASE, "",
 441             N_( "Password for stream encryption" ), NULL )
 442     add_integer( SRT_PARAM_PAYLOAD_SIZE, SRT_DEFAULT_PAYLOAD_SIZE,
 443             N_( "SRT maximum payload size (bytes)" ), NULL, true )
 444     add_integer( SRT_PARAM_KEY_LENGTH, SRT_DEFAULT_KEY_LENGTH,
 445             SRT_KEY_LENGTH_TEXT, SRT_KEY_LENGTH_TEXT, false )
 446     change_integer_list( srt_key_lengths, srt_key_length_names )
 447     add_string(SRT_PARAM_STREAMID, "",
 448             N_(" SRT Stream ID"), NULL, false)
 449     change_safe()
 450
 451     set_capability("access", 0)
 452     add_shortcut("srt")
 453
 454     set_callbacks(Open, Close)
 455 vlc_module_end ()