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