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