srt: add stream ID option
[vlc.git] / 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 ()