]> git.sesse.net Git - vlc/blob - modules/access_output/udp.c
2733660eebc34e92dfc5d3aa596cb522455f8cfd
[vlc] / modules / access_output / udp.c
1 /*****************************************************************************
2  * udp.c
3  *****************************************************************************
4  * Copyright (C) 2001-2007 the VideoLAN team
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
11  * it under the terms of the GNU General Public License as published by
12  * the Free Software Foundation; either version 2 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 General Public License for more details.
19  *
20  * You should have received a copy of the GNU General Public License
21  * along with this program; if not, write to the Free Software
22  * Foundation, 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 <sys/stat.h>
37 #include <fcntl.h>
38 #include <assert.h>
39
40 #include <vlc_sout.h>
41 #include <vlc_block.h>
42
43 #ifdef HAVE_UNISTD_H
44 #   include <unistd.h>
45 #endif
46
47 #ifdef WIN32
48 #   include <winsock2.h>
49 #   include <ws2tcpip.h>
50 #else
51 #   include <sys/socket.h>
52 #endif
53
54 #include <vlc_network.h>
55
56 #define MAX_EMPTY_BLOCKS 200
57
58 /*****************************************************************************
59  * Module descriptor
60  *****************************************************************************/
61 static int  Open ( vlc_object_t * );
62 static void Close( vlc_object_t * );
63
64 #define SOUT_CFG_PREFIX "sout-udp-"
65
66 #define CACHING_TEXT N_("Caching value (ms)")
67 #define CACHING_LONGTEXT N_( \
68     "Default caching value for outbound UDP streams. This " \
69     "value should be set in milliseconds." )
70
71 #define GROUP_TEXT N_("Group packets")
72 #define GROUP_LONGTEXT N_("Packets can be sent one by one at the right time " \
73                           "or by groups. You can choose the number " \
74                           "of packets that will be sent at a time. It " \
75                           "helps reducing the scheduling load on " \
76                           "heavily-loaded systems." )
77
78 vlc_module_begin ()
79     set_description( N_("UDP stream output") )
80     set_shortname( "UDP" )
81     set_category( CAT_SOUT )
82     set_subcategory( SUBCAT_SOUT_ACO )
83     add_integer( SOUT_CFG_PREFIX "caching", DEFAULT_PTS_DELAY / 1000, NULL, CACHING_TEXT, CACHING_LONGTEXT, true )
84     add_integer( SOUT_CFG_PREFIX "group", 1, NULL, GROUP_TEXT, GROUP_LONGTEXT,
85                                  true )
86     add_obsolete_integer( SOUT_CFG_PREFIX "late" )
87     add_obsolete_bool( SOUT_CFG_PREFIX "raw" )
88
89     set_capability( "sout access", 0 )
90     add_shortcut( "udp" )
91     set_callbacks( Open, Close )
92 vlc_module_end ()
93
94 /*****************************************************************************
95  * Exported prototypes
96  *****************************************************************************/
97
98 static const char *const ppsz_sout_options[] = {
99     "caching",
100     "group",
101     NULL
102 };
103
104 /* Options handled by the libvlc network core */
105 static const char *const ppsz_core_options[] = {
106     "dscp",
107     "ttl",
108     "miface",
109     "miface-addr",
110     NULL
111 };
112
113 static ssize_t Write   ( sout_access_out_t *, block_t * );
114 static int  Seek    ( sout_access_out_t *, off_t  );
115 static int Control( sout_access_out_t *, int, va_list );
116
117 static void* ThreadWrite( void * );
118 static block_t *NewUDPPacket( sout_access_out_t *, mtime_t );
119
120 struct sout_access_out_sys_t
121 {
122     mtime_t       i_caching;
123     int           i_handle;
124     bool          b_mtu_warning;
125     size_t        i_mtu;
126
127     block_fifo_t *p_fifo;
128     block_fifo_t *p_empty_blocks;
129     block_t      *p_buffer;
130
131     vlc_thread_t  thread;
132 };
133
134 #define DEFAULT_PORT 1234
135
136 /*****************************************************************************
137  * Open: open the file
138  *****************************************************************************/
139 static int Open( vlc_object_t *p_this )
140 {
141     sout_access_out_t       *p_access = (sout_access_out_t*)p_this;
142     sout_access_out_sys_t   *p_sys;
143
144     char                *psz_dst_addr = NULL;
145     int                 i_dst_port;
146
147     int                 i_handle;
148
149     config_ChainParse( p_access, SOUT_CFG_PREFIX,
150                        ppsz_sout_options, p_access->p_cfg );
151     config_ChainParse( p_access, "",
152                        ppsz_core_options, p_access->p_cfg );
153
154     if (var_Create (p_access, "dst-port", VLC_VAR_INTEGER)
155      || var_Create (p_access, "src-port", VLC_VAR_INTEGER)
156      || var_Create (p_access, "dst-addr", VLC_VAR_STRING)
157      || var_Create (p_access, "src-addr", VLC_VAR_STRING))
158     {
159         return VLC_ENOMEM;
160     }
161
162     if( !( p_sys = malloc ( sizeof( *p_sys ) ) ) )
163         return VLC_ENOMEM;
164     p_access->p_sys = p_sys;
165
166     i_dst_port = DEFAULT_PORT;
167     char *psz_parser = psz_dst_addr = strdup( p_access->psz_path );
168     if( !psz_dst_addr )
169     {
170         free( p_sys );
171         return VLC_ENOMEM;
172     }
173
174     if (psz_parser[0] == '[')
175         psz_parser = strchr (psz_parser, ']');
176
177     psz_parser = strchr (psz_parser ? psz_parser : psz_dst_addr, ':');
178     if (psz_parser != NULL)
179     {
180         *psz_parser++ = '\0';
181         i_dst_port = atoi (psz_parser);
182     }
183
184     i_handle = net_ConnectDgram( p_this, psz_dst_addr, i_dst_port, -1,
185                                  IPPROTO_UDP );
186     free (psz_dst_addr);
187
188     if( i_handle == -1 )
189     {
190          msg_Err( p_access, "failed to create raw UDP socket" );
191          free (p_sys);
192          return VLC_EGENERIC;
193     }
194     else
195     {
196         char addr[NI_MAXNUMERICHOST];
197         int port;
198
199         if (net_GetSockAddress (i_handle, addr, &port) == 0)
200         {
201             msg_Dbg (p_access, "source: %s port %d", addr, port);
202             var_SetString (p_access, "src-addr", addr);
203             var_SetInteger (p_access, "src-port", port);
204         }
205
206         if (net_GetPeerAddress (i_handle, addr, &port) == 0)
207         {
208             msg_Dbg (p_access, "destination: %s port %d", addr, port);
209             var_SetString (p_access, "dst-addr", addr);
210             var_SetInteger (p_access, "dst-port", port);
211         }
212     }
213     shutdown( i_handle, SHUT_RD );
214
215     p_sys->i_caching = UINT64_C(1000)
216                      * var_GetInteger( p_access, SOUT_CFG_PREFIX "caching");
217     p_sys->i_handle = i_handle;
218     p_sys->i_mtu = var_CreateGetInteger( p_this, "mtu" );
219     p_sys->b_mtu_warning = false;
220     p_sys->p_fifo = block_FifoNew();
221     p_sys->p_empty_blocks = block_FifoNew();
222     p_sys->p_buffer = NULL;
223
224     if( vlc_clone( &p_sys->thread, ThreadWrite, p_access,
225                            VLC_THREAD_PRIORITY_HIGHEST ) )
226     {
227         msg_Err( p_access, "cannot spawn sout access thread" );
228         block_FifoRelease( p_sys->p_fifo );
229         block_FifoRelease( p_sys->p_empty_blocks );
230         net_Close (i_handle);
231         free (p_sys);
232         return VLC_EGENERIC;
233     }
234
235     p_access->pf_write = Write;
236     p_access->pf_seek = Seek;
237     p_access->pf_control = Control;
238
239     return VLC_SUCCESS;
240 }
241
242 /*****************************************************************************
243  * Close: close the target
244  *****************************************************************************/
245 static void Close( vlc_object_t * p_this )
246 {
247     sout_access_out_t     *p_access = (sout_access_out_t*)p_this;
248     sout_access_out_sys_t *p_sys = p_access->p_sys;
249
250     vlc_cancel( p_sys->thread );
251     vlc_join( p_sys->thread, NULL );
252     block_FifoRelease( p_sys->p_fifo );
253     block_FifoRelease( p_sys->p_empty_blocks );
254
255     if( p_sys->p_buffer ) block_Release( p_sys->p_buffer );
256
257     net_Close( p_sys->i_handle );
258     free( p_sys );
259 }
260
261 static int Control( sout_access_out_t *p_access, int i_query, va_list args )
262 {
263     (void)p_access;
264
265     switch( i_query )
266     {
267         case ACCESS_OUT_CONTROLS_PACE:
268             *va_arg( args, bool * ) = false;
269             break;
270
271         default:
272             return VLC_EGENERIC;
273     }
274     return VLC_SUCCESS;
275 }
276
277 /*****************************************************************************
278  * Write: standard write on a file descriptor.
279  *****************************************************************************/
280 static ssize_t Write( sout_access_out_t *p_access, block_t *p_buffer )
281 {
282     sout_access_out_sys_t *p_sys = p_access->p_sys;
283     int i_len = 0;
284
285     while( p_buffer )
286     {
287         block_t *p_next;
288         int i_packets = 0;
289         mtime_t now = mdate();
290
291         if( !p_sys->b_mtu_warning && p_buffer->i_buffer > p_sys->i_mtu )
292         {
293             msg_Warn( p_access, "packet size > MTU, you should probably "
294                       "increase the MTU" );
295             p_sys->b_mtu_warning = true;
296         }
297
298         /* Check if there is enough space in the buffer */
299         if( p_sys->p_buffer &&
300             p_sys->p_buffer->i_buffer + p_buffer->i_buffer > p_sys->i_mtu )
301         {
302             if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
303             {
304                 msg_Dbg( p_access, "late packet for UDP input (%"PRId64 ")",
305                          now - p_sys->p_buffer->i_dts
306                           - p_sys->i_caching );
307             }
308             block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
309             p_sys->p_buffer = NULL;
310         }
311
312         i_len += p_buffer->i_buffer;
313         while( p_buffer->i_buffer )
314         {
315             size_t i_payload_size = p_sys->i_mtu;
316             size_t i_write = __MIN( p_buffer->i_buffer, i_payload_size );
317
318             i_packets++;
319
320             if( !p_sys->p_buffer )
321             {
322                 p_sys->p_buffer = NewUDPPacket( p_access, p_buffer->i_dts );
323                 if( !p_sys->p_buffer ) break;
324             }
325
326             memcpy( p_sys->p_buffer->p_buffer + p_sys->p_buffer->i_buffer,
327                     p_buffer->p_buffer, i_write );
328
329             p_sys->p_buffer->i_buffer += i_write;
330             p_buffer->p_buffer += i_write;
331             p_buffer->i_buffer -= i_write;
332             if ( p_buffer->i_flags & BLOCK_FLAG_CLOCK )
333             {
334                 if ( p_sys->p_buffer->i_flags & BLOCK_FLAG_CLOCK )
335                     msg_Warn( p_access, "putting two PCRs at once" );
336                 p_sys->p_buffer->i_flags |= BLOCK_FLAG_CLOCK;
337             }
338
339             if( p_sys->p_buffer->i_buffer == p_sys->i_mtu || i_packets > 1 )
340             {
341                 /* Flush */
342                 if( p_sys->p_buffer->i_dts + p_sys->i_caching < now )
343                 {
344                     msg_Dbg( p_access, "late packet for udp input (%"PRId64 ")",
345                              mdate() - p_sys->p_buffer->i_dts
346                               - p_sys->i_caching );
347                 }
348                 block_FifoPut( p_sys->p_fifo, p_sys->p_buffer );
349                 p_sys->p_buffer = NULL;
350             }
351         }
352
353         p_next = p_buffer->p_next;
354         block_Release( p_buffer );
355         p_buffer = p_next;
356     }
357
358     return i_len;
359 }
360
361 /*****************************************************************************
362  * Seek: seek to a specific location in a file
363  *****************************************************************************/
364 static int Seek( sout_access_out_t *p_access, off_t i_pos )
365 {
366     (void) i_pos;
367     msg_Err( p_access, "UDP sout access cannot seek" );
368     return -1;
369 }
370
371 /*****************************************************************************
372  * NewUDPPacket: allocate a new UDP packet of size p_sys->i_mtu
373  *****************************************************************************/
374 static block_t *NewUDPPacket( sout_access_out_t *p_access, mtime_t i_dts)
375 {
376     sout_access_out_sys_t *p_sys = p_access->p_sys;
377     block_t *p_buffer;
378
379     while ( block_FifoCount( p_sys->p_empty_blocks ) > MAX_EMPTY_BLOCKS )
380     {
381         p_buffer = block_FifoGet( p_sys->p_empty_blocks );
382         block_Release( p_buffer );
383     }
384
385     if( block_FifoCount( p_sys->p_empty_blocks ) == 0 )
386     {
387         p_buffer = block_Alloc( p_sys->i_mtu );
388     }
389     else
390     {
391         p_buffer = block_FifoGet(p_sys->p_empty_blocks );
392         p_buffer->i_flags = 0;
393         p_buffer = block_Realloc( p_buffer, 0, p_sys->i_mtu );
394     }
395
396     p_buffer->i_dts = i_dts;
397     p_buffer->i_buffer = 0;
398
399     return p_buffer;
400 }
401
402 /*****************************************************************************
403  * ThreadWrite: Write a packet on the network at the good time.
404  *****************************************************************************/
405 static void* ThreadWrite( void *data )
406 {
407     sout_access_out_t *p_access = data;
408     sout_access_out_sys_t *p_sys = p_access->p_sys;
409     mtime_t i_date_last = -1;
410     const unsigned i_group = var_GetInteger( p_access,
411                                              SOUT_CFG_PREFIX "group" );
412     mtime_t i_to_send = i_group;
413     unsigned i_dropped_packets = 0;
414
415     for (;;)
416     {
417         block_t *p_pk = block_FifoGet( p_sys->p_fifo );
418         mtime_t       i_date, i_sent;
419
420         i_date = p_sys->i_caching + p_pk->i_dts;
421         if( i_date_last > 0 )
422         {
423             if( i_date - i_date_last > 2000000 )
424             {
425                 if( !i_dropped_packets )
426                     msg_Dbg( p_access, "mmh, hole (%"PRId64" > 2s) -> drop",
427                              i_date - i_date_last );
428
429                 block_FifoPut( p_sys->p_empty_blocks, p_pk );
430
431                 i_date_last = i_date;
432                 i_dropped_packets++;
433                 continue;
434             }
435             else if( i_date - i_date_last < -1000 )
436             {
437                 if( !i_dropped_packets )
438                     msg_Dbg( p_access, "mmh, packets in the past (%"PRId64")",
439                              i_date_last - i_date );
440             }
441         }
442
443         block_cleanup_push( p_pk );
444         i_to_send--;
445         if( !i_to_send || (p_pk->i_flags & BLOCK_FLAG_CLOCK) )
446         {
447             mwait( i_date );
448             i_to_send = i_group;
449         }
450         if ( send( p_sys->i_handle, p_pk->p_buffer, p_pk->i_buffer, 0 ) == -1 )
451             msg_Warn( p_access, "send error: %m" );
452         vlc_cleanup_pop();
453
454         if( i_dropped_packets )
455         {
456             msg_Dbg( p_access, "dropped %i packets", i_dropped_packets );
457             i_dropped_packets = 0;
458         }
459
460 #if 1
461         i_sent = mdate();
462         if ( i_sent > i_date + 20000 )
463         {
464             msg_Dbg( p_access, "packet has been sent too late (%"PRId64 ")",
465                      i_sent - i_date );
466         }
467 #endif
468
469         block_FifoPut( p_sys->p_empty_blocks, p_pk );
470
471         i_date_last = i_date;
472     }
473     return NULL;
474 }