chiark / gitweb /
stop disorder(1) permuting command line
[disorder] / clients / playrtp.c
1 /*
2  * This file is part of DisOrder.
3  * Copyright (C) 2007 Richard Kettlewell
4  *
5  * This program is free software; you can redistribute it and/or modify
6  * it under the terms of the GNU General Public License as published by
7  * the Free Software Foundation; either version 2 of the License, or
8  * (at your option) any later version.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License for more details.
14  *
15  * You should have received a copy of the GNU General Public License
16  * along with this program; if not, write to the Free Software
17  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
18  * USA
19  */
20 /** @file clients/playrtp.c
21  * @brief RTP player
22  *
23  * This player supports Linux (<a href="http://www.alsa-project.org/">ALSA</a>)
24  * and Apple Mac (<a
25  * href="http://developer.apple.com/audio/coreaudio.html">Core Audio</a>)
26  * systems.  There is no support for Microsoft Windows yet, and that will in
27  * fact probably an entirely separate program.
28  *
29  * The program runs (at least) three threads.  listen_thread() is responsible
30  * for reading RTP packets off the wire and adding them to the linked list @ref
31  * received_packets, assuming they are basically sound.  queue_thread() takes
32  * packets off this linked list and adds them to @ref packets (an operation
33  * which might be much slower due to contention for @ref lock).
34  *
35  * The main thread is responsible for actually playing audio.  In ALSA this
36  * means it waits until ALSA says it's ready for more audio which it then
37  * plays.  See @ref clients/playrtp-alsa.c.
38  *
39  * In Core Audio the main thread is only responsible for starting and stopping
40  * play: the system does the actual playback in its own private thread, and
41  * calls adioproc() to fetch the audio data.  See @ref
42  * clients/playrtp-coreaudio.c.
43  *
44  * Sometimes it happens that there is no audio available to play.  This may
45  * because the server went away, or a packet was dropped, or the server
46  * deliberately did not send any sound because it encountered a silence.
47  *
48  * Assumptions:
49  * - it is safe to read uint32_t values without a lock protecting them
50  */
51
52 #include <config.h>
53 #include "types.h"
54
55 #include <getopt.h>
56 #include <stdio.h>
57 #include <stdlib.h>
58 #include <sys/socket.h>
59 #include <sys/types.h>
60 #include <sys/socket.h>
61 #include <netdb.h>
62 #include <pthread.h>
63 #include <locale.h>
64 #include <sys/uio.h>
65 #include <string.h>
66 #include <assert.h>
67 #include <errno.h>
68 #include <netinet/in.h>
69 #include <sys/time.h>
70 #include <sys/un.h>
71 #include <unistd.h>
72 #include <sys/mman.h>
73 #include <fcntl.h>
74
75 #include "log.h"
76 #include "mem.h"
77 #include "configuration.h"
78 #include "addr.h"
79 #include "syscalls.h"
80 #include "rtp.h"
81 #include "defs.h"
82 #include "vector.h"
83 #include "heap.h"
84 #include "timeval.h"
85 #include "client.h"
86 #include "playrtp.h"
87 #include "inputline.h"
88
89 #define readahead linux_headers_are_borked
90
91 /** @brief Obsolete synonym */
92 #ifndef IPV6_JOIN_GROUP
93 # define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
94 #endif
95
96 /** @brief RTP socket */
97 static int rtpfd;
98
99 /** @brief Log output */
100 static FILE *logfp;
101
102 /** @brief Output device */
103 const char *device;
104
105 /** @brief Minimum low watermark
106  *
107  * We'll stop playing if there's only this many samples in the buffer. */
108 unsigned minbuffer = 2 * 44100 / 10;  /* 0.2 seconds */
109
110 /** @brief Buffer high watermark
111  *
112  * We'll only start playing when this many samples are available. */
113 static unsigned readahead = 2 * 2 * 44100;
114
115 /** @brief Maximum buffer size
116  *
117  * We'll stop reading from the network if we have this many samples. */
118 static unsigned maxbuffer;
119
120 /** @brief Received packets
121  * Protected by @ref receive_lock
122  *
123  * Received packets are added to this list, and queue_thread() picks them off
124  * it and adds them to @ref packets.  Whenever a packet is added to it, @ref
125  * receive_cond is signalled.
126  */
127 struct packet *received_packets;
128
129 /** @brief Tail of @ref received_packets
130  * Protected by @ref receive_lock
131  */
132 struct packet **received_tail = &received_packets;
133
134 /** @brief Lock protecting @ref received_packets 
135  *
136  * Only listen_thread() and queue_thread() ever hold this lock.  It is vital
137  * that queue_thread() not hold it any longer than it strictly has to. */
138 pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
139
140 /** @brief Condition variable signalled when @ref received_packets is updated
141  *
142  * Used by listen_thread() to notify queue_thread() that it has added another
143  * packet to @ref received_packets. */
144 pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
145
146 /** @brief Length of @ref received_packets */
147 uint32_t nreceived;
148
149 /** @brief Binary heap of received packets */
150 struct pheap packets;
151
152 /** @brief Total number of samples available
153  *
154  * We make this volatile because we inspect it without a protecting lock,
155  * so the usual pthread_* guarantees aren't available.
156  */
157 volatile uint32_t nsamples;
158
159 /** @brief Timestamp of next packet to play.
160  *
161  * This is set to the timestamp of the last packet, plus the number of
162  * samples it contained.  Only valid if @ref active is nonzero.
163  */
164 uint32_t next_timestamp;
165
166 /** @brief True if actively playing
167  *
168  * This is true when playing and false when just buffering. */
169 int active;
170
171 /** @brief Lock protecting @ref packets */
172 pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
173
174 /** @brief Condition variable signalled whenever @ref packets is changed */
175 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
176
177 #if HAVE_ALSA_ASOUNDLIB_H
178 # define DEFAULT_BACKEND playrtp_alsa
179 #elif HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
180 # define DEFAULT_BACKEND playrtp_oss
181 #elif HAVE_COREAUDIO_AUDIOHARDWARE_H
182 # define DEFAULT_BACKEND playrtp_coreaudio
183 #else
184 # error No known backend
185 #endif
186
187 /** @brief Backend to play with */
188 static void (*backend)(void) = &DEFAULT_BACKEND;
189
190 HEAP_DEFINE(pheap, struct packet *, lt_packet);
191
192 /** @brief Control socket or NULL */
193 const char *control_socket;
194
195 /** @brief Buffer for debugging dump
196  *
197  * The debug dump is enabled by the @c --dump option.  It records the last 20s
198  * of audio to the specified file (which will be about 3.5Mbytes).  The file is
199  * written as as ring buffer, so the start point will progress through it.
200  *
201  * Use clients/dump2wav to convert this to a WAV file, which can then be loaded
202  * into (e.g.) Audacity for further inspection.
203  *
204  * All three backends (ALSA, OSS, Core Audio) now support this option.
205  *
206  * The idea is to allow the user a few seconds to react to an audible artefact.
207  */
208 int16_t *dump_buffer;
209
210 /** @brief Current index within debugging dump */
211 size_t dump_index;
212
213 /** @brief Size of debugging dump in samples */
214 size_t dump_size = 44100/*Hz*/ * 2/*channels*/ * 20/*seconds*/;
215
216 static const struct option options[] = {
217   { "help", no_argument, 0, 'h' },
218   { "version", no_argument, 0, 'V' },
219   { "debug", no_argument, 0, 'd' },
220   { "device", required_argument, 0, 'D' },
221   { "min", required_argument, 0, 'm' },
222   { "max", required_argument, 0, 'x' },
223   { "buffer", required_argument, 0, 'b' },
224   { "rcvbuf", required_argument, 0, 'R' },
225 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
226   { "oss", no_argument, 0, 'o' },
227 #endif
228 #if HAVE_ALSA_ASOUNDLIB_H
229   { "alsa", no_argument, 0, 'a' },
230 #endif
231 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
232   { "core-audio", no_argument, 0, 'c' },
233 #endif
234   { "dump", required_argument, 0, 'r' },
235   { "socket", required_argument, 0, 's' },
236   { "config", required_argument, 0, 'C' },
237   { 0, 0, 0, 0 }
238 };
239
240 /** @brief Control thread
241  *
242  * This thread is responsible for accepting control commands from Disobedience
243  * (or other controllers) over an AF_UNIX stream socket with a path specified
244  * by the @c --socket option.  The protocol uses simple string commands and
245  * replies:
246  *
247  * - @c stop will shut the player down
248  * - @c query will send back the reply @c running
249  * - anything else is ignored
250  *
251  * Commands and response strings terminated by shutting down the connection or
252  * by a newline.  No attempt is made to multiplex multiple clients so it is
253  * important that the command be sent as soon as the connection is made - it is
254  * assumed that both parties to the protocol are entirely cooperating with one
255  * another.
256  */
257 static void *control_thread(void attribute((unused)) *arg) {
258   struct sockaddr_un sa;
259   int sfd, cfd;
260   char *line;
261   socklen_t salen;
262   FILE *fp;
263
264   assert(control_socket);
265   unlink(control_socket);
266   memset(&sa, 0, sizeof sa);
267   sa.sun_family = AF_UNIX;
268   strcpy(sa.sun_path, control_socket);
269   sfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
270   if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0)
271     fatal(errno, "error binding to %s", control_socket);
272   if(listen(sfd, 128) < 0)
273     fatal(errno, "error calling listen on %s", control_socket);
274   info("listening on %s", control_socket);
275   for(;;) {
276     salen = sizeof sa;
277     cfd = accept(sfd, (struct sockaddr *)&sa, &salen);
278     if(cfd < 0) {
279       switch(errno) {
280       case EINTR:
281       case EAGAIN:
282         break;
283       default:
284         fatal(errno, "error calling accept on %s", control_socket);
285       }
286     }
287     if(!(fp = fdopen(cfd, "r+"))) {
288       error(errno, "error calling fdopen for %s connection", control_socket);
289       close(cfd);
290       continue;
291     }
292     if(!inputline(control_socket, fp, &line, '\n')) {
293       if(!strcmp(line, "stop")) {
294         info("stopped via %s", control_socket);
295         exit(0);                          /* terminate immediately */
296       }
297       if(!strcmp(line, "query"))
298         fprintf(fp, "running");
299       xfree(line);
300     }
301     if(fclose(fp) < 0)
302       error(errno, "error closing %s connection", control_socket);
303   }
304 }
305
306 /** @brief Drop the first packet
307  *
308  * Assumes that @ref lock is held. 
309  */
310 static void drop_first_packet(void) {
311   if(pheap_count(&packets)) {
312     struct packet *const p = pheap_remove(&packets);
313     nsamples -= p->nsamples;
314     playrtp_free_packet(p);
315     pthread_cond_broadcast(&cond);
316   }
317 }
318
319 /** @brief Background thread adding packets to heap
320  *
321  * This just transfers packets from @ref received_packets to @ref packets.  It
322  * is important that it holds @ref receive_lock for as little time as possible,
323  * in order to minimize the interval between calls to read() in
324  * listen_thread().
325  */
326 static void *queue_thread(void attribute((unused)) *arg) {
327   struct packet *p;
328
329   for(;;) {
330     /* Get the next packet */
331     pthread_mutex_lock(&receive_lock);
332     while(!received_packets)
333       pthread_cond_wait(&receive_cond, &receive_lock);
334     p = received_packets;
335     received_packets = p->next;
336     if(!received_packets)
337       received_tail = &received_packets;
338     --nreceived;
339     pthread_mutex_unlock(&receive_lock);
340     /* Add it to the heap */
341     pthread_mutex_lock(&lock);
342     pheap_insert(&packets, p);
343     nsamples += p->nsamples;
344     pthread_cond_broadcast(&cond);
345     pthread_mutex_unlock(&lock);
346   }
347 }
348
349 /** @brief Background thread collecting samples
350  *
351  * This function collects samples, perhaps converts them to the target format,
352  * and adds them to the packet list.
353  *
354  * It is crucial that the gap between successive calls to read() is as small as
355  * possible: otherwise packets will be dropped.
356  *
357  * We use a binary heap to ensure that the unavoidable effort is at worst
358  * logarithmic in the total number of packets - in fact if packets are mostly
359  * received in order then we will largely do constant work per packet since the
360  * newest packet will always be last.
361  *
362  * Of more concern is that we must acquire the lock on the heap to add a packet
363  * to it.  If this proves a problem in practice then the answer would be
364  * (probably doubly) linked list with new packets added the end and a second
365  * thread which reads packets off the list and adds them to the heap.
366  *
367  * We keep memory allocation (mostly) very fast by keeping pre-allocated
368  * packets around; see @ref playrtp_new_packet().
369  */
370 static void *listen_thread(void attribute((unused)) *arg) {
371   struct packet *p = 0;
372   int n;
373   struct rtp_header header;
374   uint16_t seq;
375   uint32_t timestamp;
376   struct iovec iov[2];
377
378   for(;;) {
379     if(!p)
380       p = playrtp_new_packet();
381     iov[0].iov_base = &header;
382     iov[0].iov_len = sizeof header;
383     iov[1].iov_base = p->samples_raw;
384     iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
385     n = readv(rtpfd, iov, 2);
386     if(n < 0) {
387       switch(errno) {
388       case EINTR:
389         continue;
390       default:
391         fatal(errno, "error reading from socket");
392       }
393     }
394     /* Ignore too-short packets */
395     if((size_t)n <= sizeof (struct rtp_header)) {
396       info("ignored a short packet");
397       continue;
398     }
399     timestamp = htonl(header.timestamp);
400     seq = htons(header.seq);
401     /* Ignore packets in the past */
402     if(active && lt(timestamp, next_timestamp)) {
403       info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32,
404            timestamp, next_timestamp);
405       continue;
406     }
407     p->next = 0;
408     p->flags = 0;
409     p->timestamp = timestamp;
410     /* Convert to target format */
411     if(header.mpt & 0x80)
412       p->flags |= IDLE;
413     switch(header.mpt & 0x7F) {
414     case 10:
415       p->nsamples = (n - sizeof header) / sizeof(uint16_t);
416       break;
417       /* TODO support other RFC3551 media types (when the speaker does) */
418     default:
419       fatal(0, "unsupported RTP payload type %d",
420             header.mpt & 0x7F);
421     }
422     if(logfp)
423       fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n",
424               seq, timestamp, p->nsamples, timestamp + p->nsamples);
425     /* Stop reading if we've reached the maximum.
426      *
427      * This is rather unsatisfactory: it means that if packets get heavily
428      * out of order then we guarantee dropouts.  But for now... */
429     if(nsamples >= maxbuffer) {
430       pthread_mutex_lock(&lock);
431       while(nsamples >= maxbuffer)
432         pthread_cond_wait(&cond, &lock);
433       pthread_mutex_unlock(&lock);
434     }
435     /* Add the packet to the receive queue */
436     pthread_mutex_lock(&receive_lock);
437     *received_tail = p;
438     received_tail = &p->next;
439     ++nreceived;
440     pthread_cond_signal(&receive_cond);
441     pthread_mutex_unlock(&receive_lock);
442     /* We'll need a new packet */
443     p = 0;
444   }
445 }
446
447 /** @brief Wait until the buffer is adequately full
448  *
449  * Must be called with @ref lock held.
450  */
451 void playrtp_fill_buffer(void) {
452   while(nsamples)
453     drop_first_packet();
454   info("Buffering...");
455   while(nsamples < readahead)
456     pthread_cond_wait(&cond, &lock);
457   next_timestamp = pheap_first(&packets)->timestamp;
458   active = 1;
459 }
460
461 /** @brief Find next packet
462  * @return Packet to play or NULL if none found
463  *
464  * The return packet is merely guaranteed not to be in the past: it might be
465  * the first packet in the future rather than one that is actually suitable to
466  * play.
467  *
468  * Must be called with @ref lock held.
469  */
470 struct packet *playrtp_next_packet(void) {
471   while(pheap_count(&packets)) {
472     struct packet *const p = pheap_first(&packets);
473     if(le(p->timestamp + p->nsamples, next_timestamp)) {
474       /* This packet is in the past.  Drop it and try another one. */
475       drop_first_packet();
476     } else
477       /* This packet is NOT in the past.  (It might be in the future
478        * however.) */
479       return p;
480   }
481   return 0;
482 }
483
484 /** @brief Play an RTP stream
485  *
486  * This is the guts of the program.  It is responsible for:
487  * - starting the listening thread
488  * - opening the audio device
489  * - reading ahead to build up a buffer
490  * - arranging for audio to be played
491  * - detecting when the buffer has got too small and re-buffering
492  */
493 static void play_rtp(void) {
494   pthread_t ltid;
495   int err;
496
497   /* We receive and convert audio data in a background thread */
498   if((err = pthread_create(&ltid, 0, listen_thread, 0)))
499     fatal(err, "pthread_create listen_thread");
500   /* We have a second thread to add received packets to the queue */
501   if((err = pthread_create(&ltid, 0, queue_thread, 0)))
502     fatal(err, "pthread_create queue_thread");
503   /* The rest of the work is backend-specific */
504   backend();
505 }
506
507 /* display usage message and terminate */
508 static void help(void) {
509   xprintf("Usage:\n"
510           "  disorder-playrtp [OPTIONS] ADDRESS [PORT]\n"
511           "Options:\n"
512           "  --device, -D DEVICE     Output device\n"
513           "  --min, -m FRAMES        Buffer low water mark\n"
514           "  --buffer, -b FRAMES     Buffer high water mark\n"
515           "  --max, -x FRAMES        Buffer maximum size\n"
516           "  --rcvbuf, -R BYTES      Socket receive buffer size\n"
517           "  --config, -C PATH       Set configuration file\n"
518 #if HAVE_ALSA_ASOUNDLIB_H
519           "  --alsa, -a              Use ALSA to play audio\n"
520 #endif
521 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
522           "  --oss, -o               Use OSS to play audio\n"
523 #endif
524 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
525           "  --core-audio, -c        Use Core Audio to play audio\n"
526 #endif
527           "  --help, -h              Display usage message\n"
528           "  --version, -V           Display version number\n"
529           );
530   xfclose(stdout);
531   exit(0);
532 }
533
534 /* display version number and terminate */
535 static void version(void) {
536   xprintf("%s", disorder_version_string);
537   xfclose(stdout);
538   exit(0);
539 }
540
541 int main(int argc, char **argv) {
542   int n, err;
543   struct addrinfo *res;
544   struct stringlist sl;
545   char *sockname;
546   int rcvbuf, target_rcvbuf = 131072;
547   socklen_t len;
548   struct ip_mreq mreq;
549   struct ipv6_mreq mreq6;
550   disorder_client *c;
551   char *address, *port;
552   int is_multicast;
553   union any_sockaddr {
554     struct sockaddr sa;
555     struct sockaddr_in in;
556     struct sockaddr_in6 in6;
557   };
558   union any_sockaddr mgroup;
559   const char *dumpfile = 0;
560
561   static const struct addrinfo prefs = {
562     AI_PASSIVE,
563     PF_INET,
564     SOCK_DGRAM,
565     IPPROTO_UDP,
566     0,
567     0,
568     0,
569     0
570   };
571
572   mem_init();
573   if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale");
574   while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:r", options, 0)) >= 0) {
575     switch(n) {
576     case 'h': help();
577     case 'V': version();
578     case 'd': debugging = 1; break;
579     case 'D': device = optarg; break;
580     case 'm': minbuffer = 2 * atol(optarg); break;
581     case 'b': readahead = 2 * atol(optarg); break;
582     case 'x': maxbuffer = 2 * atol(optarg); break;
583     case 'L': logfp = fopen(optarg, "w"); break;
584     case 'R': target_rcvbuf = atoi(optarg); break;
585 #if HAVE_ALSA_ASOUNDLIB_H
586     case 'a': backend = playrtp_alsa; break;
587 #endif
588 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
589     case 'o': backend = playrtp_oss; break;
590 #endif
591 #if HAVE_COREAUDIO_AUDIOHARDWARE_H      
592     case 'c': backend = playrtp_coreaudio; break;
593 #endif
594     case 'C': configfile = optarg; break;
595     case 's': control_socket = optarg; break;
596     case 'r': dumpfile = optarg; break;
597     default: fatal(0, "invalid option");
598     }
599   }
600   if(config_read(0)) fatal(0, "cannot read configuration");
601   if(!maxbuffer)
602     maxbuffer = 4 * readahead;
603   argc -= optind;
604   argv += optind;
605   switch(argc) {
606   case 0:
607     /* Get configuration from server */
608     if(!(c = disorder_new(1))) exit(EXIT_FAILURE);
609     if(disorder_connect(c)) exit(EXIT_FAILURE);
610     if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE);
611     sl.n = 2;
612     sl.s = xcalloc(2, sizeof *sl.s);
613     sl.s[0] = address;
614     sl.s[1] = port;
615     break;
616   case 1:
617   case 2:
618     /* Use command-line ADDRESS+PORT or just PORT */
619     sl.n = argc;
620     sl.s = argv;
621     break;
622   default:
623     fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]");
624   }
625   /* Look up address and port */
626   if(!(res = get_address(&sl, &prefs, &sockname)))
627     exit(1);
628   /* Create the socket */
629   if((rtpfd = socket(res->ai_family,
630                      res->ai_socktype,
631                      res->ai_protocol)) < 0)
632     fatal(errno, "error creating socket");
633   /* Stash the multicast group address */
634   if((is_multicast = multicast(res->ai_addr))) {
635     memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
636     switch(res->ai_addr->sa_family) {
637     case AF_INET:
638       mgroup.in.sin_port = 0;
639       break;
640     case AF_INET6:
641       mgroup.in6.sin6_port = 0;
642       break;
643     }
644   }
645   /* Bind to 0/port */
646   switch(res->ai_addr->sa_family) {
647   case AF_INET:
648     memset(&((struct sockaddr_in *)res->ai_addr)->sin_addr, 0,
649            sizeof (struct in_addr));
650     break;
651   case AF_INET6:
652     memset(&((struct sockaddr_in6 *)res->ai_addr)->sin6_addr, 0,
653            sizeof (struct in6_addr));
654     break;
655   default:
656     fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
657   }
658   if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
659     fatal(errno, "error binding socket to %s", sockname);
660   if(is_multicast) {
661     switch(mgroup.sa.sa_family) {
662     case PF_INET:
663       mreq.imr_multiaddr = mgroup.in.sin_addr;
664       mreq.imr_interface.s_addr = 0;      /* use primary interface */
665       if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
666                     &mreq, sizeof mreq) < 0)
667         fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
668       break;
669     case PF_INET6:
670       mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
671       memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
672       if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
673                     &mreq6, sizeof mreq6) < 0)
674         fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
675       break;
676     default:
677       fatal(0, "unsupported address family %d", res->ai_family);
678     }
679     info("listening on %s multicast group %s",
680          format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
681   } else
682     info("listening on %s", format_sockaddr(res->ai_addr));
683   len = sizeof rcvbuf;
684   if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
685     fatal(errno, "error calling getsockopt SO_RCVBUF");
686   if(target_rcvbuf > rcvbuf) {
687     if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
688                   &target_rcvbuf, sizeof target_rcvbuf) < 0)
689       error(errno, "error calling setsockopt SO_RCVBUF %d", 
690             target_rcvbuf);
691       /* We try to carry on anyway */
692     else
693       info("changed socket receive buffer from %d to %d",
694            rcvbuf, target_rcvbuf);
695   } else
696     info("default socket receive buffer %d", rcvbuf);
697   if(logfp)
698     info("WARNING: -L option can impact performance");
699   if(control_socket) {
700     pthread_t tid;
701
702     if((err = pthread_create(&tid, 0, control_thread, 0)))
703       fatal(err, "pthread_create control_thread");
704   }
705   if(dumpfile) {
706     int fd;
707     unsigned char buffer[65536];
708     size_t written;
709
710     if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0)
711       fatal(errno, "opening %s", dumpfile);
712     /* Fill with 0s to a suitable size */
713     memset(buffer, 0, sizeof buffer);
714     for(written = 0; written < dump_size * sizeof(int16_t);
715         written += sizeof buffer) {
716       if(write(fd, buffer, sizeof buffer) < 0)
717         fatal(errno, "clearing %s", dumpfile);
718     }
719     /* Map the buffer into memory for convenience */
720     dump_buffer = mmap(0, dump_size * sizeof(int16_t), PROT_READ|PROT_WRITE,
721                        MAP_SHARED, fd, 0);
722     if(dump_buffer == (void *)-1)
723       fatal(errno, "mapping %s", dumpfile);
724     info("dumping to %s", dumpfile);
725   }
726   play_rtp();
727   return 0;
728 }
729
730 /*
731 Local Variables:
732 c-basic-offset:2
733 comment-column:40
734 fill-column:79
735 indent-tabs-mode:nil
736 End:
737 */