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

   1 /*
   2  * This file is part of DisOrder.
   3  * Copyright (C) 2007-2009, 2011, 2013 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 #include <math.h>
  68 #include <arpa/inet.h>
  69 #include <ifaddrs.h>
  70 #include <net/if.h>
  71
  72 #include "log.h"
  73 #include "mem.h"
  74 #include "configuration.h"
  75 #include "addr.h"
  76 #include "syscalls.h"
  77 #include "printf.h"
  78 #include "rtp.h"
  79 #include "defs.h"
  80 #include "vector.h"
  81 #include "heap.h"
  82 #include "timeval.h"
  83 #include "client.h"
  84 #include "playrtp.h"
  85 #include "inputline.h"
  86 #include "version.h"
  87 #include "uaudio.h"
  88
  89 /** @brief Obsolete synonym */
  90 #ifndef IPV6_JOIN_GROUP
  91 # define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
  92 #endif
  93
  94 /** @brief RTP socket */
  95 static int rtpfd;
  96
  97 /** @brief Log output */
  98 static FILE *logfp;
  99
 100 /** @brief Output device */
 101
 102 /** @brief Buffer low watermark in samples */
 103 unsigned minbuffer;
 104
 105 /** @brief Maximum buffer size in samples
 106  *
 107  * We'll stop reading from the network if we have this many samples.
 108  */
 109 static unsigned maxbuffer;
 110
 111 /** @brief Received packets
 112  * Protected by @ref receive_lock
 113  *
 114  * Received packets are added to this list, and queue_thread() picks them off
 115  * it and adds them to @ref packets.  Whenever a packet is added to it, @ref
 116  * receive_cond is signalled.
 117  */
 118 struct packet *received_packets;
 119
 120 /** @brief Tail of @ref received_packets
 121  * Protected by @ref receive_lock
 122  */
 123 struct packet **received_tail = &received_packets;
 124
 125 /** @brief Lock protecting @ref received_packets
 126  *
 127  * Only listen_thread() and queue_thread() ever hold this lock.  It is vital
 128  * that queue_thread() not hold it any longer than it strictly has to. */
 129 pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
 130
 131 /** @brief Condition variable signalled when @ref received_packets is updated
 132  *
 133  * Used by listen_thread() to notify queue_thread() that it has added another
 134  * packet to @ref received_packets. */
 135 pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
 136
 137 /** @brief Length of @ref received_packets */
 138 uint32_t nreceived;
 139
 140 /** @brief Binary heap of received packets */
 141 struct pheap packets;
 142
 143 /** @brief Total number of samples available
 144  *
 145  * We make this volatile because we inspect it without a protecting lock,
 146  * so the usual pthread_* guarantees aren't available.
 147  */
 148 volatile uint32_t nsamples;
 149
 150 /** @brief Timestamp of next packet to play.
 151  *
 152  * This is set to the timestamp of the last packet, plus the number of
 153  * samples it contained.  Only valid if @ref active is nonzero.
 154  */
 155 uint32_t next_timestamp;
 156
 157 /** @brief True if actively playing
 158  *
 159  * This is true when playing and false when just buffering. */
 160 int active;
 161
 162 /** @brief Lock protecting @ref packets */
 163 pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
 164
 165 /** @brief Condition variable signalled whenever @ref packets is changed */
 166 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 167
 168 /** @brief Backend to play with */
 169 static const struct uaudio *backend;
 170
 171 HEAP_DEFINE(pheap, struct packet *, lt_packet);
 172
 173 /** @brief Control socket or NULL */
 174 const char *control_socket;
 175
 176 /** @brief Buffer for debugging dump
 177  *
 178  * The debug dump is enabled by the @c --dump option.  It records the last 20s
 179  * of audio to the specified file (which will be about 3.5Mbytes).  The file is
 180  * written as as ring buffer, so the start point will progress through it.
 181  *
 182  * Use clients/dump2wav to convert this to a WAV file, which can then be loaded
 183  * into (e.g.) Audacity for further inspection.
 184  *
 185  * All three backends (ALSA, OSS, Core Audio) now support this option.
 186  *
 187  * The idea is to allow the user a few seconds to react to an audible artefact.
 188  */
 189 int16_t *dump_buffer;
 190
 191 /** @brief Current index within debugging dump */
 192 size_t dump_index;
 193
 194 /** @brief Size of debugging dump in samples */
 195 size_t dump_size = 44100/*Hz*/ * 2/*channels*/ * 20/*seconds*/;
 196
 197 static const struct option options[] = {
 198   { "help", no_argument, 0, 'h' },
 199   { "version", no_argument, 0, 'V' },
 200   { "debug", no_argument, 0, 'd' },
 201   { "device", required_argument, 0, 'D' },
 202   { "min", required_argument, 0, 'm' },
 203   { "max", required_argument, 0, 'x' },
 204   { "rcvbuf", required_argument, 0, 'R' },
 205 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
 206   { "oss", no_argument, 0, 'o' },
 207 #endif
 208 #if HAVE_ALSA_ASOUNDLIB_H
 209   { "alsa", no_argument, 0, 'a' },
 210 #endif
 211 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
 212   { "core-audio", no_argument, 0, 'c' },
 213 #endif
 214   { "api", required_argument, 0, 'A' },
 215   { "dump", required_argument, 0, 'r' },
 216   { "command", required_argument, 0, 'e' },
 217   { "pause-mode", required_argument, 0, 'P' },
 218   { "socket", required_argument, 0, 's' },
 219   { "config", required_argument, 0, 'C' },
 220   { "monitor", no_argument, 0, 'M' },
 221   { 0, 0, 0, 0 }
 222 };
 223
 224 /** @brief Control thread
 225  *
 226  * This thread is responsible for accepting control commands from Disobedience
 227  * (or other controllers) over an AF_UNIX stream socket with a path specified
 228  * by the @c --socket option.  The protocol uses simple string commands and
 229  * replies:
 230  *
 231  * - @c stop will shut the player down
 232  * - @c query will send back the reply @c running
 233  * - anything else is ignored
 234  *
 235  * Commands and response strings terminated by shutting down the connection or
 236  * by a newline.  No attempt is made to multiplex multiple clients so it is
 237  * important that the command be sent as soon as the connection is made - it is
 238  * assumed that both parties to the protocol are entirely cooperating with one
 239  * another.
 240  */
 241 static void *control_thread(void attribute((unused)) *arg) {
 242   struct sockaddr_un sa;
 243   int sfd, cfd;
 244   char *line;
 245   socklen_t salen;
 246   FILE *fp;
 247   int vl, vr;
 248
 249   assert(control_socket);
 250   unlink(control_socket);
 251   memset(&sa, 0, sizeof sa);
 252   sa.sun_family = AF_UNIX;
 253   strcpy(sa.sun_path, control_socket);
 254   sfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
 255   if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0)
 256     disorder_fatal(errno, "error binding to %s", control_socket);
 257   if(listen(sfd, 128) < 0)
 258     disorder_fatal(errno, "error calling listen on %s", control_socket);
 259   disorder_info("listening on %s", control_socket);
 260   for(;;) {
 261     salen = sizeof sa;
 262     cfd = accept(sfd, (struct sockaddr *)&sa, &salen);
 263     if(cfd < 0) {
 264       switch(errno) {
 265       case EINTR:
 266       case EAGAIN:
 267         break;
 268       default:
 269         disorder_fatal(errno, "error calling accept on %s", control_socket);
 270       }
 271     }
 272     if(!(fp = fdopen(cfd, "r+"))) {
 273       disorder_error(errno, "error calling fdopen for %s connection", control_socket);
 274       close(cfd);
 275       continue;
 276     }
 277     if(!inputline(control_socket, fp, &line, '\n')) {
 278       if(!strcmp(line, "stop")) {
 279         disorder_info("stopped via %s", control_socket);
 280         exit(0);                          /* terminate immediately */
 281       } else if(!strcmp(line, "query"))
 282         fprintf(fp, "running");
 283       else if(!strcmp(line, "getvol")) {
 284         if(backend->get_volume) backend->get_volume(&vl, &vr);
 285         else vl = vr = 0;
 286         fprintf(fp, "%d %d\n", vl, vr);
 287       } else if(!strncmp(line, "setvol ", 7)) {
 288         if(!backend->set_volume)
 289           vl = vr = 0;
 290         else if(sscanf(line + 7, "%d %d", &vl, &vr) == 2)
 291           backend->set_volume(&vl, &vr);
 292         else
 293           backend->get_volume(&vl, &vr);
 294         fprintf(fp, "%d %d\n", vl, vr);
 295       }
 296       xfree(line);
 297     }
 298     if(fclose(fp) < 0)
 299       disorder_error(errno, "error closing %s connection", control_socket);
 300   }
 301 }
 302
 303 /** @brief Drop the first packet
 304  *
 305  * Assumes that @ref lock is held.
 306  */
 307 static void drop_first_packet(void) {
 308   if(pheap_count(&packets)) {
 309     struct packet *const p = pheap_remove(&packets);
 310     nsamples -= p->nsamples;
 311     playrtp_free_packet(p);
 312     pthread_cond_broadcast(&cond);
 313   }
 314 }
 315
 316 /** @brief Background thread adding packets to heap
 317  *
 318  * This just transfers packets from @ref received_packets to @ref packets.  It
 319  * is important that it holds @ref receive_lock for as little time as possible,
 320  * in order to minimize the interval between calls to read() in
 321  * listen_thread().
 322  */
 323 static void *queue_thread(void attribute((unused)) *arg) {
 324   struct packet *p;
 325
 326   for(;;) {
 327     /* Get the next packet */
 328     pthread_mutex_lock(&receive_lock);
 329     while(!received_packets) {
 330       pthread_cond_wait(&receive_cond, &receive_lock);
 331     }
 332     p = received_packets;
 333     received_packets = p->next;
 334     if(!received_packets)
 335       received_tail = &received_packets;
 336     --nreceived;
 337     pthread_mutex_unlock(&receive_lock);
 338     /* Add it to the heap */
 339     pthread_mutex_lock(&lock);
 340     pheap_insert(&packets, p);
 341     nsamples += p->nsamples;
 342     pthread_cond_broadcast(&cond);
 343     pthread_mutex_unlock(&lock);
 344   }
 345 #if HAVE_STUPID_GCC44
 346   return NULL;
 347 #endif
 348 }
 349
 350 /** @brief Background thread collecting samples
 351  *
 352  * This function collects samples, perhaps converts them to the target format,
 353  * and adds them to the packet list.
 354  *
 355  * It is crucial that the gap between successive calls to read() is as small as
 356  * possible: otherwise packets will be dropped.
 357  *
 358  * We use a binary heap to ensure that the unavoidable effort is at worst
 359  * logarithmic in the total number of packets - in fact if packets are mostly
 360  * received in order then we will largely do constant work per packet since the
 361  * newest packet will always be last.
 362  *
 363  * Of more concern is that we must acquire the lock on the heap to add a packet
 364  * to it.  If this proves a problem in practice then the answer would be
 365  * (probably doubly) linked list with new packets added the end and a second
 366  * thread which reads packets off the list and adds them to the heap.
 367  *
 368  * We keep memory allocation (mostly) very fast by keeping pre-allocated
 369  * packets around; see @ref playrtp_new_packet().
 370  */
 371 static void *listen_thread(void attribute((unused)) *arg) {
 372   struct packet *p = 0;
 373   int n;
 374   struct rtp_header header;
 375   uint16_t seq;
 376   uint32_t timestamp;
 377   struct iovec iov[2];
 378
 379   for(;;) {
 380     if(!p)
 381       p = playrtp_new_packet();
 382     iov[0].iov_base = &header;
 383     iov[0].iov_len = sizeof header;
 384     iov[1].iov_base = p->samples_raw;
 385     iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
 386     n = readv(rtpfd, iov, 2);
 387     if(n < 0) {
 388       switch(errno) {
 389       case EINTR:
 390         continue;
 391       default:
 392         disorder_fatal(errno, "error reading from socket");
 393       }
 394     }
 395     /* Ignore too-short packets */
 396     if((size_t)n <= sizeof (struct rtp_header)) {
 397       disorder_info("ignored a short packet");
 398       continue;
 399     }
 400     timestamp = htonl(header.timestamp);
 401     seq = htons(header.seq);
 402     /* Ignore packets in the past */
 403     if(active && lt(timestamp, next_timestamp)) {
 404       disorder_info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32,
 405            timestamp, next_timestamp);
 406       continue;
 407     }
 408     /* Ignore packets with the extension bit set. */
 409     if(header.vpxcc & 0x10)
 410       continue;
 411     p->next = 0;
 412     p->flags = 0;
 413     p->timestamp = timestamp;
 414     /* Convert to target format */
 415     if(header.mpt & 0x80)
 416       p->flags |= IDLE;
 417     switch(header.mpt & 0x7F) {
 418     case 10:                            /* L16 */
 419       p->nsamples = (n - sizeof header) / sizeof(uint16_t);
 420       break;
 421       /* TODO support other RFC3551 media types (when the speaker does) */
 422     default:
 423       disorder_fatal(0, "unsupported RTP payload type %d", header.mpt & 0x7F);
 424     }
 425     /* See if packet is silent */
 426     const uint16_t *s = p->samples_raw;
 427     n = p->nsamples;
 428     for(; n > 0; --n)
 429       if(*s++)
 430         break;
 431     if(!n)
 432       p->flags |= SILENT;
 433     if(logfp)
 434       fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n",
 435               seq, timestamp, p->nsamples, timestamp + p->nsamples);
 436     /* Stop reading if we've reached the maximum.
 437      *
 438      * This is rather unsatisfactory: it means that if packets get heavily
 439      * out of order then we guarantee dropouts.  But for now... */
 440     if(nsamples >= maxbuffer) {
 441       pthread_mutex_lock(&lock);
 442       while(nsamples >= maxbuffer) {
 443         pthread_cond_wait(&cond, &lock);
 444       }
 445       pthread_mutex_unlock(&lock);
 446     }
 447     /* Add the packet to the receive queue */
 448     pthread_mutex_lock(&receive_lock);
 449     *received_tail = p;
 450     received_tail = &p->next;
 451     ++nreceived;
 452     pthread_cond_signal(&receive_cond);
 453     pthread_mutex_unlock(&receive_lock);
 454     /* We'll need a new packet */
 455     p = 0;
 456   }
 457 }
 458
 459 /** @brief Wait until the buffer is adequately full
 460  *
 461  * Must be called with @ref lock held.
 462  */
 463 void playrtp_fill_buffer(void) {
 464   /* Discard current buffer contents */
 465   while(nsamples) {
 466     //fprintf(stderr, "%8u/%u (%u) DROPPING\n", nsamples, maxbuffer, minbuffer);
 467     drop_first_packet();
 468   }
 469   disorder_info("Buffering...");
 470   /* Wait until there's at least minbuffer samples available */
 471   while(nsamples < minbuffer) {
 472     //fprintf(stderr, "%8u/%u (%u) FILLING\n", nsamples, maxbuffer, minbuffer);
 473     pthread_cond_wait(&cond, &lock);
 474   }
 475   /* Start from whatever is earliest */
 476   next_timestamp = pheap_first(&packets)->timestamp;
 477   active = 1;
 478 }
 479
 480 /** @brief Find next packet
 481  * @return Packet to play or NULL if none found
 482  *
 483  * The return packet is merely guaranteed not to be in the past: it might be
 484  * the first packet in the future rather than one that is actually suitable to
 485  * play.
 486  *
 487  * Must be called with @ref lock held.
 488  */
 489 struct packet *playrtp_next_packet(void) {
 490   while(pheap_count(&packets)) {
 491     struct packet *const p = pheap_first(&packets);
 492     if(le(p->timestamp + p->nsamples, next_timestamp)) {
 493       /* This packet is in the past.  Drop it and try another one. */
 494       drop_first_packet();
 495     } else
 496       /* This packet is NOT in the past.  (It might be in the future
 497        * however.) */
 498       return p;
 499   }
 500   return 0;
 501 }
 502
 503 /* display usage message and terminate */
 504 static void attribute((noreturn)) help(void) {
 505   xprintf("Usage:\n"
 506           "  disorder-playrtp [OPTIONS] [[ADDRESS] PORT]\n"
 507           "Options:\n"
 508           "  --device, -D DEVICE     Output device\n"
 509           "  --min, -m FRAMES        Buffer low water mark\n"
 510           "  --max, -x FRAMES        Buffer maximum size\n"
 511           "  --rcvbuf, -R BYTES      Socket receive buffer size\n"
 512           "  --config, -C PATH       Set configuration file\n"
 513           "  --api, -A API           Select audio API.  Possibilities:\n"
 514           "                            ");
 515   int first = 1;
 516   for(int n = 0; uaudio_apis[n]; ++n) {
 517     if(uaudio_apis[n]->flags & UAUDIO_API_CLIENT) {
 518       if(first)
 519         first = 0;
 520       else
 521         xprintf(", ");
 522       xprintf("%s", uaudio_apis[n]->name);
 523     }
 524   }
 525   xprintf("\n"
 526           "  --command, -e COMMAND   Pipe audio to command.\n"
 527           "  --pause-mode, -P silence  For -e: pauses send silence (default)\n"
 528           "  --pause-mode, -P suspend  For -e: pauses suspend writes\n"
 529           "  --help, -h              Display usage message\n"
 530           "  --version, -V           Display version number\n"
 531           );
 532   xfclose(stdout);
 533   exit(0);
 534 }
 535
 536 static size_t playrtp_callback(void *buffer,
 537                                size_t max_samples,
 538                                void attribute((unused)) *userdata) {
 539   size_t samples;
 540   int silent = 0;
 541
 542   pthread_mutex_lock(&lock);
 543   /* Get the next packet, junking any that are now in the past */
 544   const struct packet *p = playrtp_next_packet();
 545   if(p && contains(p, next_timestamp)) {
 546     /* This packet is ready to play; the desired next timestamp points
 547      * somewhere into it. */
 548
 549     /* Timestamp of end of packet */
 550     const uint32_t packet_end = p->timestamp + p->nsamples;
 551
 552     /* Offset of desired next timestamp into current packet */
 553     const uint32_t offset = next_timestamp - p->timestamp;
 554
 555     /* Pointer to audio data */
 556     const uint16_t *ptr = (void *)(p->samples_raw + offset);
 557
 558     /* Compute number of samples left in packet, limited to output buffer
 559      * size */
 560     samples = packet_end - next_timestamp;
 561     if(samples > max_samples)
 562       samples = max_samples;
 563
 564     /* Copy into buffer, converting to native endianness */
 565     size_t i = samples;
 566     int16_t *bufptr = buffer;
 567     while(i > 0) {
 568       *bufptr++ = (int16_t)ntohs(*ptr++);
 569       --i;
 570     }
 571     silent = !!(p->flags & SILENT);
 572   } else {
 573     /* There is no suitable packet.  We introduce 0s up to the next packet, or
 574      * to fill the buffer if there's no next packet or that's too many.  The
 575      * comparison with max_samples deals with the otherwise troubling overflow
 576      * case. */
 577     samples = p ? p->timestamp - next_timestamp : max_samples;
 578     if(samples > max_samples)
 579       samples = max_samples;
 580     //info("infill by %zu", samples);
 581     memset(buffer, 0, samples * uaudio_sample_size);
 582     silent = 1;
 583   }
 584   /* Debug dump */
 585   if(dump_buffer) {
 586     for(size_t i = 0; i < samples; ++i) {
 587       dump_buffer[dump_index++] = ((int16_t *)buffer)[i];
 588       dump_index %= dump_size;
 589     }
 590   }
 591   /* Advance timestamp */
 592   next_timestamp += samples;
 593   /* If we're getting behind then try to drop just silent packets
 594    *
 595    * In theory this shouldn't be necessary.  The server is supposed to send
 596    * packets at the right rate and compares the number of samples sent with the
 597    * time in order to ensure this.
 598    *
 599    * However, various things could throw this off:
 600    *
 601    * - the server's clock could advance at the wrong rate.  This would cause it
 602    *   to mis-estimate the right number of samples to have sent and
 603    *   inappropriately throttle or speed up.
 604    *
 605    * - playback could happen at the wrong rate.  If the playback host's sound
 606    *   card has a slightly incorrect clock then eventually it will get out
 607    *   of step.
 608    *
 609    * So if we play back slightly slower than the server sends for either of
 610    * these reasons then eventually our buffer, and the socket's buffer, will
 611    * fill, and the kernel will start dropping packets.  The result is audible
 612    * and not very nice.
 613    *
 614    * Therefore if we're getting behind, we pre-emptively drop silent packets,
 615    * since a change in the duration of a silence is less noticeable than a
 616    * dropped packet from the middle of continuous music.
 617    *
 618    * (If things go wrong the other way then eventually we run out of packets to
 619    * play and are forced to play silence.  This doesn't seem to happen in
 620    * practice but if it does then in the same way we can artificially extend
 621    * silent packets to compensate.)
 622    *
 623    * Dropped packets are always logged; use 'disorder-playrtp --monitor' to
 624    * track how close to target buffer occupancy we are on a once-a-minute
 625    * basis.
 626    */
 627   if(nsamples > minbuffer && silent) {
 628     disorder_info("dropping %zu samples (%"PRIu32" > %"PRIu32")",
 629                   samples, nsamples, minbuffer);
 630     samples = 0;
 631   }
 632   /* Junk obsolete packets */
 633   playrtp_next_packet();
 634   pthread_mutex_unlock(&lock);
 635   return samples;
 636 }
 637
 638 int main(int argc, char **argv) {
 639   int n, err;
 640   struct addrinfo *res;
 641   struct stringlist sl;
 642   char *sockname;
 643   int rcvbuf, target_rcvbuf = -1;
 644   socklen_t len;
 645   struct ip_mreq mreq;
 646   struct ipv6_mreq mreq6;
 647   disorder_client *c = NULL;
 648   char *address, *port;
 649   int is_multicast;
 650   union any_sockaddr {
 651     struct sockaddr sa;
 652     struct sockaddr_in in;
 653     struct sockaddr_in6 in6;
 654   };
 655   union any_sockaddr mgroup;
 656   const char *dumpfile = 0;
 657   pthread_t ltid;
 658   int monitor = 0;
 659   static const int one = 1;
 660
 661   struct addrinfo prefs = {
 662     .ai_flags = AI_PASSIVE,
 663     .ai_family = PF_INET,
 664     .ai_socktype = SOCK_DGRAM,
 665     .ai_protocol = IPPROTO_UDP
 666   };
 667
 668   /* Timing information is often important to debugging playrtp, so we include
 669    * timestamps in the logs */
 670   logdate = 1;
 671   mem_init();
 672   if(!setlocale(LC_CTYPE, "")) disorder_fatal(errno, "error calling setlocale");
 673   while((n = getopt_long(argc, argv, "hVdD:m:x:L:R:aocC:re:P:MA:", options, 0)) >= 0) {
 674     switch(n) {
 675     case 'h': help();
 676     case 'V': version("disorder-playrtp");
 677     case 'd': debugging = 1; break;
 678     case 'D': uaudio_set("device", optarg); break;
 679     case 'm': minbuffer = 2 * atol(optarg); break;
 680     case 'x': maxbuffer = 2 * atol(optarg); break;
 681     case 'L': logfp = fopen(optarg, "w"); break;
 682     case 'R': target_rcvbuf = atoi(optarg); break;
 683 #if HAVE_ALSA_ASOUNDLIB_H
 684     case 'a':
 685       disorder_error(0, "deprecated option; use --api alsa instead");
 686       backend = &uaudio_alsa; break;
 687 #endif
 688 #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST
 689     case 'o':
 690       disorder_error(0, "deprecated option; use --api oss instead");
 691       backend = &uaudio_oss;
 692       break;
 693 #endif
 694 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
 695     case 'c':
 696       disorder_error(0, "deprecated option; use --api coreaudio instead");
 697       backend = &uaudio_coreaudio;
 698       break;
 699 #endif
 700     case 'A': backend = uaudio_find(optarg); break;
 701     case 'C': configfile = optarg; break;
 702     case 's': control_socket = optarg; break;
 703     case 'r': dumpfile = optarg; break;
 704     case 'e': backend = &uaudio_command; uaudio_set("command", optarg); break;
 705     case 'P': uaudio_set("pause-mode", optarg); break;
 706     case 'M': monitor = 1; break;
 707     default: disorder_fatal(0, "invalid option");
 708     }
 709   }
 710   if(config_read(0, NULL)) disorder_fatal(0, "cannot read configuration");
 711   /* Choose a sensible default audio backend */
 712   if(!backend) {
 713     backend = uaudio_default(uaudio_apis, UAUDIO_API_CLIENT);
 714     if(!backend)
 715       disorder_fatal(0, "no default uaudio API found");
 716     disorder_info("default audio API %s", backend->name);
 717   }
 718   if(backend == &uaudio_rtp) {
 719     /* This means that you have NO local sound output.  This can happen if you
 720      * use a non-Apple GCC on a Mac (because it doesn't know how to compile
 721      * CoreAudio/AudioHardware.h). */
 722     disorder_fatal(0, "cannot play RTP through RTP");
 723   }
 724   /* Set buffering parameters if not overridden */
 725   if(!minbuffer) {
 726     minbuffer = config->rtp_minbuffer;
 727     if(!minbuffer) minbuffer = (2*44100)*4/10;
 728   }
 729   if(!maxbuffer) {
 730     maxbuffer = config->rtp_maxbuffer;
 731     if(!maxbuffer) maxbuffer = 2 * minbuffer;
 732   }
 733   if(target_rcvbuf < 0) target_rcvbuf = config->rtp_rcvbuf;
 734   argc -= optind;
 735   argv += optind;
 736   switch(argc) {
 737   case 0:
 738     sl.s = xcalloc(3, sizeof *sl.s);
 739     if(config->rtp_always_request) {
 740       sl.s[0] = sl.s[1] = (/*unconst*/ char *)"-";
 741       sl.n = 2;
 742     } else {
 743       /* Get configuration from server */
 744       if(!(c = disorder_new(1))) exit(EXIT_FAILURE);
 745       if(disorder_connect(c)) exit(EXIT_FAILURE);
 746       if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE);
 747       sl.s[0] = address;
 748       sl.s[1] = port;
 749       sl.n = 2;
 750     }
 751     /* If we're requesting a new stream then apply the local network address
 752      * overrides.
 753      */
 754     if(!strcmp(sl.s[0], "-")) {
 755       if(config->rtp_request_address.port)
 756         byte_xasprintf(&sl.s[1], "%d", config->rtp_request_address.port);
 757       if(config->rtp_request_address.address) {
 758         sl.s[2] = sl.s[1];
 759         sl.s[1] = config->rtp_request_address.address;
 760         sl.n = 3;
 761       }
 762     }
 763     break;
 764   case 1: case 2: case 3:
 765     /* Use command-line ADDRESS+PORT or just PORT */
 766     sl.n = argc;
 767     sl.s = argv;
 768     break;
 769   default:
 770     disorder_fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]");
 771   }
 772   disorder_info("version "VERSION" process ID %lu",
 773                 (unsigned long)getpid());
 774   struct sockaddr *addr;
 775   socklen_t addr_len;
 776   if(!strcmp(sl.s[0], "-")) {
 777     /* Syntax: - [[ADDRESS] PORT].  Here, the PORT may be `-' to get the local
 778      * kernel to choose.  The ADDRESS may be omitted or `-' to pick something
 779      * suitable. */
 780     const char *node, *svc;
 781     struct sockaddr *sa = 0;
 782     switch (sl.n) {
 783 #define NULLDASH(s) (strcmp((s), "-") ? (s) : 0)
 784       case 1: node = 0; svc = 0; break;
 785       case 2: node = 0; svc = NULLDASH(sl.s[1]); break;
 786       case 3: node = NULLDASH(sl.s[1]); svc = NULLDASH(sl.s[2]); break;
 787       default: disorder_fatal(0, "too many listening-address compoennts");
 788 #undef NULLDASH
 789     }
 790     /* We'll need a connection to request the incoming stream, so open one if
 791      * we don't have one already */
 792     if(!c) {
 793       if(!(c = disorder_new(1))) exit(EXIT_FAILURE);
 794       if(disorder_connect(c)) exit(EXIT_FAILURE);
 795     }
 796     /* If no address was given, we need to pick one.  But we already have a
 797      * connection to the server, so we can probably use the address from that.
 798      */
 799     struct sockaddr_storage ss;
 800     if(!node) {
 801       addr_len = sizeof ss;
 802       if(disorder_client_sockname(c, (struct sockaddr *)&ss, &addr_len))
 803         exit(EXIT_FAILURE);
 804       if(ss.ss_family != AF_INET && ss.ss_family != AF_INET6) {
 805         /* We're using a Unix-domain socket, so use a loopback address.  I'm
 806          * cowardly using IPv4 here. */
 807         struct sockaddr_in *sin = (struct sockaddr_in *)&ss;
 808         sin->sin_family = AF_INET;
 809         sin->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
 810       }
 811       sa = (struct sockaddr *)&ss;
 812       prefs.ai_family = sa->sa_family;
 813     }
 814     /* If we have an address or port to resolve then do that now */
 815     if (node || svc) {
 816       struct addrinfo *ai;
 817       char errbuf[1024];
 818       int rc;
 819       if((rc = getaddrinfo(node, svc, &prefs, &ai)))
 820         disorder_fatal(0, "failed to resolve address `%s' and service `%s': %s",
 821                        node ? node : "-", svc ? svc : "-",
 822                        format_error(ec_getaddrinfo, rc,
 823                                     errbuf, sizeof(errbuf)));
 824       if(!sa)
 825         sa = ai->ai_addr;
 826       else {
 827         assert(sa->sa_family == ai->ai_addr->sa_family);
 828         switch(sa->sa_family) {
 829           case AF_INET:
 830             ((struct sockaddr_in *)sa)->sin_port =
 831               ((struct sockaddr_in *)ai->ai_addr)->sin_port;
 832             break;
 833           case AF_INET6:
 834             ((struct sockaddr_in6 *)sa)->sin6_port =
 835               ((struct sockaddr_in6 *)ai->ai_addr)->sin6_port;
 836             break;
 837           default:
 838             assert(!"unexpected address family");
 839         }
 840       }
 841     }
 842     if((rtpfd = socket(sa->sa_family, SOCK_DGRAM, IPPROTO_UDP)) < 0)
 843       disorder_fatal(errno, "error creating socket (family %d)",
 844                      sa->sa_family);
 845     /* Bind the address */
 846     if(bind(rtpfd, sa,
 847             sa->sa_family == AF_INET
 848             ? sizeof (struct sockaddr_in) : sizeof (struct sockaddr_in6)) < 0)
 849       disorder_fatal(errno, "error binding socket");
 850     static struct sockaddr_storage bound_address;
 851     addr = (struct sockaddr *)&bound_address;
 852     addr_len = sizeof bound_address;
 853     if(getsockname(rtpfd, addr, &addr_len) < 0)
 854       disorder_fatal(errno, "error getting socket address");
 855     /* Convert to string */
 856     char addrname[128], portname[32];
 857     if(getnameinfo(addr, addr_len,
 858                    addrname, sizeof addrname,
 859                    portname, sizeof portname,
 860                    NI_NUMERICHOST|NI_NUMERICSERV) < 0)
 861       disorder_fatal(errno, "getnameinfo");
 862     /* Ask for audio data */
 863     if(disorder_rtp_request(c, addrname, portname)) exit(EXIT_FAILURE);
 864     /* Report what we did */
 865     disorder_info("listening on %s (stream requested)",
 866                   format_sockaddr(addr));
 867   } else {
 868     if(sl.n > 2) disorder_fatal(0, "too many address components");
 869     /* Look up address and port */
 870     if(!(res = get_address(&sl, &prefs, &sockname)))
 871       exit(1);
 872     addr = res->ai_addr;
 873     addr_len = res->ai_addrlen;
 874     /* Create the socket */
 875     if((rtpfd = socket(res->ai_family,
 876                        res->ai_socktype,
 877                        res->ai_protocol)) < 0)
 878       disorder_fatal(errno, "error creating socket");
 879     /* Allow multiple listeners */
 880     xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
 881     is_multicast = multicast(addr);
 882     /* The multicast and unicast/broadcast cases are different enough that they
 883      * are totally split.  Trying to find commonality between them causes more
 884      * trouble that it's worth. */
 885     if(is_multicast) {
 886       /* Stash the multicast group address */
 887       memcpy(&mgroup, addr, addr_len);
 888       switch(res->ai_addr->sa_family) {
 889       case AF_INET:
 890         mgroup.in.sin_port = 0;
 891         break;
 892       case AF_INET6:
 893         mgroup.in6.sin6_port = 0;
 894         break;
 895       default:
 896         disorder_fatal(0, "unsupported address family %d",
 897                        (int)addr->sa_family);
 898       }
 899       /* Bind to to the multicast group address */
 900       if(bind(rtpfd, addr, addr_len) < 0)
 901         disorder_fatal(errno, "error binding socket to %s",
 902                        format_sockaddr(addr));
 903       /* Add multicast group membership */
 904       switch(mgroup.sa.sa_family) {
 905       case PF_INET:
 906         mreq.imr_multiaddr = mgroup.in.sin_addr;
 907         mreq.imr_interface.s_addr = 0;      /* use primary interface */
 908         if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 909                       &mreq, sizeof mreq) < 0)
 910           disorder_fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
 911         break;
 912       case PF_INET6:
 913         mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
 914         memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
 915         if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
 916                       &mreq6, sizeof mreq6) < 0)
 917           disorder_fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
 918         break;
 919       default:
 920         disorder_fatal(0, "unsupported address family %d", res->ai_family);
 921       }
 922       /* Report what we did */
 923       disorder_info("listening on %s multicast group %s",
 924                     format_sockaddr(addr), format_sockaddr(&mgroup.sa));
 925     } else {
 926       /* Bind to 0/port */
 927       switch(addr->sa_family) {
 928       case AF_INET: {
 929         struct sockaddr_in *in = (struct sockaddr_in *)addr;
 930
 931         memset(&in->sin_addr, 0, sizeof (struct in_addr));
 932         break;
 933       }
 934       case AF_INET6: {
 935         struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr;
 936
 937         memset(&in6->sin6_addr, 0, sizeof (struct in6_addr));
 938         break;
 939       }
 940       default:
 941         disorder_fatal(0, "unsupported family %d", (int)addr->sa_family);
 942       }
 943       if(bind(rtpfd, addr, addr_len) < 0)
 944         disorder_fatal(errno, "error binding socket to %s",
 945                        format_sockaddr(addr));
 946       /* Report what we did */
 947       disorder_info("listening on %s", format_sockaddr(addr));
 948     }
 949   }
 950   len = sizeof rcvbuf;
 951   if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
 952     disorder_fatal(errno, "error calling getsockopt SO_RCVBUF");
 953   if(target_rcvbuf > rcvbuf) {
 954     if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
 955                   &target_rcvbuf, sizeof target_rcvbuf) < 0)
 956       disorder_error(errno, "error calling setsockopt SO_RCVBUF %d",
 957                      target_rcvbuf);
 958       /* We try to carry on anyway */
 959     else
 960       disorder_info("changed socket receive buffer from %d to %d",
 961                     rcvbuf, target_rcvbuf);
 962   } else
 963     disorder_info("default socket receive buffer %d", rcvbuf);
 964   //info("minbuffer %u maxbuffer %u", minbuffer, maxbuffer);
 965   if(logfp)
 966     disorder_info("WARNING: -L option can impact performance");
 967   if(control_socket) {
 968     pthread_t tid;
 969
 970     if((err = pthread_create(&tid, 0, control_thread, 0)))
 971       disorder_fatal(err, "pthread_create control_thread");
 972   }
 973   if(dumpfile) {
 974     int fd;
 975     unsigned char buffer[65536];
 976     size_t written;
 977
 978     if((fd = open(dumpfile, O_RDWR|O_TRUNC|O_CREAT, 0666)) < 0)
 979       disorder_fatal(errno, "opening %s", dumpfile);
 980     /* Fill with 0s to a suitable size */
 981     memset(buffer, 0, sizeof buffer);
 982     for(written = 0; written < dump_size * sizeof(int16_t);
 983         written += sizeof buffer) {
 984       if(write(fd, buffer, sizeof buffer) < 0)
 985         disorder_fatal(errno, "clearing %s", dumpfile);
 986     }
 987     /* Map the buffer into memory for convenience */
 988     dump_buffer = mmap(0, dump_size * sizeof(int16_t), PROT_READ|PROT_WRITE,
 989                        MAP_SHARED, fd, 0);
 990     if(dump_buffer == (void *)-1)
 991       disorder_fatal(errno, "mapping %s", dumpfile);
 992     disorder_info("dumping to %s", dumpfile);
 993   }
 994   /* Set up output.  Currently we only support L16 so there's no harm setting
 995    * the format before we know what it is! */
 996   uaudio_set_format(44100/*Hz*/, 2/*channels*/,
 997                     16/*bits/channel*/, 1/*signed*/);
 998   uaudio_set("application", "disorder-playrtp");
 999   backend->configure();
1000   backend->start(playrtp_callback, NULL);
1001   if(backend->open_mixer) backend->open_mixer();
1002   /* We receive and convert audio data in a background thread */
1003   if((err = pthread_create(&ltid, 0, listen_thread, 0)))
1004     disorder_fatal(err, "pthread_create listen_thread");
1005   /* We have a second thread to add received packets to the queue */
1006   if((err = pthread_create(&ltid, 0, queue_thread, 0)))
1007     disorder_fatal(err, "pthread_create queue_thread");
1008   pthread_mutex_lock(&lock);
1009   time_t lastlog = 0;
1010   for(;;) {
1011     /* Wait for the buffer to fill up a bit */
1012     playrtp_fill_buffer();
1013     /* Start playing now */
1014     disorder_info("Playing...");
1015     next_timestamp = pheap_first(&packets)->timestamp;
1016     active = 1;
1017     pthread_mutex_unlock(&lock);
1018     backend->activate();
1019     pthread_mutex_lock(&lock);
1020     /* Wait until the buffer empties out
1021      *
1022      * If there's a packet that we can play right now then we definitely
1023      * continue.
1024      *
1025      * Also if there's at least minbuffer samples we carry on regardless and
1026      * insert silence.  The assumption is there's been a pause but more data
1027      * is now available.
1028      */
1029     while(nsamples >= minbuffer
1030           || (nsamples > 0
1031               && contains(pheap_first(&packets), next_timestamp))) {
1032       if(monitor) {
1033         time_t now = xtime(0);
1034
1035         if(now >= lastlog + 60) {
1036           int offset = nsamples - minbuffer;
1037           double offtime = (double)offset / (uaudio_rate * uaudio_channels);
1038           disorder_info("%+d samples off (%d.%02ds, %d bytes)",
1039                         offset,
1040                         (int)fabs(offtime) * (offtime < 0 ? -1 : 1),
1041                         (int)(fabs(offtime) * 100) % 100,
1042                         offset * uaudio_bits / CHAR_BIT);
1043           lastlog = now;
1044         }
1045       }
1046       //fprintf(stderr, "%8u/%u (%u) PLAYING\n", nsamples, maxbuffer, minbuffer);
1047       pthread_cond_wait(&cond, &lock);
1048     }
1049 #if 0
1050     if(nsamples) {
1051       struct packet *p = pheap_first(&packets);
1052       fprintf(stderr, "nsamples=%u (%u) next_timestamp=%"PRIx32", first packet is [%"PRIx32",%"PRIx32")\n",
1053               nsamples, minbuffer, next_timestamp,p->timestamp,p->timestamp+p->nsamples);
1054     }
1055 #endif
1056     /* Stop playing for a bit until the buffer re-fills */
1057     pthread_mutex_unlock(&lock);
1058     backend->deactivate();
1059     pthread_mutex_lock(&lock);
1060     active = 0;
1061     /* Go back round */
1062   }
1063   return 0;
1064 }
1065
1066 /*
1067 Local Variables:
1068 c-basic-offset:2
1069 comment-column:40
1070 fill-column:79
1071 indent-tabs-mode:nil
1072 End:
1073 */