X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/d4170ca70fd4f87f77fa2213219eb30f89d3d0d0..cca034e58b00be2f53f9145957346bcd3e451c45:/clients/playrtp.c diff --git a/clients/playrtp.c b/clients/playrtp.c index 92cbb75..89e458e 100644 --- a/clients/playrtp.c +++ b/clients/playrtp.c @@ -1,6 +1,6 @@ /* * This file is part of DisOrder. - * Copyright (C) 2007, 2008 Richard Kettlewell + * Copyright (C) 2007-2009, 2011, 2013 Richard Kettlewell * * 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 @@ -24,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: * - * 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. + * 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. * - * 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. + * 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). + * + * control_thread() accepts commands from Disobedience (or anything else). + * + * 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 @@ -64,6 +64,10 @@ #include #include #include +#include +#include +#include +#include #include "log.h" #include "mem.h" @@ -81,8 +85,6 @@ #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 @@ -96,19 +98,13 @@ static FILE *logfp; /** @brief Output 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 low watermark in samples */ +unsigned minbuffer = 4 * (2 * 44100) / 10; /* 0.4 seconds */ -/** @brief Buffer high watermark +/** @brief Maximum buffer size in samples * - * We'll only start playing when this many samples are available. */ -static unsigned readahead = 44100; /* 0.5 seconds */ - -/** @brief Maximum buffer size - * - * 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 @@ -204,7 +200,6 @@ 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' }, #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST { "oss", no_argument, 0, 'o' }, @@ -215,10 +210,13 @@ static const struct option options[] = { #if HAVE_COREAUDIO_AUDIOHARDWARE_H { "core-audio", no_argument, 0, 'c' }, #endif + { "api", required_argument, 0, 'A' }, { "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 } }; @@ -253,10 +251,10 @@ static void *control_thread(void attribute((unused)) *arg) { 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); + disorder_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); + disorder_fatal(errno, "error calling listen on %s", control_socket); + disorder_info("listening on %s", control_socket); for(;;) { salen = sizeof sa; cfd = accept(sfd, (struct sockaddr *)&sa, &salen); @@ -266,17 +264,17 @@ static void *control_thread(void attribute((unused)) *arg) { case EAGAIN: break; default: - fatal(errno, "error calling accept on %s", control_socket); + disorder_fatal(errno, "error calling accept on %s", control_socket); } } if(!(fp = fdopen(cfd, "r+"))) { - error(errno, "error calling fdopen for %s connection", control_socket); + disorder_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); + disorder_info("stopped via %s", control_socket); exit(0); /* terminate immediately */ } if(!strcmp(line, "query")) @@ -284,7 +282,7 @@ static void *control_thread(void attribute((unused)) *arg) { xfree(line); } if(fclose(fp) < 0) - error(errno, "error closing %s connection", control_socket); + disorder_error(errno, "error closing %s connection", control_socket); } } @@ -330,6 +328,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 @@ -374,19 +375,19 @@ static void *listen_thread(void attribute((unused)) *arg) { case EINTR: continue; default: - fatal(errno, "error reading from socket"); + disorder_fatal(errno, "error reading from socket"); } } /* Ignore too-short packets */ if((size_t)n <= sizeof (struct rtp_header)) { - info("ignored a short packet"); + disorder_info("ignored a short packet"); continue; } timestamp = htonl(header.timestamp); seq = htons(header.seq); /* Ignore packets in the past */ if(active && lt(timestamp, next_timestamp)) { - info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32, + disorder_info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32, timestamp, next_timestamp); continue; } @@ -405,9 +406,16 @@ static void *listen_thread(void attribute((unused)) *arg) { break; /* TODO support other RFC3551 media types (when the speaker does) */ default: - fatal(0, "unsupported RTP payload type %d", - header.mpt & 0x7F); + disorder_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); @@ -439,12 +447,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) { + } + disorder_info("Buffering..."); + /* 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; } @@ -479,20 +493,25 @@ static void help(void) { "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" " --config, -C PATH Set configuration file\n" -#if HAVE_ALSA_ASOUNDLIB_H - " --alsa, -a Use ALSA to play audio\n" -#endif -#if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST - " --oss, -o Use OSS to play audio\n" -#endif -#if HAVE_COREAUDIO_AUDIOHARDWARE_H - " --core-audio, -c Use Core Audio to play audio\n" -#endif - " --command, -e COMMAND Pipe audio to command\n" + " --api, -A API Select audio API. Possibilities:\n" + " "); + int first = 1; + for(int n = 0; uaudio_apis[n]; ++n) { + if(uaudio_apis[n]->flags & UAUDIO_API_CLIENT) { + if(first) + first = 0; + else + xprintf(", "); + xprintf("%s", uaudio_apis[n]->name); + } + } + xprintf("\n" + " --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" ); @@ -504,6 +523,7 @@ 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 */ @@ -534,8 +554,7 @@ static size_t playrtp_callback(void *buffer, *bufptr++ = (int16_t)ntohs(*ptr++); --i; } - /* We don't junk the packet here; a subsequent call to - * playrtp_next_packet() will dispose of it (if it's actually done with). */ + 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 @@ -546,6 +565,7 @@ static size_t playrtp_callback(void *buffer, samples = max_samples; //info("infill by %zu", samples); memset(buffer, 0, samples * uaudio_sample_size); + silent = 1; } /* Debug dump */ if(dump_buffer) { @@ -556,20 +576,106 @@ static size_t playrtp_callback(void *buffer, } /* 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) { + disorder_info("dropping %zu samples (%"PRIu32" > %"PRIu32")", + samples, nsamples, minbuffer); + samples = 0; + } + /* Junk obsolete packets */ + playrtp_next_packet(); pthread_mutex_unlock(&lock); return samples; } +static int compare_family(const struct ifaddrs *a, + const struct ifaddrs *b, + int family) { + int afamily = a->ifa_addr->sa_family; + int bfamily = b->ifa_addr->sa_family; + if(afamily != bfamily) { + /* Preferred family wins */ + if(afamily == family) return 1; + if(bfamily == family) return -1; + /* Either there's no preference or it doesn't help. Prefer IPv4 */ + if(afamily == AF_INET) return 1; + if(bfamily == AF_INET) return -1; + /* Failing that prefer IPv6 */ + if(afamily == AF_INET6) return 1; + if(bfamily == AF_INET6) return -1; + } + return 0; +} + +static int compare_flags(const struct ifaddrs *a, + const struct ifaddrs *b) { + unsigned aflags = a->ifa_flags, bflags = b->ifa_flags; + /* Up interfaces are better than down ones */ + unsigned aup = aflags & IFF_UP, bup = bflags & IFF_UP; + if(aup != bup) + return aup > bup ? 1 : -1; + /* Static addresses are better than dynamic */ + unsigned adynamic = aflags & IFF_DYNAMIC, bdynamic = bflags & IFF_DYNAMIC; + if(adynamic != bdynamic) + return adynamic < bdynamic ? 1 : -1; + unsigned aloopback = aflags & IFF_LOOPBACK, bloopback = bflags & IFF_LOOPBACK; + /* Static addresses are better than dynamic */ + if(aloopback != bloopback) + return aloopback < bloopback ? 1 : -1; + return 0; +} + +static int compare_interfaces(const struct ifaddrs *a, + const struct ifaddrs *b, + int family) { + int c; + if((c = compare_family(a, b, family))) return c; + return compare_flags(a, b); +} + int main(int argc, char **argv) { int n, err; struct addrinfo *res; struct stringlist sl; char *sockname; - int rcvbuf, target_rcvbuf = 131072; + int rcvbuf, target_rcvbuf = 0; socklen_t len; struct ip_mreq mreq; struct ipv6_mreq mreq6; - disorder_client *c; + disorder_client *c = NULL; char *address, *port; int is_multicast; union any_sockaddr { @@ -580,6 +686,8 @@ int main(int argc, char **argv) { 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_flags = AI_PASSIVE, @@ -588,39 +696,63 @@ int main(int argc, char **argv) { .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"); - backend = uaudio_apis[0]; - while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:re:", options, 0)) >= 0) { + if(!setlocale(LC_CTYPE, "")) disorder_fatal(errno, "error calling setlocale"); + while((n = getopt_long(argc, argv, "hVdD:m:x:L:R:aocC:re:P:MA:", options, 0)) >= 0) { switch(n) { case 'h': help(); case 'V': version("disorder-playrtp"); case 'd': debugging = 1; 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; #if HAVE_ALSA_ASOUNDLIB_H - case 'a': backend = &uaudio_alsa; break; + case 'a': + disorder_error(0, "deprecated option; use --api alsa instead"); + backend = &uaudio_alsa; break; #endif #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST - case 'o': backend = &uaudio_oss; break; + case 'o': + disorder_error(0, "deprecated option; use --api oss instead"); + backend = &uaudio_oss; + break; #endif #if HAVE_COREAUDIO_AUDIOHARDWARE_H - case 'c': backend = &uaudio_coreaudio; break; + case 'c': + disorder_error(0, "deprecated option; use --api coreaudio instead"); + backend = &uaudio_coreaudio; + break; #endif + case 'A': backend = uaudio_find(optarg); break; 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; - default: fatal(0, "invalid option"); + case 'P': uaudio_set("pause-mode", optarg); break; + case 'M': monitor = 1; break; + default: disorder_fatal(0, "invalid option"); } } - if(config_read(0)) fatal(0, "cannot read configuration"); + if(config_read(0, NULL)) disorder_fatal(0, "cannot read configuration"); + if(!backend) { + backend = uaudio_default(uaudio_apis, UAUDIO_API_CLIENT); + if(!backend) + disorder_fatal(0, "no default uaudio API found"); + disorder_info("default audio API %s", backend->name); + } + if(backend == &uaudio_rtp) { + /* This means that you have NO local sound output. This can happen if you + * use a non-Apple GCC on a Mac (because it doesn't know how to compile + * CoreAudio/AudioHardware.h). */ + disorder_fatal(0, "cannot play RTP through RTP"); + } if(!maxbuffer) - maxbuffer = 4 * readahead; + maxbuffer = 2 * minbuffer; argc -= optind; argv += optind; switch(argc) { @@ -641,87 +773,157 @@ int main(int argc, char **argv) { sl.s = argv; break; default: - fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]"); + disorder_fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]"); } - /* Look up address and port */ - if(!(res = get_address(&sl, &prefs, &sockname))) - exit(1); - /* Create the socket */ - if((rtpfd = socket(res->ai_family, - res->ai_socktype, - res->ai_protocol)) < 0) - fatal(errno, "error creating socket"); - /* Stash the multicast group address */ - if((is_multicast = multicast(res->ai_addr))) { - 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; + disorder_info("version "VERSION" process ID %lu", + (unsigned long)getpid()); + struct sockaddr *addr; + socklen_t addr_len; + if(!strcmp(sl.s[0], "-")) { + /* Pick address family to match known-working connectivity to the server */ + int family = disorder_client_af(c); + /* Get a list of interfaces */ + struct ifaddrs *ifa, *bestifa = NULL; + if(getifaddrs(&ifa) < 0) + disorder_fatal(errno, "error calling getifaddrs"); + /* Try to pick a good one */ + for(; ifa; ifa = ifa->ifa_next) { + if(bestifa == NULL + || compare_interfaces(ifa, bestifa, family) > 0) + bestifa = ifa; } - } - /* Bind to 0/port */ - switch(res->ai_addr->sa_family) { - case AF_INET: - memset(&((struct sockaddr_in *)res->ai_addr)->sin_addr, 0, - sizeof (struct in_addr)); - break; - case AF_INET6: - memset(&((struct sockaddr_in6 *)res->ai_addr)->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", sockname); - if(is_multicast) { - switch(mgroup.sa.sa_family) { - case PF_INET: - 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 = 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) - fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP"); - break; - default: - fatal(0, "unsupported address family %d", res->ai_family); + if(!bestifa) + disorder_fatal(0, "failed to select a network interface"); + family = bestifa->ifa_addr->sa_family; + if((rtpfd = socket(family, + SOCK_DGRAM, + IPPROTO_UDP)) < 0) + disorder_fatal(errno, "error creating socket (family %d)", family); + /* Bind the address */ + if(bind(rtpfd, bestifa->ifa_addr, + family == AF_INET + ? sizeof (struct sockaddr_in) : sizeof (struct sockaddr_in6)) < 0) + disorder_fatal(errno, "error binding socket"); + static struct sockaddr_storage bound_address; + addr = (struct sockaddr *)&bound_address; + addr_len = sizeof bound_address; + if(getsockname(rtpfd, addr, &addr_len) < 0) + disorder_fatal(errno, "error getting socket address"); + /* Convert to string */ + char addrname[128], portname[32]; + if(getnameinfo(addr, addr_len, + addrname, sizeof addrname, + portname, sizeof portname, + NI_NUMERICHOST|NI_NUMERICSERV) < 0) + disorder_fatal(errno, "getnameinfo"); + /* Ask for audio data */ + if(disorder_rtp_request(c, addrname, portname)) exit(EXIT_FAILURE); + /* Report what we did */ + disorder_info("listening on %s", format_sockaddr(addr)); + } else { + /* Look up address and port */ + if(!(res = get_address(&sl, &prefs, &sockname))) + exit(1); + addr = res->ai_addr; + addr_len = res->ai_addrlen; + /* Create the socket */ + if((rtpfd = socket(res->ai_family, + res->ai_socktype, + res->ai_protocol)) < 0) + disorder_fatal(errno, "error creating socket"); + /* Allow multiple listeners */ + xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); + is_multicast = multicast(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, addr, addr_len); + 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: + disorder_fatal(0, "unsupported address family %d", + (int)addr->sa_family); + } + /* Bind to to the multicast group address */ + if(bind(rtpfd, addr, addr_len) < 0) + disorder_fatal(errno, "error binding socket to %s", + format_sockaddr(addr)); + /* Add multicast group membership */ + switch(mgroup.sa.sa_family) { + case PF_INET: + 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) + disorder_fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP"); + break; + case PF_INET6: + 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) + disorder_fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP"); + break; + default: + disorder_fatal(0, "unsupported address family %d", res->ai_family); + } + /* Report what we did */ + disorder_info("listening on %s multicast group %s", + format_sockaddr(addr), format_sockaddr(&mgroup.sa)); + } else { + /* Bind to 0/port */ + switch(addr->sa_family) { + case AF_INET: { + struct sockaddr_in *in = (struct sockaddr_in *)addr; + + memset(&in->sin_addr, 0, sizeof (struct in_addr)); + break; + } + case AF_INET6: { + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr; + + memset(&in6->sin6_addr, 0, sizeof (struct in6_addr)); + break; + } + default: + disorder_fatal(0, "unsupported family %d", (int)addr->sa_family); + } + if(bind(rtpfd, addr, addr_len) < 0) + disorder_fatal(errno, "error binding socket to %s", + format_sockaddr(addr)); + /* Report what we did */ + disorder_info("listening on %s", format_sockaddr(addr)); } - info("listening on %s multicast group %s", - format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa)); - } else - info("listening on %s", format_sockaddr(res->ai_addr)); + } len = sizeof rcvbuf; if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0) - fatal(errno, "error calling getsockopt SO_RCVBUF"); + disorder_fatal(errno, "error calling getsockopt SO_RCVBUF"); if(target_rcvbuf > rcvbuf) { if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &target_rcvbuf, sizeof target_rcvbuf) < 0) - error(errno, "error calling setsockopt SO_RCVBUF %d", - target_rcvbuf); + disorder_error(errno, "error calling setsockopt SO_RCVBUF %d", + target_rcvbuf); /* We try to carry on anyway */ else - info("changed socket receive buffer from %d to %d", - rcvbuf, target_rcvbuf); + disorder_info("changed socket receive buffer from %d to %d", + rcvbuf, target_rcvbuf); } else - info("default socket receive buffer %d", rcvbuf); + disorder_info("default socket receive buffer %d", rcvbuf); + //info("minbuffer %u maxbuffer %u", minbuffer, maxbuffer); if(logfp) - info("WARNING: -L option can impact performance"); + disorder_info("WARNING: -L option can impact performance"); if(control_socket) { pthread_t tid; if((err = pthread_create(&tid, 0, control_thread, 0))) - fatal(err, "pthread_create control_thread"); + disorder_fatal(err, "pthread_create control_thread"); } if(dumpfile) { int fd; @@ -729,49 +931,81 @@ int main(int argc, char **argv) { size_t written; if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0) - fatal(errno, "opening %s", dumpfile); + disorder_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); + disorder_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); + disorder_fatal(errno, "mapping %s", dumpfile); + disorder_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*/); + uaudio_set("application", "disorder-playrtp"); 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"); + disorder_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"); + disorder_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..."); + disorder_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 */ + /* 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); + disorder_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();