2 * This file is part of DisOrder.
3 * Copyright (C) 2007, 2008 Richard Kettlewell
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.
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.
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
20 /** @file clients/playrtp.c
23 * This player supports Linux (<a href="http://www.alsa-project.org/">ALSA</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.
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).
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.
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.
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.
49 * - it is safe to read uint32_t values without a lock protecting them
55 #include <sys/socket.h>
56 #include <sys/types.h>
57 #include <sys/socket.h>
63 #include <netinet/in.h>
72 #include "configuration.h"
82 #include "inputline.h"
85 #define readahead linux_headers_are_borked
87 /** @brief Obsolete synonym */
88 #ifndef IPV6_JOIN_GROUP
89 # define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
92 /** @brief RTP socket */
95 /** @brief Log output */
98 /** @brief Output device */
101 /** @brief Minimum low watermark
103 * We'll stop playing if there's only this many samples in the buffer. */
104 unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */
106 /** @brief Buffer high watermark
108 * We'll only start playing when this many samples are available. */
109 static unsigned readahead = 2 * 2 * 44100;
111 /** @brief Maximum buffer size
113 * We'll stop reading from the network if we have this many samples. */
114 static unsigned maxbuffer;
116 /** @brief Received packets
117 * Protected by @ref receive_lock
119 * Received packets are added to this list, and queue_thread() picks them off
120 * it and adds them to @ref packets. Whenever a packet is added to it, @ref
121 * receive_cond is signalled.
123 struct packet *received_packets;
125 /** @brief Tail of @ref received_packets
126 * Protected by @ref receive_lock
128 struct packet **received_tail = &received_packets;
130 /** @brief Lock protecting @ref received_packets
132 * Only listen_thread() and queue_thread() ever hold this lock. It is vital
133 * that queue_thread() not hold it any longer than it strictly has to. */
134 pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
136 /** @brief Condition variable signalled when @ref received_packets is updated
138 * Used by listen_thread() to notify queue_thread() that it has added another
139 * packet to @ref received_packets. */
140 pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
142 /** @brief Length of @ref received_packets */
145 /** @brief Binary heap of received packets */
146 struct pheap packets;
148 /** @brief Total number of samples available
150 * We make this volatile because we inspect it without a protecting lock,
151 * so the usual pthread_* guarantees aren't available.
153 volatile uint32_t nsamples;
155 /** @brief Timestamp of next packet to play.
157 * This is set to the timestamp of the last packet, plus the number of
158 * samples it contained. Only valid if @ref active is nonzero.
160 uint32_t next_timestamp;
162 /** @brief True if actively playing
164 * This is true when playing and false when just buffering. */
167 /** @brief Lock protecting @ref packets */
168 pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
170 /** @brief Condition variable signalled whenever @ref packets is changed */
171 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
173 #if DEFAULT_BACKEND == BACKEND_ALSA
174 # define DEFAULT_PLAYRTP_BACKEND playrtp_alsa
175 #elif DEFAULT_BACKEND == BACKEND_OSS
176 # define DEFAULT_PLAYRTP_BACKEND playrtp_oss
177 #elif DEFAULT_BACKEND == BACKEND_COREAUDIO
178 # define DEFAULT_PLAYRTP_BACKEND playrtp_coreaudio
181 /** @brief Backend to play with */
182 static void (*backend)(void) = DEFAULT_PLAYRTP_BACKEND;
184 HEAP_DEFINE(pheap, struct packet *, lt_packet);
186 /** @brief Control socket or NULL */
187 const char *control_socket;
189 /** @brief Buffer for debugging dump
191 * The debug dump is enabled by the @c --dump option. It records the last 20s
192 * of audio to the specified file (which will be about 3.5Mbytes). The file is
193 * written as as ring buffer, so the start point will progress through it.
195 * Use clients/dump2wav to convert this to a WAV file, which can then be loaded
196 * into (e.g.) Audacity for further inspection.
198 * All three backends (ALSA, OSS, Core Audio) now support this option.
200 * The idea is to allow the user a few seconds to react to an audible artefact.
202 int16_t *dump_buffer;
204 /** @brief Current index within debugging dump */
207 /** @brief Size of debugging dump in samples */
208 size_t dump_size = 44100/*Hz*/ * 2/*channels*/ * 20/*seconds*/;
210 static const struct option options[] = {
211 { "help", no_argument, 0, 'h' },
212 { "version", no_argument, 0, 'V' },
213 { "debug", no_argument, 0, 'd' },
214 { "device", required_argument, 0, 'D' },
215 { "min", required_argument, 0, 'm' },
216 { "max", required_argument, 0, 'x' },
217 { "buffer", required_argument, 0, 'b' },
218 { "rcvbuf", required_argument, 0, 'R' },
219 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
220 { "oss", no_argument, 0, 'o' },
222 #if HAVE_ALSA_ASOUNDLIB_H
223 { "alsa", no_argument, 0, 'a' },
225 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
226 { "core-audio", no_argument, 0, 'c' },
228 { "dump", required_argument, 0, 'r' },
229 { "socket", required_argument, 0, 's' },
230 { "config", required_argument, 0, 'C' },
234 /** @brief Control thread
236 * This thread is responsible for accepting control commands from Disobedience
237 * (or other controllers) over an AF_UNIX stream socket with a path specified
238 * by the @c --socket option. The protocol uses simple string commands and
241 * - @c stop will shut the player down
242 * - @c query will send back the reply @c running
243 * - anything else is ignored
245 * Commands and response strings terminated by shutting down the connection or
246 * by a newline. No attempt is made to multiplex multiple clients so it is
247 * important that the command be sent as soon as the connection is made - it is
248 * assumed that both parties to the protocol are entirely cooperating with one
251 static void *control_thread(void attribute((unused)) *arg) {
252 struct sockaddr_un sa;
258 assert(control_socket);
259 unlink(control_socket);
260 memset(&sa, 0, sizeof sa);
261 sa.sun_family = AF_UNIX;
262 strcpy(sa.sun_path, control_socket);
263 sfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
264 if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0)
265 fatal(errno, "error binding to %s", control_socket);
266 if(listen(sfd, 128) < 0)
267 fatal(errno, "error calling listen on %s", control_socket);
268 info("listening on %s", control_socket);
271 cfd = accept(sfd, (struct sockaddr *)&sa, &salen);
278 fatal(errno, "error calling accept on %s", control_socket);
281 if(!(fp = fdopen(cfd, "r+"))) {
282 error(errno, "error calling fdopen for %s connection", control_socket);
286 if(!inputline(control_socket, fp, &line, '\n')) {
287 if(!strcmp(line, "stop")) {
288 info("stopped via %s", control_socket);
289 exit(0); /* terminate immediately */
291 if(!strcmp(line, "query"))
292 fprintf(fp, "running");
296 error(errno, "error closing %s connection", control_socket);
300 /** @brief Drop the first packet
302 * Assumes that @ref lock is held.
304 static void drop_first_packet(void) {
305 if(pheap_count(&packets)) {
306 struct packet *const p = pheap_remove(&packets);
307 nsamples -= p->nsamples;
308 playrtp_free_packet(p);
309 pthread_cond_broadcast(&cond);
313 /** @brief Background thread adding packets to heap
315 * This just transfers packets from @ref received_packets to @ref packets. It
316 * is important that it holds @ref receive_lock for as little time as possible,
317 * in order to minimize the interval between calls to read() in
320 static void *queue_thread(void attribute((unused)) *arg) {
324 /* Get the next packet */
325 pthread_mutex_lock(&receive_lock);
326 while(!received_packets) {
327 pthread_cond_wait(&receive_cond, &receive_lock);
329 p = received_packets;
330 received_packets = p->next;
331 if(!received_packets)
332 received_tail = &received_packets;
334 pthread_mutex_unlock(&receive_lock);
335 /* Add it to the heap */
336 pthread_mutex_lock(&lock);
337 pheap_insert(&packets, p);
338 nsamples += p->nsamples;
339 pthread_cond_broadcast(&cond);
340 pthread_mutex_unlock(&lock);
344 /** @brief Background thread collecting samples
346 * This function collects samples, perhaps converts them to the target format,
347 * and adds them to the packet list.
349 * It is crucial that the gap between successive calls to read() is as small as
350 * possible: otherwise packets will be dropped.
352 * We use a binary heap to ensure that the unavoidable effort is at worst
353 * logarithmic in the total number of packets - in fact if packets are mostly
354 * received in order then we will largely do constant work per packet since the
355 * newest packet will always be last.
357 * Of more concern is that we must acquire the lock on the heap to add a packet
358 * to it. If this proves a problem in practice then the answer would be
359 * (probably doubly) linked list with new packets added the end and a second
360 * thread which reads packets off the list and adds them to the heap.
362 * We keep memory allocation (mostly) very fast by keeping pre-allocated
363 * packets around; see @ref playrtp_new_packet().
365 static void *listen_thread(void attribute((unused)) *arg) {
366 struct packet *p = 0;
368 struct rtp_header header;
375 p = playrtp_new_packet();
376 iov[0].iov_base = &header;
377 iov[0].iov_len = sizeof header;
378 iov[1].iov_base = p->samples_raw;
379 iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
380 n = readv(rtpfd, iov, 2);
386 fatal(errno, "error reading from socket");
389 /* Ignore too-short packets */
390 if((size_t)n <= sizeof (struct rtp_header)) {
391 info("ignored a short packet");
394 timestamp = htonl(header.timestamp);
395 seq = htons(header.seq);
396 /* Ignore packets in the past */
397 if(active && lt(timestamp, next_timestamp)) {
398 info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32,
399 timestamp, next_timestamp);
404 p->timestamp = timestamp;
405 /* Convert to target format */
406 if(header.mpt & 0x80)
408 switch(header.mpt & 0x7F) {
410 p->nsamples = (n - sizeof header) / sizeof(uint16_t);
412 /* TODO support other RFC3551 media types (when the speaker does) */
414 fatal(0, "unsupported RTP payload type %d",
418 fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n",
419 seq, timestamp, p->nsamples, timestamp + p->nsamples);
420 /* Stop reading if we've reached the maximum.
422 * This is rather unsatisfactory: it means that if packets get heavily
423 * out of order then we guarantee dropouts. But for now... */
424 if(nsamples >= maxbuffer) {
425 pthread_mutex_lock(&lock);
426 while(nsamples >= maxbuffer) {
427 pthread_cond_wait(&cond, &lock);
429 pthread_mutex_unlock(&lock);
431 /* Add the packet to the receive queue */
432 pthread_mutex_lock(&receive_lock);
434 received_tail = &p->next;
436 pthread_cond_signal(&receive_cond);
437 pthread_mutex_unlock(&receive_lock);
438 /* We'll need a new packet */
443 /** @brief Wait until the buffer is adequately full
445 * Must be called with @ref lock held.
447 void playrtp_fill_buffer(void) {
450 info("Buffering...");
451 while(nsamples < readahead) {
452 pthread_cond_wait(&cond, &lock);
454 next_timestamp = pheap_first(&packets)->timestamp;
458 /** @brief Find next packet
459 * @return Packet to play or NULL if none found
461 * The return packet is merely guaranteed not to be in the past: it might be
462 * the first packet in the future rather than one that is actually suitable to
465 * Must be called with @ref lock held.
467 struct packet *playrtp_next_packet(void) {
468 while(pheap_count(&packets)) {
469 struct packet *const p = pheap_first(&packets);
470 if(le(p->timestamp + p->nsamples, next_timestamp)) {
471 /* This packet is in the past. Drop it and try another one. */
474 /* This packet is NOT in the past. (It might be in the future
481 /** @brief Play an RTP stream
483 * This is the guts of the program. It is responsible for:
484 * - starting the listening thread
485 * - opening the audio device
486 * - reading ahead to build up a buffer
487 * - arranging for audio to be played
488 * - detecting when the buffer has got too small and re-buffering
490 static void play_rtp(void) {
494 /* We receive and convert audio data in a background thread */
495 if((err = pthread_create(<id, 0, listen_thread, 0)))
496 fatal(err, "pthread_create listen_thread");
497 /* We have a second thread to add received packets to the queue */
498 if((err = pthread_create(<id, 0, queue_thread, 0)))
499 fatal(err, "pthread_create queue_thread");
500 /* The rest of the work is backend-specific */
504 /* display usage message and terminate */
505 static void help(void) {
507 " disorder-playrtp [OPTIONS] ADDRESS [PORT]\n"
509 " --device, -D DEVICE Output device\n"
510 " --min, -m FRAMES Buffer low water mark\n"
511 " --buffer, -b FRAMES Buffer high water mark\n"
512 " --max, -x FRAMES Buffer maximum size\n"
513 " --rcvbuf, -R BYTES Socket receive buffer size\n"
514 " --config, -C PATH Set configuration file\n"
515 #if HAVE_ALSA_ASOUNDLIB_H
516 " --alsa, -a Use ALSA to play audio\n"
518 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
519 " --oss, -o Use OSS to play audio\n"
521 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
522 " --core-audio, -c Use Core Audio to play audio\n"
524 " --help, -h Display usage message\n"
525 " --version, -V Display version number\n"
531 int main(int argc, char **argv) {
533 struct addrinfo *res;
534 struct stringlist sl;
536 int rcvbuf, target_rcvbuf = 131072;
539 struct ipv6_mreq mreq6;
541 char *address, *port;
545 struct sockaddr_in in;
546 struct sockaddr_in6 in6;
548 union any_sockaddr mgroup;
549 const char *dumpfile = 0;
551 static const struct addrinfo prefs = {
552 .ai_flags = AI_PASSIVE,
553 .ai_family = PF_INET,
554 .ai_socktype = SOCK_DGRAM,
555 .ai_protocol = IPPROTO_UDP
559 if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale");
560 while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:r", options, 0)) >= 0) {
563 case 'V': version("disorder-playrtp");
564 case 'd': debugging = 1; break;
565 case 'D': device = optarg; break;
566 case 'm': minbuffer = 2 * atol(optarg); break;
567 case 'b': readahead = 2 * atol(optarg); break;
568 case 'x': maxbuffer = 2 * atol(optarg); break;
569 case 'L': logfp = fopen(optarg, "w"); break;
570 case 'R': target_rcvbuf = atoi(optarg); break;
571 #if HAVE_ALSA_ASOUNDLIB_H
572 case 'a': backend = playrtp_alsa; break;
574 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
575 case 'o': backend = playrtp_oss; break;
577 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
578 case 'c': backend = playrtp_coreaudio; break;
580 case 'C': configfile = optarg; break;
581 case 's': control_socket = optarg; break;
582 case 'r': dumpfile = optarg; break;
583 default: fatal(0, "invalid option");
586 if(config_read(0)) fatal(0, "cannot read configuration");
588 maxbuffer = 4 * readahead;
593 /* Get configuration from server */
594 if(!(c = disorder_new(1))) exit(EXIT_FAILURE);
595 if(disorder_connect(c)) exit(EXIT_FAILURE);
596 if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE);
598 sl.s = xcalloc(2, sizeof *sl.s);
604 /* Use command-line ADDRESS+PORT or just PORT */
609 fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]");
611 /* Look up address and port */
612 if(!(res = get_address(&sl, &prefs, &sockname)))
614 /* Create the socket */
615 if((rtpfd = socket(res->ai_family,
617 res->ai_protocol)) < 0)
618 fatal(errno, "error creating socket");
619 /* Stash the multicast group address */
620 if((is_multicast = multicast(res->ai_addr))) {
621 memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
622 switch(res->ai_addr->sa_family) {
624 mgroup.in.sin_port = 0;
627 mgroup.in6.sin6_port = 0;
632 switch(res->ai_addr->sa_family) {
634 memset(&((struct sockaddr_in *)res->ai_addr)->sin_addr, 0,
635 sizeof (struct in_addr));
638 memset(&((struct sockaddr_in6 *)res->ai_addr)->sin6_addr, 0,
639 sizeof (struct in6_addr));
642 fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
644 if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
645 fatal(errno, "error binding socket to %s", sockname);
647 switch(mgroup.sa.sa_family) {
649 mreq.imr_multiaddr = mgroup.in.sin_addr;
650 mreq.imr_interface.s_addr = 0; /* use primary interface */
651 if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
652 &mreq, sizeof mreq) < 0)
653 fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
656 mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
657 memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
658 if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
659 &mreq6, sizeof mreq6) < 0)
660 fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
663 fatal(0, "unsupported address family %d", res->ai_family);
665 info("listening on %s multicast group %s",
666 format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
668 info("listening on %s", format_sockaddr(res->ai_addr));
670 if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
671 fatal(errno, "error calling getsockopt SO_RCVBUF");
672 if(target_rcvbuf > rcvbuf) {
673 if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
674 &target_rcvbuf, sizeof target_rcvbuf) < 0)
675 error(errno, "error calling setsockopt SO_RCVBUF %d",
677 /* We try to carry on anyway */
679 info("changed socket receive buffer from %d to %d",
680 rcvbuf, target_rcvbuf);
682 info("default socket receive buffer %d", rcvbuf);
684 info("WARNING: -L option can impact performance");
688 if((err = pthread_create(&tid, 0, control_thread, 0)))
689 fatal(err, "pthread_create control_thread");
693 unsigned char buffer[65536];
696 if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0)
697 fatal(errno, "opening %s", dumpfile);
698 /* Fill with 0s to a suitable size */
699 memset(buffer, 0, sizeof buffer);
700 for(written = 0; written < dump_size * sizeof(int16_t);
701 written += sizeof buffer) {
702 if(write(fd, buffer, sizeof buffer) < 0)
703 fatal(errno, "clearing %s", dumpfile);
705 /* Map the buffer into memory for convenience */
706 dump_buffer = mmap(0, dump_size * sizeof(int16_t), PROT_READ|PROT_WRITE,
708 if(dump_buffer == (void *)-1)
709 fatal(errno, "mapping %s", dumpfile);
710 info("dumping to %s", dumpfile);