access_out: srt: refactor to support connection recovery
[vlc.git] / modules / access_output / srt.c
1 /*****************************************************************************
2  * srt.c: SRT (Secure Reliable Transport) output 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
25 #ifdef HAVE_CONFIG_H
26 # include "config.h"
27 #endif
28
29 #include <vlc_common.h>
30 #include <vlc_interrupt.h>
31 #include <vlc_fs.h>
32 #include <vlc_plugin.h>
33 #include <vlc_sout.h>
34 #include <vlc_block.h>
35 #include <vlc_network.h>
36
37 #include <srt/srt.h>
38
39 /* libsrt defines default packet size as 1316 internally
40  * so srt module takes same value. */
41 #define SRT_DEFAULT_CHUNK_SIZE 1316
42 /* libsrt tutorial uses 9000 as a default binding port */
43 #define SRT_DEFAULT_PORT 9000
44 /* The default timeout is -1 (infinite) */
45 #define SRT_DEFAULT_POLL_TIMEOUT 100
46 /* The default latency is 125
47  * which uses srt library internally */
48 #define SRT_DEFAULT_LATENCY 125
49 /* Crypto key length in bytes. */
50 #define SRT_KEY_LENGTH_TEXT N_("Crypto key length in bytes")
51 #define SRT_DEFAULT_KEY_LENGTH 16
52 static const int srt_key_lengths[] = {
53     16, 24, 32,
54 };
55
56 static const char *const srt_key_length_names[] = {
57     N_("16 bytes"), N_("24 bytes"), N_("32 bytes"),
58 };
59
60 struct sout_access_out_sys_t
61 {
62     SRTSOCKET     sock;
63     int           i_poll_id;
64     bool          b_interrupted;
65     vlc_mutex_t   lock;
66 };
67
68 static void srt_wait_interrupted(void *p_data)
69 {
70     sout_access_out_t *p_access = p_data;
71     sout_access_out_sys_t *p_sys = p_access->p_sys;
72
73     vlc_mutex_lock( &p_sys->lock );
74     if ( p_sys->i_poll_id >= 0 &&  p_sys->sock != SRT_INVALID_SOCK )
75     {
76         p_sys->b_interrupted = true;
77
78         msg_Dbg( p_access, "Waking up srt_epoll_wait");
79
80         /* Removing all socket descriptors from the monitoring list
81          * wakes up SRT's threads. We only have one to remove. */
82         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
83     }
84     vlc_mutex_unlock( &p_sys->lock );
85 }
86
87 static bool srt_schedule_reconnect(sout_access_out_t *p_access)
88 {
89     char                    *psz_dst_addr = NULL;
90     int                      i_dst_port;
91     int                      i_latency;
92     int                      stat;
93     char                    *psz_passphrase = NULL;
94
95     struct addrinfo hints = {
96         .ai_socktype = SOCK_DGRAM,
97     }, *res = NULL;
98
99     sout_access_out_sys_t *p_sys = p_access->p_sys;
100     bool failed = false;
101
102     psz_passphrase = var_InheritString( p_access, "passphrase" );
103
104     i_dst_port = SRT_DEFAULT_PORT;
105     char *psz_parser = psz_dst_addr = strdup( p_access->psz_path );
106     if( !psz_dst_addr )
107     {
108         failed = true;
109         goto out;
110     }
111
112     if ( psz_parser[0] == '[' )
113         psz_parser = strchr( psz_parser, ']' );
114
115     psz_parser = strchr( psz_parser ? psz_parser : psz_dst_addr, ':' );
116     if ( psz_parser != NULL )
117     {
118         *psz_parser++ = '\0';
119         i_dst_port = atoi( psz_parser );
120     }
121
122     stat = vlc_getaddrinfo( psz_dst_addr, i_dst_port, &hints, &res );
123     if ( stat )
124     {
125         msg_Err( p_access, "Cannot resolve [%s]:%d (reason: %s)",
126                  psz_dst_addr,
127                  i_dst_port,
128                  gai_strerror( stat ) );
129
130         failed = true;
131         goto out;
132     }
133
134     /* Always start with a fresh socket */
135     if ( p_sys->sock != SRT_INVALID_SOCK )
136     {
137         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
138         srt_close( p_sys->sock );
139     }
140
141     p_sys->sock = srt_socket( res->ai_family, SOCK_DGRAM, 0 );
142     if ( p_sys->sock == SRT_INVALID_SOCK )
143     {
144         msg_Err( p_access, "Failed to open socket." );
145         failed = true;
146         goto out;
147     }
148
149     /* Make SRT non-blocking */
150     srt_setsockopt( p_sys->sock, 0, SRTO_SNDSYN,
151         &(bool) { false }, sizeof( bool ) );
152     srt_setsockopt( p_sys->sock, 0, SRTO_RCVSYN,
153         &(bool) { false }, sizeof( bool ) );
154
155     /* Make sure TSBPD mode is enable (SRT mode) */
156     srt_setsockopt( p_sys->sock, 0, SRTO_TSBPDMODE,
157         &(int) { 1 }, sizeof( int ) );
158
159     /* This is an access_out so it is always a sender */
160     srt_setsockopt( p_sys->sock, 0, SRTO_SENDER,
161         &(int) { 1 }, sizeof( int ) );
162
163     /* Set latency */
164     i_latency = var_InheritInteger( p_access, "latency" );
165     srt_setsockopt( p_sys->sock, 0, SRTO_TSBPDDELAY,
166         &i_latency, sizeof( int ) );
167
168     if ( psz_passphrase != NULL && psz_passphrase[0] != '\0')
169     {
170         int i_key_length = var_InheritInteger( p_access, "key-length" );
171         srt_setsockopt( p_sys->sock, 0, SRTO_PASSPHRASE,
172             psz_passphrase, strlen( psz_passphrase ) );
173         srt_setsockopt( p_sys->sock, 0, SRTO_PBKEYLEN,
174             &i_key_length, sizeof( int ) );
175     }
176
177     srt_setsockopt( p_sys->sock, 0, SRTO_SENDER, &(int) { 1 }, sizeof(int) );
178
179     srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
180         &(int) { SRT_EPOLL_ERR | SRT_EPOLL_OUT });
181
182     /* Schedule a connect */
183     msg_Dbg( p_access, "Schedule SRT connect (dest addresss: %s, port: %d).",
184         psz_dst_addr, i_dst_port );
185
186     stat = srt_connect( p_sys->sock, res->ai_addr, res->ai_addrlen );
187     if ( stat == SRT_ERROR )
188     {
189         msg_Err( p_access, "Failed to connect to server (reason: %s)",
190                  srt_getlasterror_str() );
191         failed = true;
192     }
193
194 out:
195     if (failed && p_sys->sock != SRT_INVALID_SOCK)
196     {
197         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
198         srt_close(p_sys->sock);
199         p_sys->sock = SRT_INVALID_SOCK;
200     }
201
202     free( psz_passphrase );
203     free( psz_dst_addr );
204     freeaddrinfo( res );
205
206     return !failed;
207 }
208
209 static ssize_t Write( sout_access_out_t *p_access, block_t *p_buffer )
210 {
211     sout_access_out_sys_t *p_sys = p_access->p_sys;
212     int i_len = 0;
213     size_t i_chunk_size = var_InheritInteger( p_access, "chunk-size" );
214     int i_poll_timeout = var_InheritInteger( p_access, "poll-timeout" );
215     bool b_interrupted = false;
216
217     vlc_interrupt_register( srt_wait_interrupted, p_access);
218
219     while( p_buffer )
220     {
221         block_t *p_next;
222
223         i_len += p_buffer->i_buffer;
224
225         while( p_buffer->i_buffer )
226         {
227             if ( vlc_killed() )
228             {
229                 /* We are told to stop. Stop. */
230                 i_len = VLC_EGENERIC;
231                 goto out;
232             }
233
234             switch( srt_getsockstate( p_sys->sock ) )
235             {
236                 case SRTS_CONNECTED:
237                     /* Good to go */
238                     break;
239                 case SRTS_BROKEN:
240                 case SRTS_NONEXIST:
241                 case SRTS_CLOSED:
242                     /* Failed. Schedule recovery. */
243                     if ( !srt_schedule_reconnect( p_access ) )
244                         msg_Err( p_access, "Failed to schedule connect");
245                     /* Fall-through */
246                 default:
247                     /* Not ready */
248                     i_len = VLC_EGENERIC;
249                     goto out;
250             }
251
252             SRTSOCKET ready[1];
253             int readycnt = 1;
254             if ( srt_epoll_wait( p_sys->i_poll_id,
255                 0, 0, &ready[0], &readycnt,
256                 i_poll_timeout, NULL, 0, NULL, 0 ) < 0)
257             {
258                 if ( vlc_killed() )
259                 {
260                     /* We are told to stop. Stop. */
261                     i_len = VLC_EGENERIC;
262                     goto out;
263                 }
264
265                 /* if 'srt_epoll_wait' is interrupted, we still need to
266                 *  finish sending current block or it may be sent only
267                 *  partially. TODO: this delay can be prevented,
268                 *  possibly with a FIFO and an additional thread.
269                 */
270                 vlc_mutex_lock( &p_sys->lock );
271                 if ( p_sys->b_interrupted )
272                 {
273                     srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
274                         &(int) { SRT_EPOLL_ERR | SRT_EPOLL_OUT });
275                     p_sys->b_interrupted = false;
276                     b_interrupted = true;
277                 }
278                 vlc_mutex_unlock( &p_sys->lock );
279
280                 if ( !b_interrupted )
281                 {
282                     continue;
283                 }
284                 else if ( (true) )
285                 {
286                     msg_Dbg( p_access, "srt_epoll_wait was interrupted");
287                 }
288             }
289
290             if ( readycnt > 0  && ready[0] == p_sys->sock
291                 && srt_getsockstate( p_sys->sock ) == SRTS_CONNECTED)
292             {
293                 size_t i_write = __MIN( p_buffer->i_buffer, i_chunk_size );
294                 if (srt_sendmsg2( p_sys->sock,
295                     (char *)p_buffer->p_buffer, i_write, 0 ) == SRT_ERROR )
296                 {
297                     msg_Warn( p_access, "send error: %s", srt_getlasterror_str() );
298                     i_len = VLC_EGENERIC;
299                     goto out;
300                 }
301
302                 p_buffer->p_buffer += i_write;
303                 p_buffer->i_buffer -= i_write;
304             }
305         }
306
307         p_next = p_buffer->p_next;
308         block_Release( p_buffer );
309         p_buffer = p_next;
310
311         if ( b_interrupted )
312         {
313             goto out;
314         }
315     }
316
317 out:
318     vlc_interrupt_unregister();
319
320     /* Re-add the socket to the poll if we were interrupted */
321     vlc_mutex_lock( &p_sys->lock );
322     if ( p_sys->b_interrupted )
323     {
324         srt_epoll_add_usock( p_sys->i_poll_id, p_sys->sock,
325             &(int) { SRT_EPOLL_ERR | SRT_EPOLL_OUT } );
326         p_sys->b_interrupted = false;
327     }
328     vlc_mutex_unlock( &p_sys->lock );
329
330     if ( i_len <= 0 ) block_ChainRelease( p_buffer );
331     return i_len;
332 }
333
334 static int Control( sout_access_out_t *p_access, int i_query, va_list args )
335 {
336     VLC_UNUSED( p_access );
337
338     int i_ret = VLC_SUCCESS;
339
340     switch( i_query )
341     {
342         case ACCESS_OUT_CONTROLS_PACE:
343             *va_arg( args, bool * ) = false;
344             break;
345
346         default:
347             i_ret = VLC_EGENERIC;
348             break;
349     }
350
351     return i_ret;
352 }
353
354 static int Open( vlc_object_t *p_this )
355 {
356     sout_access_out_t       *p_access = (sout_access_out_t*)p_this;
357     sout_access_out_sys_t   *p_sys = NULL;
358
359     if (var_Create ( p_access, "dst-port", VLC_VAR_INTEGER )
360      || var_Create ( p_access, "src-port", VLC_VAR_INTEGER )
361      || var_Create ( p_access, "dst-addr", VLC_VAR_STRING )
362      || var_Create ( p_access, "src-addr", VLC_VAR_STRING ) )
363     {
364          msg_Err( p_access, "Valid network information is required." );
365         return VLC_ENOMEM;
366     }
367
368     p_sys = vlc_obj_calloc( p_this, 1, sizeof( *p_sys ) );
369     if( unlikely( p_sys == NULL ) )
370         return VLC_ENOMEM;
371
372     srt_startup();
373
374     vlc_mutex_init( &p_sys->lock );
375
376     p_access->p_sys = p_sys;
377
378     p_sys->i_poll_id = srt_epoll_create();
379     if ( p_sys->i_poll_id == -1 )
380     {
381         msg_Err( p_access, "Failed to create poll id for SRT socket (reason: %s)",
382                  srt_getlasterror_str() );
383
384         goto failed;
385     }
386
387     if ( !srt_schedule_reconnect( p_access ) )
388     {
389         msg_Err( p_access, "Failed to schedule connect");
390
391         goto failed;
392     }
393
394     p_access->pf_write = Write;
395     p_access->pf_control = Control;
396
397     return VLC_SUCCESS;
398
399 failed:
400     vlc_mutex_destroy( &p_sys->lock );
401
402     if ( p_sys != NULL )
403     {
404         if ( p_sys->sock != -1 ) srt_close( p_sys->sock );
405         if ( p_sys->i_poll_id != -1 ) srt_epoll_release( p_sys->i_poll_id );
406
407         vlc_obj_free( p_this, p_sys );
408         p_access->p_sys = NULL;
409     }
410
411     return VLC_EGENERIC;
412 }
413
414 static void Close( vlc_object_t * p_this )
415 {
416     sout_access_out_t     *p_access = (sout_access_out_t*)p_this;
417     sout_access_out_sys_t *p_sys = p_access->p_sys;
418
419     if ( p_sys )
420     {
421         vlc_mutex_destroy( &p_sys->lock );
422
423         srt_epoll_remove_usock( p_sys->i_poll_id, p_sys->sock );
424         srt_close( p_sys->sock );
425         srt_epoll_release( p_sys->i_poll_id );
426
427         vlc_obj_free( p_this, p_sys );
428         p_access->p_sys = NULL;
429     }
430
431     srt_cleanup();
432 }
433
434 /* Module descriptor */
435 vlc_module_begin()
436     set_shortname( N_("SRT") )
437     set_description( N_("SRT stream output") )
438     set_category( CAT_SOUT )
439     set_subcategory( SUBCAT_SOUT_ACO )
440
441     add_integer( "chunk-size", SRT_DEFAULT_CHUNK_SIZE,
442             N_("SRT chunk size (bytes)"), NULL, true )
443     add_integer( "poll-timeout", SRT_DEFAULT_POLL_TIMEOUT,
444             N_("Return poll wait after timeout milliseconds (-1 = infinite)"), NULL, true )
445     add_integer( "latency", SRT_DEFAULT_LATENCY, N_("SRT latency (ms)"), NULL, true )
446     add_password( "passphrase", "", N_("Password for stream encryption"), NULL, false )
447     add_integer( "key-length", SRT_DEFAULT_KEY_LENGTH,
448             SRT_KEY_LENGTH_TEXT, SRT_KEY_LENGTH_TEXT, false )
449         change_integer_list( srt_key_lengths, srt_key_length_names )
450
451     set_capability( "sout access", 0 )
452     add_shortcut( "srt" )
453
454     set_callbacks( Open, Close )
455 vlc_module_end ()