chiark - git - mdw - disorder/blob - clients/playrtp.c

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