chiark / gitweb /
c9005f87d4dcc712e1cfc8f2210d8330c6f470ad
[disorder] / clients / playrtp.c
1 /*
2  * This file is part of DisOrder.
3  * Copyright (C) 2007-2009 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 3 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,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU 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, see <http://www.gnu.org/licenses/>.
17  */
18 /** @file clients/playrtp.c
19  * @brief RTP player
20  *
21  * This player supports Linux (<a href="http://www.alsa-project.org/">ALSA</a>)
22  * and Apple Mac (<a
23  * href="http://developer.apple.com/audio/coreaudio.html">Core Audio</a>)
24  * systems.  There is no support for Microsoft Windows yet, and that will in
25  * fact probably an entirely separate program.
26  *
27  * The program runs (at least) three threads:
28  *
29  * listen_thread() is responsible for reading RTP packets off the wire and
30  * adding them to the linked list @ref received_packets, assuming they are
31  * basically sound.
32  *
33  * queue_thread() takes packets off this linked list and adds them to @ref
34  * packets (an operation which might be much slower due to contention for @ref
35  * lock).
36  *
37  * control_thread() accepts commands from Disobedience (or anything else).
38  *
39  * The main thread activates and deactivates audio playing via the @ref
40  * lib/uaudio.h API (which probably implies at least one further thread).
41  *
42  * Sometimes it happens that there is no audio available to play.  This may
43  * because the server went away, or a packet was dropped, or the server
44  * deliberately did not send any sound because it encountered a silence.
45  *
46  * Assumptions:
47  * - it is safe to read uint32_t values without a lock protecting them
48  */
49
50 #include "common.h"
51
52 #include <getopt.h>
53 #include <sys/socket.h>
54 #include <sys/types.h>
55 #include <sys/socket.h>
56 #include <netdb.h>
57 #include <pthread.h>
58 #include <locale.h>
59 #include <sys/uio.h>
60 #include <errno.h>
61 #include <netinet/in.h>
62 #include <sys/time.h>
63 #include <sys/un.h>
64 #include <unistd.h>
65 #include <sys/mman.h>
66 #include <fcntl.h>
67 #include <math.h>
68
69 #include "log.h"
70 #include "mem.h"
71 #include "configuration.h"
72 #include "addr.h"
73 #include "syscalls.h"
74 #include "rtp.h"
75 #include "defs.h"
76 #include "vector.h"
77 #include "heap.h"
78 #include "timeval.h"
79 #include "client.h"
80 #include "playrtp.h"
81 #include "inputline.h"
82 #include "version.h"
83 #include "uaudio.h"
84
85 /** @brief Obsolete synonym */
86 #ifndef IPV6_JOIN_GROUP
87 # define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
88 #endif
89
90 /** @brief RTP socket */
91 static int rtpfd;
92
93 /** @brief Log output */
94 static FILE *logfp;
95
96 /** @brief Output device */
97
98 /** @brief Buffer low watermark in samples */
99 unsigned minbuffer = 4 * (2 * 44100) / 10;  /* 0.4 seconds */
100
101 /** @brief Maximum buffer size in samples
102  *
103  * We'll stop reading from the network if we have this many samples.
104  */
105 static unsigned maxbuffer;
106
107 /** @brief Received packets
108  * Protected by @ref receive_lock
109  *
110  * Received packets are added to this list, and queue_thread() picks them off
111  * it and adds them to @ref packets.  Whenever a packet is added to it, @ref
112  * receive_cond is signalled.
113  */
114 struct packet *received_packets;
115
116 /** @brief Tail of @ref received_packets
117  * Protected by @ref receive_lock
118  */
119 struct packet **received_tail = &received_packets;
120
121 /** @brief Lock protecting @ref received_packets 
122  *
123  * Only listen_thread() and queue_thread() ever hold this lock.  It is vital
124  * that queue_thread() not hold it any longer than it strictly has to. */
125 pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
126
127 /** @brief Condition variable signalled when @ref received_packets is updated
128  *
129  * Used by listen_thread() to notify queue_thread() that it has added another
130  * packet to @ref received_packets. */
131 pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
132
133 /** @brief Length of @ref received_packets */
134 uint32_t nreceived;
135
136 /** @brief Binary heap of received packets */
137 struct pheap packets;
138
139 /** @brief Total number of samples available
140  *
141  * We make this volatile because we inspect it without a protecting lock,
142  * so the usual pthread_* guarantees aren't available.
143  */
144 volatile uint32_t nsamples;
145
146 /** @brief Timestamp of next packet to play.
147  *
148  * This is set to the timestamp of the last packet, plus the number of
149  * samples it contained.  Only valid if @ref active is nonzero.
150  */
151 uint32_t next_timestamp;
152
153 /** @brief True if actively playing
154  *
155  * This is true when playing and false when just buffering. */
156 int active;
157
158 /** @brief Lock protecting @ref packets */
159 pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
160
161 /** @brief Condition variable signalled whenever @ref packets is changed */
162 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
163
164 /** @brief Backend to play with */
165 static const struct uaudio *backend;
166
167 HEAP_DEFINE(pheap, struct packet *, lt_packet);
168
169 /** @brief Control socket or NULL */
170 const char *control_socket;
171
172 /** @brief Buffer for debugging dump
173  *
174  * The debug dump is enabled by the @c --dump option.  It records the last 20s
175  * of audio to the specified file (which will be about 3.5Mbytes).  The file is
176  * written as as ring buffer, so the start point will progress through it.
177  *
178  * Use clients/dump2wav to convert this to a WAV file, which can then be loaded
179  * into (e.g.) Audacity for further inspection.
180  *
181  * All three backends (ALSA, OSS, Core Audio) now support this option.
182  *
183  * The idea is to allow the user a few seconds to react to an audible artefact.
184  */
185 int16_t *dump_buffer;
186
187 /** @brief Current index within debugging dump */
188 size_t dump_index;
189
190 /** @brief Size of debugging dump in samples */
191 size_t dump_size = 44100/*Hz*/ * 2/*channels*/ * 20/*seconds*/;
192
193 static const struct option options[] = {
194   { "help", no_argument, 0, 'h' },
195   { "version", no_argument, 0, 'V' },
196   { "debug", no_argument, 0, 'd' },
197   { "device", required_argument, 0, 'D' },
198   { "min", required_argument, 0, 'm' },
199   { "max", required_argument, 0, 'x' },
200   { "rcvbuf", required_argument, 0, 'R' },
201 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
202   { "oss", no_argument, 0, 'o' },
203 #endif
204 #if HAVE_ALSA_ASOUNDLIB_H
205   { "alsa", no_argument, 0, 'a' },
206 #endif
207 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
208   { "core-audio", no_argument, 0, 'c' },
209 #endif
210   { "dump", required_argument, 0, 'r' },
211   { "command", required_argument, 0, 'e' },
212   { "pause-mode", required_argument, 0, 'P' },
213   { "socket", required_argument, 0, 's' },
214   { "config", required_argument, 0, 'C' },
215   { "monitor", no_argument, 0, 'M' },
216   { 0, 0, 0, 0 }
217 };
218
219 /** @brief Control thread
220  *
221  * This thread is responsible for accepting control commands from Disobedience
222  * (or other controllers) over an AF_UNIX stream socket with a path specified
223  * by the @c --socket option.  The protocol uses simple string commands and
224  * replies:
225  *
226  * - @c stop will shut the player down
227  * - @c query will send back the reply @c running
228  * - anything else is ignored
229  *
230  * Commands and response strings terminated by shutting down the connection or
231  * by a newline.  No attempt is made to multiplex multiple clients so it is
232  * important that the command be sent as soon as the connection is made - it is
233  * assumed that both parties to the protocol are entirely cooperating with one
234  * another.
235  */
236 static void *control_thread(void attribute((unused)) *arg) {
237   struct sockaddr_un sa;
238   int sfd, cfd;
239   char *line;
240   socklen_t salen;
241   FILE *fp;
242
243   assert(control_socket);
244   unlink(control_socket);
245   memset(&sa, 0, sizeof sa);
246   sa.sun_family = AF_UNIX;
247   strcpy(sa.sun_path, control_socket);
248   sfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
249   if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0)
250     disorder_fatal(errno, "error binding to %s", control_socket);
251   if(listen(sfd, 128) < 0)
252     disorder_fatal(errno, "error calling listen on %s", control_socket);
253   disorder_info("listening on %s", control_socket);
254   for(;;) {
255     salen = sizeof sa;
256     cfd = accept(sfd, (struct sockaddr *)&sa, &salen);
257     if(cfd < 0) {
258       switch(errno) {
259       case EINTR:
260       case EAGAIN:
261         break;
262       default:
263         disorder_fatal(errno, "error calling accept on %s", control_socket);
264       }
265     }
266     if(!(fp = fdopen(cfd, "r+"))) {
267       disorder_error(errno, "error calling fdopen for %s connection", control_socket);
268       close(cfd);
269       continue;
270     }
271     if(!inputline(control_socket, fp, &line, '\n')) {
272       if(!strcmp(line, "stop")) {
273         disorder_info("stopped via %s", control_socket);
274         exit(0);                          /* terminate immediately */
275       }
276       if(!strcmp(line, "query"))
277         fprintf(fp, "running");
278       xfree(line);
279     }
280     if(fclose(fp) < 0)
281       disorder_error(errno, "error closing %s connection", control_socket);
282   }
283 }
284
285 /** @brief Drop the first packet
286  *
287  * Assumes that @ref lock is held. 
288  */
289 static void drop_first_packet(void) {
290   if(pheap_count(&packets)) {
291     struct packet *const p = pheap_remove(&packets);
292     nsamples -= p->nsamples;
293     playrtp_free_packet(p);
294     pthread_cond_broadcast(&cond);
295   }
296 }
297
298 /** @brief Background thread adding packets to heap
299  *
300  * This just transfers packets from @ref received_packets to @ref packets.  It
301  * is important that it holds @ref receive_lock for as little time as possible,
302  * in order to minimize the interval between calls to read() in
303  * listen_thread().
304  */
305 static void *queue_thread(void attribute((unused)) *arg) {
306   struct packet *p;
307
308   for(;;) {
309     /* Get the next packet */
310     pthread_mutex_lock(&receive_lock);
311     while(!received_packets) {
312       pthread_cond_wait(&receive_cond, &receive_lock);
313     }
314     p = received_packets;
315     received_packets = p->next;
316     if(!received_packets)
317       received_tail = &received_packets;
318     --nreceived;
319     pthread_mutex_unlock(&receive_lock);
320     /* Add it to the heap */
321     pthread_mutex_lock(&lock);
322     pheap_insert(&packets, p);
323     nsamples += p->nsamples;
324     pthread_cond_broadcast(&cond);
325     pthread_mutex_unlock(&lock);
326   }
327 #if HAVE_STUPID_GCC44
328   return NULL;
329 #endif
330 }
331
332 /** @brief Background thread collecting samples
333  *
334  * This function collects samples, perhaps converts them to the target format,
335  * and adds them to the packet list.
336  *
337  * It is crucial that the gap between successive calls to read() is as small as
338  * possible: otherwise packets will be dropped.
339  *
340  * We use a binary heap to ensure that the unavoidable effort is at worst
341  * logarithmic in the total number of packets - in fact if packets are mostly
342  * received in order then we will largely do constant work per packet since the
343  * newest packet will always be last.
344  *
345  * Of more concern is that we must acquire the lock on the heap to add a packet
346  * to it.  If this proves a problem in practice then the answer would be
347  * (probably doubly) linked list with new packets added the end and a second
348  * thread which reads packets off the list and adds them to the heap.
349  *
350  * We keep memory allocation (mostly) very fast by keeping pre-allocated
351  * packets around; see @ref playrtp_new_packet().
352  */
353 static void *listen_thread(void attribute((unused)) *arg) {
354   struct packet *p = 0;
355   int n;
356   struct rtp_header header;
357   uint16_t seq;
358   uint32_t timestamp;
359   struct iovec iov[2];
360
361   for(;;) {
362     if(!p)
363       p = playrtp_new_packet();
364     iov[0].iov_base = &header;
365     iov[0].iov_len = sizeof header;
366     iov[1].iov_base = p->samples_raw;
367     iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
368     n = readv(rtpfd, iov, 2);
369     if(n < 0) {
370       switch(errno) {
371       case EINTR:
372         continue;
373       default:
374         disorder_fatal(errno, "error reading from socket");
375       }
376     }
377     /* Ignore too-short packets */
378     if((size_t)n <= sizeof (struct rtp_header)) {
379       disorder_info("ignored a short packet");
380       continue;
381     }
382     timestamp = htonl(header.timestamp);
383     seq = htons(header.seq);
384     /* Ignore packets in the past */
385     if(active && lt(timestamp, next_timestamp)) {
386       disorder_info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32,
387            timestamp, next_timestamp);
388       continue;
389     }
390     /* Ignore packets with the extension bit set. */
391     if(header.vpxcc & 0x10)
392       continue;
393     p->next = 0;
394     p->flags = 0;
395     p->timestamp = timestamp;
396     /* Convert to target format */
397     if(header.mpt & 0x80)
398       p->flags |= IDLE;
399     switch(header.mpt & 0x7F) {
400     case 10:                            /* L16 */
401       p->nsamples = (n - sizeof header) / sizeof(uint16_t);
402       break;
403       /* TODO support other RFC3551 media types (when the speaker does) */
404     default:
405       disorder_fatal(0, "unsupported RTP payload type %d", header.mpt & 0x7F);
406     }
407     /* See if packet is silent */
408     const uint16_t *s = p->samples_raw;
409     n = p->nsamples;
410     for(; n > 0; --n)
411       if(*s++)
412         break;
413     if(!n)
414       p->flags |= SILENT;
415     if(logfp)
416       fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n",
417               seq, timestamp, p->nsamples, timestamp + p->nsamples);
418     /* Stop reading if we've reached the maximum.
419      *
420      * This is rather unsatisfactory: it means that if packets get heavily
421      * out of order then we guarantee dropouts.  But for now... */
422     if(nsamples >= maxbuffer) {
423       pthread_mutex_lock(&lock);
424       while(nsamples >= maxbuffer) {
425         pthread_cond_wait(&cond, &lock);
426       }
427       pthread_mutex_unlock(&lock);
428     }
429     /* Add the packet to the receive queue */
430     pthread_mutex_lock(&receive_lock);
431     *received_tail = p;
432     received_tail = &p->next;
433     ++nreceived;
434     pthread_cond_signal(&receive_cond);
435     pthread_mutex_unlock(&receive_lock);
436     /* We'll need a new packet */
437     p = 0;
438   }
439 }
440
441 /** @brief Wait until the buffer is adequately full
442  *
443  * Must be called with @ref lock held.
444  */
445 void playrtp_fill_buffer(void) {
446   /* Discard current buffer contents */
447   while(nsamples) {
448     //fprintf(stderr, "%8u/%u (%u) DROPPING\n", nsamples, maxbuffer, minbuffer);
449     drop_first_packet();
450   }
451   disorder_info("Buffering...");
452   /* Wait until there's at least minbuffer samples available */
453   while(nsamples < minbuffer) {
454     //fprintf(stderr, "%8u/%u (%u) FILLING\n", nsamples, maxbuffer, minbuffer);
455     pthread_cond_wait(&cond, &lock);
456   }
457   /* Start from whatever is earliest */
458   next_timestamp = pheap_first(&packets)->timestamp;
459   active = 1;
460 }
461
462 /** @brief Find next packet
463  * @return Packet to play or NULL if none found
464  *
465  * The return packet is merely guaranteed not to be in the past: it might be
466  * the first packet in the future rather than one that is actually suitable to
467  * play.
468  *
469  * Must be called with @ref lock held.
470  */
471 struct packet *playrtp_next_packet(void) {
472   while(pheap_count(&packets)) {
473     struct packet *const p = pheap_first(&packets);
474     if(le(p->timestamp + p->nsamples, next_timestamp)) {
475       /* This packet is in the past.  Drop it and try another one. */
476       drop_first_packet();
477     } else
478       /* This packet is NOT in the past.  (It might be in the future
479        * however.) */
480       return p;
481   }
482   return 0;
483 }
484
485 /* display usage message and terminate */
486 static void help(void) {
487   xprintf("Usage:\n"
488           "  disorder-playrtp [OPTIONS] [[ADDRESS] PORT]\n"
489           "Options:\n"
490           "  --device, -D DEVICE     Output device\n"
491           "  --min, -m FRAMES        Buffer low water mark\n"
492           "  --max, -x FRAMES        Buffer maximum size\n"
493           "  --rcvbuf, -R BYTES      Socket receive buffer size\n"
494           "  --config, -C PATH       Set configuration file\n"
495 #if HAVE_ALSA_ASOUNDLIB_H
496           "  --alsa, -a              Use ALSA to play audio\n"
497 #endif
498 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
499           "  --oss, -o               Use OSS to play audio\n"
500 #endif
501 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
502           "  --core-audio, -c        Use Core Audio to play audio\n"
503 #endif
504           "  --command, -e COMMAND   Pipe audio to command.\n"
505           "  --pause-mode, -P silence  For -e: pauses send silence (default)\n"
506           "  --pause-mode, -P suspend  For -e: pauses suspend writes\n"
507           "  --help, -h              Display usage message\n"
508           "  --version, -V           Display version number\n"
509           );
510   xfclose(stdout);
511   exit(0);
512 }
513
514 static size_t playrtp_callback(void *buffer,
515                                size_t max_samples,
516                                void attribute((unused)) *userdata) {
517   size_t samples;
518   int silent = 0;
519
520   pthread_mutex_lock(&lock);
521   /* Get the next packet, junking any that are now in the past */
522   const struct packet *p = playrtp_next_packet();
523   if(p && contains(p, next_timestamp)) {
524     /* This packet is ready to play; the desired next timestamp points
525      * somewhere into it. */
526
527     /* Timestamp of end of packet */
528     const uint32_t packet_end = p->timestamp + p->nsamples;
529
530     /* Offset of desired next timestamp into current packet */
531     const uint32_t offset = next_timestamp - p->timestamp;
532
533     /* Pointer to audio data */
534     const uint16_t *ptr = (void *)(p->samples_raw + offset);
535
536     /* Compute number of samples left in packet, limited to output buffer
537      * size */
538     samples = packet_end - next_timestamp;
539     if(samples > max_samples)
540       samples = max_samples;
541
542     /* Copy into buffer, converting to native endianness */
543     size_t i = samples;
544     int16_t *bufptr = buffer;
545     while(i > 0) {
546       *bufptr++ = (int16_t)ntohs(*ptr++);
547       --i;
548     }
549     silent = !!(p->flags & SILENT);
550   } else {
551     /* There is no suitable packet.  We introduce 0s up to the next packet, or
552      * to fill the buffer if there's no next packet or that's too many.  The
553      * comparison with max_samples deals with the otherwise troubling overflow
554      * case. */
555     samples = p ? p->timestamp - next_timestamp : max_samples;
556     if(samples > max_samples)
557       samples = max_samples;
558     //info("infill by %zu", samples);
559     memset(buffer, 0, samples * uaudio_sample_size);
560     silent = 1;
561   }
562   /* Debug dump */
563   if(dump_buffer) {
564     for(size_t i = 0; i < samples; ++i) {
565       dump_buffer[dump_index++] = ((int16_t *)buffer)[i];
566       dump_index %= dump_size;
567     }
568   }
569   /* Advance timestamp */
570   next_timestamp += samples;
571   /* If we're getting behind then try to drop just silent packets
572    *
573    * In theory this shouldn't be necessary.  The server is supposed to send
574    * packets at the right rate and compares the number of samples sent with the
575    * time in order to ensure this.
576    *
577    * However, various things could throw this off:
578    *
579    * - the server's clock could advance at the wrong rate.  This would cause it
580    *   to mis-estimate the right number of samples to have sent and
581    *   inappropriately throttle or speed up.
582    *
583    * - playback could happen at the wrong rate.  If the playback host's sound
584    *   card has a slightly incorrect clock then eventually it will get out
585    *   of step.
586    *
587    * So if we play back slightly slower than the server sends for either of
588    * these reasons then eventually our buffer, and the socket's buffer, will
589    * fill, and the kernel will start dropping packets.  The result is audible
590    * and not very nice.
591    *
592    * Therefore if we're getting behind, we pre-emptively drop silent packets,
593    * since a change in the duration of a silence is less noticeable than a
594    * dropped packet from the middle of continuous music.
595    *
596    * (If things go wrong the other way then eventually we run out of packets to
597    * play and are forced to play silence.  This doesn't seem to happen in
598    * practice but if it does then in the same way we can artificially extend
599    * silent packets to compensate.)
600    *
601    * Dropped packets are always logged; use 'disorder-playrtp --monitor' to
602    * track how close to target buffer occupancy we are on a once-a-minute
603    * basis.
604    */
605   if(nsamples > minbuffer && silent) {
606     disorder_info("dropping %zu samples (%"PRIu32" > %"PRIu32")",
607                   samples, nsamples, minbuffer);
608     samples = 0;
609   }
610   /* Junk obsolete packets */
611   playrtp_next_packet();
612   pthread_mutex_unlock(&lock);
613   return samples;
614 }
615
616 int main(int argc, char **argv) {
617   int n, err;
618   struct addrinfo *res;
619   struct stringlist sl;
620   char *sockname;
621   int rcvbuf, target_rcvbuf = 0;
622   socklen_t len;
623   struct ip_mreq mreq;
624   struct ipv6_mreq mreq6;
625   disorder_client *c;
626   char *address, *port;
627   int is_multicast;
628   union any_sockaddr {
629     struct sockaddr sa;
630     struct sockaddr_in in;
631     struct sockaddr_in6 in6;
632   };
633   union any_sockaddr mgroup;
634   const char *dumpfile = 0;
635   pthread_t ltid;
636   int monitor = 0;
637   static const int one = 1;
638
639   static const struct addrinfo prefs = {
640     .ai_flags = AI_PASSIVE,
641     .ai_family = PF_INET,
642     .ai_socktype = SOCK_DGRAM,
643     .ai_protocol = IPPROTO_UDP
644   };
645
646   /* Timing information is often important to debugging playrtp, so we include
647    * timestamps in the logs */
648   logdate = 1;
649   mem_init();
650   if(!setlocale(LC_CTYPE, "")) disorder_fatal(errno, "error calling setlocale");
651   backend = uaudio_apis[0];
652   while((n = getopt_long(argc, argv, "hVdD:m:x:L:R:aocC:re:P:M", options, 0)) >= 0) {
653     switch(n) {
654     case 'h': help();
655     case 'V': version("disorder-playrtp");
656     case 'd': debugging = 1; break;
657     case 'D': uaudio_set("device", optarg); break;
658     case 'm': minbuffer = 2 * atol(optarg); break;
659     case 'x': maxbuffer = 2 * atol(optarg); break;
660     case 'L': logfp = fopen(optarg, "w"); break;
661     case 'R': target_rcvbuf = atoi(optarg); break;
662 #if HAVE_ALSA_ASOUNDLIB_H
663     case 'a': backend = &uaudio_alsa; break;
664 #endif
665 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
666     case 'o': backend = &uaudio_oss; break;
667 #endif
668 #if HAVE_COREAUDIO_AUDIOHARDWARE_H      
669     case 'c': backend = &uaudio_coreaudio; break;
670 #endif
671     case 'C': configfile = optarg; break;
672     case 's': control_socket = optarg; break;
673     case 'r': dumpfile = optarg; break;
674     case 'e': backend = &uaudio_command; uaudio_set("command", optarg); break;
675     case 'P': uaudio_set("pause-mode", optarg); break;
676     case 'M': monitor = 1; break;
677     default: disorder_fatal(0, "invalid option");
678     }
679   }
680   if(config_read(0, NULL)) disorder_fatal(0, "cannot read configuration");
681   if(backend == &uaudio_rtp) {
682     /* This means that you have NO local sound output.  This can happen if you
683      * use a non-Apple GCC on a Mac (because it doesn't know how to compile
684      * CoreAudio/AudioHardware.h). */
685     disorder_fatal(0, "cannot play RTP through RTP");
686   }
687   if(!maxbuffer)
688     maxbuffer = 2 * minbuffer;
689   argc -= optind;
690   argv += optind;
691   switch(argc) {
692   case 0:
693     /* Get configuration from server */
694     if(!(c = disorder_new(1))) exit(EXIT_FAILURE);
695     if(disorder_connect(c)) exit(EXIT_FAILURE);
696     if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE);
697     sl.n = 2;
698     sl.s = xcalloc(2, sizeof *sl.s);
699     sl.s[0] = address;
700     sl.s[1] = port;
701     break;
702   case 1:
703   case 2:
704     /* Use command-line ADDRESS+PORT or just PORT */
705     sl.n = argc;
706     sl.s = argv;
707     break;
708   default:
709     disorder_fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]");
710   }
711   disorder_info("version "VERSION" process ID %lu",
712                 (unsigned long)getpid());
713   /* Look up address and port */
714   if(!(res = get_address(&sl, &prefs, &sockname)))
715     exit(1);
716   /* Create the socket */
717   if((rtpfd = socket(res->ai_family,
718                      res->ai_socktype,
719                      res->ai_protocol)) < 0)
720     disorder_fatal(errno, "error creating socket");
721   /* Allow multiple listeners */
722   xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
723   is_multicast = multicast(res->ai_addr);
724   /* The multicast and unicast/broadcast cases are different enough that they
725    * are totally split.  Trying to find commonality between them causes more
726    * trouble that it's worth. */
727   if(is_multicast) {
728     /* Stash the multicast group address */
729     memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
730     switch(res->ai_addr->sa_family) {
731     case AF_INET:
732       mgroup.in.sin_port = 0;
733       break;
734     case AF_INET6:
735       mgroup.in6.sin6_port = 0;
736       break;
737     default:
738       disorder_fatal(0, "unsupported address family %d",
739                      (int)res->ai_addr->sa_family);
740     }
741     /* Bind to to the multicast group address */
742     if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
743       disorder_fatal(errno, "error binding socket to %s",
744                      format_sockaddr(res->ai_addr));
745     /* Add multicast group membership */
746     switch(mgroup.sa.sa_family) {
747     case PF_INET:
748       mreq.imr_multiaddr = mgroup.in.sin_addr;
749       mreq.imr_interface.s_addr = 0;      /* use primary interface */
750       if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
751                     &mreq, sizeof mreq) < 0)
752         disorder_fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
753       break;
754     case PF_INET6:
755       mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
756       memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
757       if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
758                     &mreq6, sizeof mreq6) < 0)
759         disorder_fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
760       break;
761     default:
762       disorder_fatal(0, "unsupported address family %d", res->ai_family);
763     }
764     /* Report what we did */
765     disorder_info("listening on %s multicast group %s",
766                   format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
767   } else {
768     /* Bind to 0/port */
769     switch(res->ai_addr->sa_family) {
770     case AF_INET: {
771       struct sockaddr_in *in = (struct sockaddr_in *)res->ai_addr;
772       
773       memset(&in->sin_addr, 0, sizeof (struct in_addr));
774       break;
775     }
776     case AF_INET6: {
777       struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)res->ai_addr;
778       
779       memset(&in6->sin6_addr, 0, sizeof (struct in6_addr));
780       break;
781     }
782     default:
783       disorder_fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
784     }
785     if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
786       disorder_fatal(errno, "error binding socket to %s",
787                      format_sockaddr(res->ai_addr));
788     /* Report what we did */
789     disorder_info("listening on %s", format_sockaddr(res->ai_addr));
790   }
791   len = sizeof rcvbuf;
792   if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
793     disorder_fatal(errno, "error calling getsockopt SO_RCVBUF");
794   if(target_rcvbuf > rcvbuf) {
795     if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
796                   &target_rcvbuf, sizeof target_rcvbuf) < 0)
797       disorder_error(errno, "error calling setsockopt SO_RCVBUF %d", 
798                      target_rcvbuf);
799       /* We try to carry on anyway */
800     else
801       disorder_info("changed socket receive buffer from %d to %d",
802                     rcvbuf, target_rcvbuf);
803   } else
804     disorder_info("default socket receive buffer %d", rcvbuf);
805   //info("minbuffer %u maxbuffer %u", minbuffer, maxbuffer);
806   if(logfp)
807     disorder_info("WARNING: -L option can impact performance");
808   if(control_socket) {
809     pthread_t tid;
810
811     if((err = pthread_create(&tid, 0, control_thread, 0)))
812       disorder_fatal(err, "pthread_create control_thread");
813   }
814   if(dumpfile) {
815     int fd;
816     unsigned char buffer[65536];
817     size_t written;
818
819     if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0)
820       disorder_fatal(errno, "opening %s", dumpfile);
821     /* Fill with 0s to a suitable size */
822     memset(buffer, 0, sizeof buffer);
823     for(written = 0; written < dump_size * sizeof(int16_t);
824         written += sizeof buffer) {
825       if(write(fd, buffer, sizeof buffer) < 0)
826         disorder_fatal(errno, "clearing %s", dumpfile);
827     }
828     /* Map the buffer into memory for convenience */
829     dump_buffer = mmap(0, dump_size * sizeof(int16_t), PROT_READ|PROT_WRITE,
830                        MAP_SHARED, fd, 0);
831     if(dump_buffer == (void *)-1)
832       disorder_fatal(errno, "mapping %s", dumpfile);
833     disorder_info("dumping to %s", dumpfile);
834   }
835   /* Set up output.  Currently we only support L16 so there's no harm setting
836    * the format before we know what it is! */
837   uaudio_set_format(44100/*Hz*/, 2/*channels*/,
838                     16/*bits/channel*/, 1/*signed*/);
839   backend->start(playrtp_callback, NULL);
840   /* We receive and convert audio data in a background thread */
841   if((err = pthread_create(&ltid, 0, listen_thread, 0)))
842     disorder_fatal(err, "pthread_create listen_thread");
843   /* We have a second thread to add received packets to the queue */
844   if((err = pthread_create(&ltid, 0, queue_thread, 0)))
845     disorder_fatal(err, "pthread_create queue_thread");
846   pthread_mutex_lock(&lock);
847   time_t lastlog = 0;
848   for(;;) {
849     /* Wait for the buffer to fill up a bit */
850     playrtp_fill_buffer();
851     /* Start playing now */
852     disorder_info("Playing...");
853     next_timestamp = pheap_first(&packets)->timestamp;
854     active = 1;
855     pthread_mutex_unlock(&lock);
856     backend->activate();
857     pthread_mutex_lock(&lock);
858     /* Wait until the buffer empties out
859      *
860      * If there's a packet that we can play right now then we definitely
861      * continue.
862      *
863      * Also if there's at least minbuffer samples we carry on regardless and
864      * insert silence.  The assumption is there's been a pause but more data
865      * is now available.
866      */
867     while(nsamples >= minbuffer
868           || (nsamples > 0
869               && contains(pheap_first(&packets), next_timestamp))) {
870       if(monitor) {
871         time_t now = xtime(0);
872
873         if(now >= lastlog + 60) {
874           int offset = nsamples - minbuffer;
875           double offtime = (double)offset / (uaudio_rate * uaudio_channels);
876           disorder_info("%+d samples off (%d.%02ds, %d bytes)",
877                         offset,
878                         (int)fabs(offtime) * (offtime < 0 ? -1 : 1),
879                         (int)(fabs(offtime) * 100) % 100,
880                         offset * uaudio_bits / CHAR_BIT);
881           lastlog = now;
882         }
883       }
884       //fprintf(stderr, "%8u/%u (%u) PLAYING\n", nsamples, maxbuffer, minbuffer);
885       pthread_cond_wait(&cond, &lock);
886     }
887 #if 0
888     if(nsamples) {
889       struct packet *p = pheap_first(&packets);
890       fprintf(stderr, "nsamples=%u (%u) next_timestamp=%"PRIx32", first packet is [%"PRIx32",%"PRIx32")\n",
891               nsamples, minbuffer, next_timestamp,p->timestamp,p->timestamp+p->nsamples);
892     }
893 #endif
894     /* Stop playing for a bit until the buffer re-fills */
895     pthread_mutex_unlock(&lock);
896     backend->deactivate();
897     pthread_mutex_lock(&lock);
898     active = 0;
899     /* Go back round */
900   }
901   return 0;
902 }
903
904 /*
905 Local Variables:
906 c-basic-offset:2
907 comment-column:40
908 fill-column:79
909 indent-tabs-mode:nil
910 End:
911 */