X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/a7e9570adf37c18f12f6b1dfed0184b70343575e..4942ee7d61bf22ba38bf026c7d05028cb7db0d54:/clients/playrtp.c diff --git a/clients/playrtp.c b/clients/playrtp.c index 0b5bc04..7eed9eb 100644 --- a/clients/playrtp.c +++ b/clients/playrtp.c @@ -1,21 +1,19 @@ /* * This file is part of DisOrder. - * Copyright (C) 2007 Richard Kettlewell + * Copyright (C) 2007-2009 Richard Kettlewell * - * This program is free software; you can redistribute it and/or modify + * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or + * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * General Public License for more details. - * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 - * USA + * along with this program. If not, see . */ /** @file clients/playrtp.c * @brief RTP player @@ -26,20 +24,20 @@ * systems. There is no support for Microsoft Windows yet, and that will in * fact probably an entirely separate program. * - * The program runs (at least) three threads. listen_thread() is responsible - * for reading RTP packets off the wire and adding them to the linked list @ref - * received_packets, assuming they are basically sound. queue_thread() takes - * packets off this linked list and adds them to @ref packets (an operation - * which might be much slower due to contention for @ref lock). + * The program runs (at least) three threads: + * + * listen_thread() is responsible for reading RTP packets off the wire and + * adding them to the linked list @ref received_packets, assuming they are + * basically sound. + * + * queue_thread() takes packets off this linked list and adds them to @ref + * packets (an operation which might be much slower due to contention for @ref + * lock). * - * The main thread is responsible for actually playing audio. In ALSA this - * means it waits until ALSA says it's ready for more audio which it then - * plays. See @ref clients/playrtp-alsa.c. + * control_thread() accepts commands from Disobedience (or anything else). * - * In Core Audio the main thread is only responsible for starting and stopping - * play: the system does the actual playback in its own private thread, and - * calls adioproc() to fetch the audio data. See @ref - * clients/playrtp-coreaudio.c. + * The main thread activates and deactivates audio playing via the @ref + * lib/uaudio.h API (which probably implies at least one further thread). * * Sometimes it happens that there is no audio available to play. This may * because the server went away, or a packet was dropped, or the server @@ -49,12 +47,9 @@ * - it is safe to read uint32_t values without a lock protecting them */ -#include -#include "types.h" +#include "common.h" #include -#include -#include #include #include #include @@ -62,9 +57,14 @@ #include #include #include -#include -#include #include +#include +#include +#include +#include +#include +#include +#include #include "log.h" #include "mem.h" @@ -78,8 +78,14 @@ #include "timeval.h" #include "client.h" #include "playrtp.h" +#include "inputline.h" +#include "version.h" +#include "uaudio.h" -#define readahead linux_headers_are_borked +/** @brief Obsolete synonym */ +#ifndef IPV6_JOIN_GROUP +# define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP +#endif /** @brief RTP socket */ static int rtpfd; @@ -88,21 +94,14 @@ static int rtpfd; static FILE *logfp; /** @brief Output device */ -const char *device; - -/** @brief Minimum low watermark - * - * We'll stop playing if there's only this many samples in the buffer. */ -unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */ -/** @brief Buffer high watermark - * - * We'll only start playing when this many samples are available. */ -static unsigned readahead = 2 * 2 * 44100; +/** @brief Buffer low watermark in samples */ +unsigned minbuffer = 4 * (2 * 44100) / 10; /* 0.4 seconds */ -/** @brief Maximum buffer size +/** @brief Maximum buffer size in samples * - * We'll stop reading from the network if we have this many samples. */ + * We'll stop reading from the network if we have this many samples. + */ static unsigned maxbuffer; /** @brief Received packets @@ -162,21 +161,35 @@ pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; /** @brief Condition variable signalled whenever @ref packets is changed */ pthread_cond_t cond = PTHREAD_COND_INITIALIZER; -#if HAVE_ALSA_ASOUNDLIB_H -# define DEFAULT_BACKEND playrtp_alsa -#elif HAVE_SYS_SOUNDCARD_H || EMPEG_HOST -# define DEFAULT_BACKEND playrtp_oss -#elif HAVE_COREAUDIO_AUDIOHARDWARE_H -# define DEFAULT_BACKEND playrtp_coreaudio -#else -# error No known backend -#endif - /** @brief Backend to play with */ -static void (*backend)(void) = &DEFAULT_BACKEND; +static const struct uaudio *backend; HEAP_DEFINE(pheap, struct packet *, lt_packet); +/** @brief Control socket or NULL */ +const char *control_socket; + +/** @brief Buffer for debugging dump + * + * The debug dump is enabled by the @c --dump option. It records the last 20s + * of audio to the specified file (which will be about 3.5Mbytes). The file is + * written as as ring buffer, so the start point will progress through it. + * + * Use clients/dump2wav to convert this to a WAV file, which can then be loaded + * into (e.g.) Audacity for further inspection. + * + * All three backends (ALSA, OSS, Core Audio) now support this option. + * + * The idea is to allow the user a few seconds to react to an audible artefact. + */ +int16_t *dump_buffer; + +/** @brief Current index within debugging dump */ +size_t dump_index; + +/** @brief Size of debugging dump in samples */ +size_t dump_size = 44100/*Hz*/ * 2/*channels*/ * 20/*seconds*/; + static const struct option options[] = { { "help", no_argument, 0, 'h' }, { "version", no_argument, 0, 'V' }, @@ -184,9 +197,7 @@ static const struct option options[] = { { "device", required_argument, 0, 'D' }, { "min", required_argument, 0, 'm' }, { "max", required_argument, 0, 'x' }, - { "buffer", required_argument, 0, 'b' }, { "rcvbuf", required_argument, 0, 'R' }, - { "multicast", required_argument, 0, 'M' }, #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST { "oss", no_argument, 0, 'o' }, #endif @@ -196,10 +207,81 @@ static const struct option options[] = { #if HAVE_COREAUDIO_AUDIOHARDWARE_H { "core-audio", no_argument, 0, 'c' }, #endif + { "dump", required_argument, 0, 'r' }, + { "command", required_argument, 0, 'e' }, + { "pause-mode", required_argument, 0, 'P' }, + { "socket", required_argument, 0, 's' }, { "config", required_argument, 0, 'C' }, + { "monitor", no_argument, 0, 'M' }, { 0, 0, 0, 0 } }; +/** @brief Control thread + * + * This thread is responsible for accepting control commands from Disobedience + * (or other controllers) over an AF_UNIX stream socket with a path specified + * by the @c --socket option. The protocol uses simple string commands and + * replies: + * + * - @c stop will shut the player down + * - @c query will send back the reply @c running + * - anything else is ignored + * + * Commands and response strings terminated by shutting down the connection or + * by a newline. No attempt is made to multiplex multiple clients so it is + * important that the command be sent as soon as the connection is made - it is + * assumed that both parties to the protocol are entirely cooperating with one + * another. + */ +static void *control_thread(void attribute((unused)) *arg) { + struct sockaddr_un sa; + int sfd, cfd; + char *line; + socklen_t salen; + FILE *fp; + + assert(control_socket); + unlink(control_socket); + memset(&sa, 0, sizeof sa); + sa.sun_family = AF_UNIX; + strcpy(sa.sun_path, control_socket); + sfd = xsocket(PF_UNIX, SOCK_STREAM, 0); + if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0) + fatal(errno, "error binding to %s", control_socket); + if(listen(sfd, 128) < 0) + fatal(errno, "error calling listen on %s", control_socket); + info("listening on %s", control_socket); + for(;;) { + salen = sizeof sa; + cfd = accept(sfd, (struct sockaddr *)&sa, &salen); + if(cfd < 0) { + switch(errno) { + case EINTR: + case EAGAIN: + break; + default: + fatal(errno, "error calling accept on %s", control_socket); + } + } + if(!(fp = fdopen(cfd, "r+"))) { + error(errno, "error calling fdopen for %s connection", control_socket); + close(cfd); + continue; + } + if(!inputline(control_socket, fp, &line, '\n')) { + if(!strcmp(line, "stop")) { + info("stopped via %s", control_socket); + exit(0); /* terminate immediately */ + } + if(!strcmp(line, "query")) + fprintf(fp, "running"); + xfree(line); + } + if(fclose(fp) < 0) + error(errno, "error closing %s connection", control_socket); + } +} + /** @brief Drop the first packet * * Assumes that @ref lock is held. @@ -226,8 +308,9 @@ static void *queue_thread(void attribute((unused)) *arg) { for(;;) { /* Get the next packet */ pthread_mutex_lock(&receive_lock); - while(!received_packets) + while(!received_packets) { pthread_cond_wait(&receive_cond, &receive_lock); + } p = received_packets; received_packets = p->next; if(!received_packets) @@ -241,6 +324,9 @@ static void *queue_thread(void attribute((unused)) *arg) { pthread_cond_broadcast(&cond); pthread_mutex_unlock(&lock); } +#if HAVE_STUPID_GCC44 + return NULL; +#endif } /** @brief Background thread collecting samples @@ -301,6 +387,9 @@ static void *listen_thread(void attribute((unused)) *arg) { timestamp, next_timestamp); continue; } + /* Ignore packets with the extension bit set. */ + if(header.vpxcc & 0x10) + continue; p->next = 0; p->flags = 0; p->timestamp = timestamp; @@ -308,7 +397,7 @@ static void *listen_thread(void attribute((unused)) *arg) { if(header.mpt & 0x80) p->flags |= IDLE; switch(header.mpt & 0x7F) { - case 10: + case 10: /* L16 */ p->nsamples = (n - sizeof header) / sizeof(uint16_t); break; /* TODO support other RFC3551 media types (when the speaker does) */ @@ -316,6 +405,14 @@ static void *listen_thread(void attribute((unused)) *arg) { fatal(0, "unsupported RTP payload type %d", header.mpt & 0x7F); } + /* See if packet is silent */ + const uint16_t *s = p->samples_raw; + n = p->nsamples; + for(; n > 0; --n) + if(*s++) + break; + if(!n) + p->flags |= SILENT; if(logfp) fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n", seq, timestamp, p->nsamples, timestamp + p->nsamples); @@ -325,8 +422,9 @@ static void *listen_thread(void attribute((unused)) *arg) { * out of order then we guarantee dropouts. But for now... */ if(nsamples >= maxbuffer) { pthread_mutex_lock(&lock); - while(nsamples >= maxbuffer) + while(nsamples >= maxbuffer) { pthread_cond_wait(&cond, &lock); + } pthread_mutex_unlock(&lock); } /* Add the packet to the receive queue */ @@ -346,11 +444,18 @@ static void *listen_thread(void attribute((unused)) *arg) { * Must be called with @ref lock held. */ void playrtp_fill_buffer(void) { - while(nsamples) + /* Discard current buffer contents */ + while(nsamples) { + //fprintf(stderr, "%8u/%u (%u) DROPPING\n", nsamples, maxbuffer, minbuffer); drop_first_packet(); + } info("Buffering..."); - while(nsamples < readahead) + /* Wait until there's at least minbuffer samples available */ + while(nsamples < minbuffer) { + //fprintf(stderr, "%8u/%u (%u) FILLING\n", nsamples, maxbuffer, minbuffer); pthread_cond_wait(&cond, &lock); + } + /* Start from whatever is earliest */ next_timestamp = pheap_first(&packets)->timestamp; active = 1; } @@ -378,37 +483,15 @@ struct packet *playrtp_next_packet(void) { return 0; } -/** @brief Play an RTP stream - * - * This is the guts of the program. It is responsible for: - * - starting the listening thread - * - opening the audio device - * - reading ahead to build up a buffer - * - arranging for audio to be played - * - detecting when the buffer has got too small and re-buffering - */ -static void play_rtp(void) { - pthread_t ltid; - - /* We receive and convert audio data in a background thread */ - pthread_create(<id, 0, listen_thread, 0); - /* We have a second thread to add received packets to the queue */ - pthread_create(<id, 0, queue_thread, 0); - /* The rest of the work is backend-specific */ - backend(); -} - /* display usage message and terminate */ static void help(void) { xprintf("Usage:\n" - " disorder-playrtp [OPTIONS] ADDRESS [PORT]\n" + " disorder-playrtp [OPTIONS] [[ADDRESS] PORT]\n" "Options:\n" " --device, -D DEVICE Output device\n" " --min, -m FRAMES Buffer low water mark\n" - " --buffer, -b FRAMES Buffer high water mark\n" " --max, -x FRAMES Buffer maximum size\n" " --rcvbuf, -R BYTES Socket receive buffer size\n" - " --multicast, -M GROUP Join multicast group\n" " --config, -C PATH Set configuration file\n" #if HAVE_ALSA_ASOUNDLIB_H " --alsa, -a Use ALSA to play audio\n" @@ -419,6 +502,9 @@ static void help(void) { #if HAVE_COREAUDIO_AUDIOHARDWARE_H " --core-audio, -c Use Core Audio to play audio\n" #endif + " --command, -e COMMAND Pipe audio to command.\n" + " --pause-mode, -P silence For -e: pauses send silence (default)\n" + " --pause-mode, -P suspend For -e: pauses suspend writes\n" " --help, -h Display usage message\n" " --version, -V Display version number\n" ); @@ -426,109 +512,238 @@ static void help(void) { exit(0); } -/* display version number and terminate */ -static void version(void) { - xprintf("disorder-playrtp version %s\n", disorder_version_string); - xfclose(stdout); - exit(0); +static size_t playrtp_callback(void *buffer, + size_t max_samples, + void attribute((unused)) *userdata) { + size_t samples; + int silent = 0; + + pthread_mutex_lock(&lock); + /* Get the next packet, junking any that are now in the past */ + const struct packet *p = playrtp_next_packet(); + if(p && contains(p, next_timestamp)) { + /* This packet is ready to play; the desired next timestamp points + * somewhere into it. */ + + /* Timestamp of end of packet */ + const uint32_t packet_end = p->timestamp + p->nsamples; + + /* Offset of desired next timestamp into current packet */ + const uint32_t offset = next_timestamp - p->timestamp; + + /* Pointer to audio data */ + const uint16_t *ptr = (void *)(p->samples_raw + offset); + + /* Compute number of samples left in packet, limited to output buffer + * size */ + samples = packet_end - next_timestamp; + if(samples > max_samples) + samples = max_samples; + + /* Copy into buffer, converting to native endianness */ + size_t i = samples; + int16_t *bufptr = buffer; + while(i > 0) { + *bufptr++ = (int16_t)ntohs(*ptr++); + --i; + } + silent = !!(p->flags & SILENT); + } else { + /* There is no suitable packet. We introduce 0s up to the next packet, or + * to fill the buffer if there's no next packet or that's too many. The + * comparison with max_samples deals with the otherwise troubling overflow + * case. */ + samples = p ? p->timestamp - next_timestamp : max_samples; + if(samples > max_samples) + samples = max_samples; + //info("infill by %zu", samples); + memset(buffer, 0, samples * uaudio_sample_size); + silent = 1; + } + /* Debug dump */ + if(dump_buffer) { + for(size_t i = 0; i < samples; ++i) { + dump_buffer[dump_index++] = ((int16_t *)buffer)[i]; + dump_index %= dump_size; + } + } + /* Advance timestamp */ + next_timestamp += samples; + /* If we're getting behind then try to drop just silent packets + * + * In theory this shouldn't be necessary. The server is supposed to send + * packets at the right rate and compares the number of samples sent with the + * time in order to ensure this. + * + * However, various things could throw this off: + * + * - the server's clock could advance at the wrong rate. This would cause it + * to mis-estimate the right number of samples to have sent and + * inappropriately throttle or speed up. + * + * - playback could happen at the wrong rate. If the playback host's sound + * card has a slightly incorrect clock then eventually it will get out + * of step. + * + * So if we play back slightly slower than the server sends for either of + * these reasons then eventually our buffer, and the socket's buffer, will + * fill, and the kernel will start dropping packets. The result is audible + * and not very nice. + * + * Therefore if we're getting behind, we pre-emptively drop silent packets, + * since a change in the duration of a silence is less noticeable than a + * dropped packet from the middle of continuous music. + * + * (If things go wrong the other way then eventually we run out of packets to + * play and are forced to play silence. This doesn't seem to happen in + * practice but if it does then in the same way we can artificially extend + * silent packets to compensate.) + * + * Dropped packets are always logged; use 'disorder-playrtp --monitor' to + * track how close to target buffer occupancy we are on a once-a-minute + * basis. + */ + if(nsamples > minbuffer && silent) { + info("dropping %zu samples (%"PRIu32" > %"PRIu32")", + samples, nsamples, minbuffer); + samples = 0; + } + /* Junk obsolete packets */ + playrtp_next_packet(); + pthread_mutex_unlock(&lock); + return samples; } int main(int argc, char **argv) { - int n; + int n, err; struct addrinfo *res; struct stringlist sl; char *sockname; - int rcvbuf, target_rcvbuf = 131072; + int rcvbuf, target_rcvbuf = 0; socklen_t len; - char *multicast_group = 0; struct ip_mreq mreq; struct ipv6_mreq mreq6; disorder_client *c; char *address, *port; + int is_multicast; + union any_sockaddr { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; + }; + union any_sockaddr mgroup; + const char *dumpfile = 0; + pthread_t ltid; + int monitor = 0; + static const int one = 1; static const struct addrinfo prefs = { - AI_PASSIVE, - PF_INET, - SOCK_DGRAM, - IPPROTO_UDP, - 0, - 0, - 0, - 0 + .ai_flags = AI_PASSIVE, + .ai_family = PF_INET, + .ai_socktype = SOCK_DGRAM, + .ai_protocol = IPPROTO_UDP }; + /* Timing information is often important to debugging playrtp, so we include + * timestamps in the logs */ + logdate = 1; mem_init(); if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale"); - while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:", options, 0)) >= 0) { + backend = uaudio_apis[0]; + while((n = getopt_long(argc, argv, "hVdD:m:x:L:R:aocC:re:P:M", options, 0)) >= 0) { switch(n) { case 'h': help(); - case 'V': version(); + case 'V': version("disorder-playrtp"); case 'd': debugging = 1; break; - case 'D': device = optarg; break; + case 'D': uaudio_set("device", optarg); break; case 'm': minbuffer = 2 * atol(optarg); break; - case 'b': readahead = 2 * atol(optarg); break; case 'x': maxbuffer = 2 * atol(optarg); break; case 'L': logfp = fopen(optarg, "w"); break; case 'R': target_rcvbuf = atoi(optarg); break; - case 'M': multicast_group = optarg; break; #if HAVE_ALSA_ASOUNDLIB_H - case 'a': backend = playrtp_alsa; break; + case 'a': backend = &uaudio_alsa; break; #endif #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST - case 'o': backend = playrtp_oss; break; + case 'o': backend = &uaudio_oss; break; #endif #if HAVE_COREAUDIO_AUDIOHARDWARE_H - case 'c': backend = playrtp_coreaudio; break; + case 'c': backend = &uaudio_coreaudio; break; #endif case 'C': configfile = optarg; break; + case 's': control_socket = optarg; break; + case 'r': dumpfile = optarg; break; + case 'e': backend = &uaudio_command; uaudio_set("command", optarg); break; + case 'P': uaudio_set("pause-mode", optarg); break; + case 'M': monitor = 1; break; default: fatal(0, "invalid option"); } } - if(config_read(0)) fatal(0, "cannot read configuration"); + if(config_read(0, NULL)) fatal(0, "cannot read configuration"); if(!maxbuffer) - maxbuffer = 4 * readahead; + maxbuffer = 2 * minbuffer; argc -= optind; argv += optind; switch(argc) { case 0: - case 1: + /* Get configuration from server */ if(!(c = disorder_new(1))) exit(EXIT_FAILURE); if(disorder_connect(c)) exit(EXIT_FAILURE); if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE); - sl.n = 1; - sl.s = &port; - /* set multicast_group if address is a multicast address */ + sl.n = 2; + sl.s = xcalloc(2, sizeof *sl.s); + sl.s[0] = address; + sl.s[1] = port; break; + case 1: case 2: + /* Use command-line ADDRESS+PORT or just PORT */ sl.n = argc; sl.s = argv; break; default: - fatal(0, "usage: disorder-playrtp [OPTIONS] [ADDRESS [PORT]]"); + fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]"); } - /* Listen for inbound audio data */ + /* Look up address and port */ if(!(res = get_address(&sl, &prefs, &sockname))) exit(1); - info("listening on %s", sockname); + /* Create the socket */ if((rtpfd = socket(res->ai_family, res->ai_socktype, res->ai_protocol)) < 0) fatal(errno, "error creating socket"); - if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) - fatal(errno, "error binding socket to %s", sockname); - if(multicast_group) { - if((n = getaddrinfo(multicast_group, 0, &prefs, &res))) - fatal(0, "getaddrinfo %s: %s", multicast_group, gai_strerror(n)); - switch(res->ai_family) { + /* Allow multiple listeners */ + xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); + is_multicast = multicast(res->ai_addr); + /* The multicast and unicast/broadcast cases are different enough that they + * are totally split. Trying to find commonality between them causes more + * trouble that it's worth. */ + if(is_multicast) { + /* Stash the multicast group address */ + memcpy(&mgroup, res->ai_addr, res->ai_addrlen); + switch(res->ai_addr->sa_family) { + case AF_INET: + mgroup.in.sin_port = 0; + break; + case AF_INET6: + mgroup.in6.sin6_port = 0; + break; + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); + } + /* Bind to to the multicast group address */ + if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) + fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr)); + /* Add multicast group membership */ + switch(mgroup.sa.sa_family) { case PF_INET: - mreq.imr_multiaddr = ((struct sockaddr_in *)res->ai_addr)->sin_addr; + mreq.imr_multiaddr = mgroup.in.sin_addr; mreq.imr_interface.s_addr = 0; /* use primary interface */ if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof mreq) < 0) fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP"); break; case PF_INET6: - mreq6.ipv6mr_multiaddr = ((struct sockaddr_in6 *)res->ai_addr)->sin6_addr; + mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr; memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface); if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq6, sizeof mreq6) < 0) @@ -537,6 +752,31 @@ int main(int argc, char **argv) { default: fatal(0, "unsupported address family %d", res->ai_family); } + /* Report what we did */ + info("listening on %s multicast group %s", + format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa)); + } else { + /* Bind to 0/port */ + switch(res->ai_addr->sa_family) { + case AF_INET: { + struct sockaddr_in *in = (struct sockaddr_in *)res->ai_addr; + + memset(&in->sin_addr, 0, sizeof (struct in_addr)); + break; + } + case AF_INET6: { + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)res->ai_addr; + + memset(&in6->sin6_addr, 0, sizeof (struct in6_addr)); + break; + } + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); + } + if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) + fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr)); + /* Report what we did */ + info("listening on %s", format_sockaddr(res->ai_addr)); } len = sizeof rcvbuf; if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0) @@ -552,9 +792,102 @@ int main(int argc, char **argv) { rcvbuf, target_rcvbuf); } else info("default socket receive buffer %d", rcvbuf); + //info("minbuffer %u maxbuffer %u", minbuffer, maxbuffer); if(logfp) info("WARNING: -L option can impact performance"); - play_rtp(); + if(control_socket) { + pthread_t tid; + + if((err = pthread_create(&tid, 0, control_thread, 0))) + fatal(err, "pthread_create control_thread"); + } + if(dumpfile) { + int fd; + unsigned char buffer[65536]; + size_t written; + + if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0) + fatal(errno, "opening %s", dumpfile); + /* Fill with 0s to a suitable size */ + memset(buffer, 0, sizeof buffer); + for(written = 0; written < dump_size * sizeof(int16_t); + written += sizeof buffer) { + if(write(fd, buffer, sizeof buffer) < 0) + fatal(errno, "clearing %s", dumpfile); + } + /* Map the buffer into memory for convenience */ + dump_buffer = mmap(0, dump_size * sizeof(int16_t), PROT_READ|PROT_WRITE, + MAP_SHARED, fd, 0); + if(dump_buffer == (void *)-1) + fatal(errno, "mapping %s", dumpfile); + info("dumping to %s", dumpfile); + } + /* Set up output. Currently we only support L16 so there's no harm setting + * the format before we know what it is! */ + uaudio_set_format(44100/*Hz*/, 2/*channels*/, + 16/*bits/channel*/, 1/*signed*/); + backend->start(playrtp_callback, NULL); + /* We receive and convert audio data in a background thread */ + if((err = pthread_create(<id, 0, listen_thread, 0))) + fatal(err, "pthread_create listen_thread"); + /* We have a second thread to add received packets to the queue */ + if((err = pthread_create(<id, 0, queue_thread, 0))) + fatal(err, "pthread_create queue_thread"); + pthread_mutex_lock(&lock); + time_t lastlog = 0; + for(;;) { + /* Wait for the buffer to fill up a bit */ + playrtp_fill_buffer(); + /* Start playing now */ + info("Playing..."); + next_timestamp = pheap_first(&packets)->timestamp; + active = 1; + pthread_mutex_unlock(&lock); + backend->activate(); + pthread_mutex_lock(&lock); + /* Wait until the buffer empties out + * + * If there's a packet that we can play right now then we definitely + * continue. + * + * Also if there's at least minbuffer samples we carry on regardless and + * insert silence. The assumption is there's been a pause but more data + * is now available. + */ + while(nsamples >= minbuffer + || (nsamples > 0 + && contains(pheap_first(&packets), next_timestamp))) { + if(monitor) { + time_t now = xtime(0); + + if(now >= lastlog + 60) { + int offset = nsamples - minbuffer; + double offtime = (double)offset / (uaudio_rate * uaudio_channels); + info("%+d samples off (%d.%02ds, %d bytes)", + offset, + (int)fabs(offtime) * (offtime < 0 ? -1 : 1), + (int)(fabs(offtime) * 100) % 100, + offset * uaudio_bits / CHAR_BIT); + lastlog = now; + } + } + //fprintf(stderr, "%8u/%u (%u) PLAYING\n", nsamples, maxbuffer, minbuffer); + pthread_cond_wait(&cond, &lock); + } +#if 0 + if(nsamples) { + struct packet *p = pheap_first(&packets); + fprintf(stderr, "nsamples=%u (%u) next_timestamp=%"PRIx32", first packet is [%"PRIx32",%"PRIx32")\n", + nsamples, minbuffer, next_timestamp,p->timestamp,p->timestamp+p->nsamples); + } +#endif + /* Stop playing for a bit until the buffer re-fills */ + pthread_mutex_unlock(&lock); + backend->deactivate(); + pthread_mutex_lock(&lock); + active = 0; + /* Go back round */ + } return 0; }