chiark / gitweb /
playrtp: support multiple unicast mode
[disorder] / clients / playrtp.c
index c6f0ad9ee2a9d3c0ca574cfa5f2cee6b1ca59e11..89e458e48504c471f327b0171b81bfe1099b91c9 100644 (file)
@@ -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
 #include <unistd.h>
 #include <sys/mman.h>
 #include <fcntl.h>
+#include <math.h>
+#include <arpa/inet.h>
+#include <ifaddrs.h>
+#include <net/if.h>
 
 #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 high watermark
- *
- * We'll only start playing when this many samples are available. */
-static unsigned readahead = 44100;      /* 0.5 seconds */
+/** @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
@@ -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,11 +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 }
 };
 
@@ -254,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);
@@ -267,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"))
@@ -285,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);
   }
 }
 
@@ -331,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
@@ -375,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;
     }
@@ -406,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);
@@ -440,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;
 }
@@ -480,19 +493,22 @@ 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
+          "  --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"
@@ -507,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 */
@@ -537,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
@@ -549,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) {
@@ -559,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 {
@@ -583,6 +686,7 @@ 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 = {
@@ -592,40 +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:P:", 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;
     case 'P': uaudio_set("pause-mode", optarg); break;
-    default: fatal(0, "invalid option");
+    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) {
@@ -646,107 +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");
-  /* 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 = 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);
+  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;
     }
+    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 */
-    info("listening on %s multicast group %s",
-         format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
+    disorder_info("listening on %s", format_sockaddr(addr));
   } else {
-    /* Bind to 0/port */
-    switch(res->ai_addr->sa_family) {
-    case AF_INET: {
-      struct sockaddr_in *in = (struct sockaddr_in *)res->ai_addr;
+    /* 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));
-      if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
-        fatal(errno, "error binding socket to 0.0.0.0 port %d",
-              ntohs(in->sin_port));
-      break;
-    }
-    case AF_INET6: {
-      struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)res->ai_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:
-      fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
+        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));
     }
-    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)
-    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;
@@ -754,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(&ltid, 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(&ltid, 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();