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

   1 /*
   2  * This file is part of DisOrder.
   3  * Copyright (C) 2007 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 2 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, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * 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, write to the Free Software
  17  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  18  * USA
  19  */
  20 /** @file clients/playrtp.c
  21  * @brief RTP player
  22  *
  23  * This player supports Linux (<a href="http://www.alsa-project.org/">ALSA</a>)
  24  * and Apple Mac (<a
  25  * href="http://developer.apple.com/audio/coreaudio.html">Core Audio</a>)
  26  * systems.  There is no support for Microsoft Windows yet, and that will in
  27  * fact probably an entirely separate program.
  28  *
  29  * The program runs (at least) three threads.  listen_thread() is responsible
  30  * for reading RTP packets off the wire and adding them to the linked list @ref
  31  * received_packets, assuming they are basically sound.  queue_thread() takes
  32  * packets off this linked list and adds them to @ref packets (an operation
  33  * which might be much slower due to contention for @ref lock).
  34  *
  35  * The main thread is responsible for actually playing audio.  In ALSA this
  36  * means it waits until ALSA says it's ready for more audio which it then
  37  * plays.
  38  *
  39  * InCore Audio the main thread is only responsible for starting and stopping
  40  * play: the system does the actual playback in its own private thread, and
  41  * calls adioproc() to fetch the audio data.
  42  *
  43  * Sometimes it happens that there is no audio available to play.  This may
  44  * because the server went away, or a packet was dropped, or the server
  45  * deliberately did not send any sound because it encountered a silence.
  46  *
  47  * Assumptions:
  48  * - it is safe to read uint32_t values without a lock protecting them
  49  */
  50
  51 #include <config.h>
  52 #include "types.h"
  53
  54 #include <getopt.h>
  55 #include <stdio.h>
  56 #include <stdlib.h>
  57 #include <sys/socket.h>
  58 #include <sys/types.h>
  59 #include <sys/socket.h>
  60 #include <netdb.h>
  61 #include <pthread.h>
  62 #include <locale.h>
  63 #include <sys/uio.h>
  64 #include <string.h>
  65
  66 #include "log.h"
  67 #include "mem.h"
  68 #include "configuration.h"
  69 #include "addr.h"
  70 #include "syscalls.h"
  71 #include "rtp.h"
  72 #include "defs.h"
  73 #include "vector.h"
  74 #include "heap.h"
  75 #include "timeval.h"
  76
  77 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
  78 # include <CoreAudio/AudioHardware.h>
  79 #endif
  80 #if API_ALSA
  81 #include <alsa/asoundlib.h>
  82 #endif
  83
  84 #define readahead linux_headers_are_borked
  85
  86 /** @brief RTP socket */
  87 static int rtpfd;
  88
  89 /** @brief Log output */
  90 static FILE *logfp;
  91
  92 /** @brief Output device */
  93 static const char *device;
  94
  95 /** @brief Maximum samples per packet we'll support
  96  *
  97  * NB that two channels = two samples in this program.
  98  */
  99 #define MAXSAMPLES 2048
 100
 101 /** @brief Minimum low watermark
 102  *
 103  * We'll stop playing if there's only this many samples in the buffer. */
 104 static unsigned minbuffer = 2 * 44100 / 10;  /* 0.2 seconds */
 105
 106 /** @brief Buffer high watermark
 107  *
 108  * We'll only start playing when this many samples are available. */
 109 static unsigned readahead = 2 * 2 * 44100;
 110
 111 /** @brief Maximum buffer size
 112  *
 113  * We'll stop reading from the network if we have this many samples. */
 114 static unsigned maxbuffer;
 115
 116 /** @brief Number of samples to infill by in one go
 117  *
 118  * This is an upper bound - in practice we expect the underlying audio API to
 119  * only ask for a much smaller number of samples in any one go.
 120  */
 121 #define INFILL_SAMPLES (44100 * 2)      /* 1s */
 122
 123 /** @brief Received packet
 124  *
 125  * Received packets are kept in a binary heap (see @ref pheap) ordered by
 126  * timestamp.
 127  */
 128 struct packet {
 129   /** @brief Next packet in @ref next_free_packet or @ref received_packets */
 130   struct packet *next;
 131
 132   /** @brief Number of samples in this packet */
 133   uint32_t nsamples;
 134
 135   /** @brief Timestamp from RTP packet
 136    *
 137    * NB that "timestamps" are really sample counters.  Use lt() or lt_packet()
 138    * to compare timestamps.
 139    */
 140   uint32_t timestamp;
 141
 142   /** @brief Flags
 143    *
 144    * Valid values are:
 145    * - @ref IDLE - the idle bit was set in the RTP packet
 146    */
 147   unsigned flags;
 148 /** @brief idle bit set in RTP packet*/
 149 #define IDLE 0x0001
 150
 151   /** @brief Raw sample data
 152    *
 153    * Only the first @p nsamples samples are defined; the rest is uninitialized
 154    * data.
 155    */
 156   uint16_t samples_raw[MAXSAMPLES];
 157 };
 158
 159 /** @brief Return true iff \f$a < b\f$ in sequence-space arithmetic
 160  *
 161  * Specifically it returns true if \f$(a-b) mod 2^{32} < 2^{31}\f$.
 162  *
 163  * See also lt_packet().
 164  */
 165 static inline int lt(uint32_t a, uint32_t b) {
 166   return (uint32_t)(a - b) & 0x80000000;
 167 }
 168
 169 /** @brief Return true iff a >= b in sequence-space arithmetic */
 170 static inline int ge(uint32_t a, uint32_t b) {
 171   return !lt(a, b);
 172 }
 173
 174 /** @brief Return true iff a > b in sequence-space arithmetic */
 175 static inline int gt(uint32_t a, uint32_t b) {
 176   return lt(b, a);
 177 }
 178
 179 /** @brief Return true iff a <= b in sequence-space arithmetic */
 180 static inline int le(uint32_t a, uint32_t b) {
 181   return !lt(b, a);
 182 }
 183
 184 /** @brief Ordering for packets, used by @ref pheap */
 185 static inline int lt_packet(const struct packet *a, const struct packet *b) {
 186   return lt(a->timestamp, b->timestamp);
 187 }
 188
 189 /** @brief Received packets
 190  * Protected by @ref receive_lock
 191  *
 192  * Received packets are added to this list, and queue_thread() picks them off
 193  * it and adds them to @ref packets.  Whenever a packet is added to it, @ref
 194  * receive_cond is signalled.
 195  */
 196 static struct packet *received_packets;
 197
 198 /** @brief Tail of @ref received_packets
 199  * Protected by @ref receive_lock
 200  */
 201 static struct packet **received_tail = &received_packets;
 202
 203 /** @brief Lock protecting @ref received_packets
 204  *
 205  * Only listen_thread() and queue_thread() ever hold this lock.  It is vital
 206  * that queue_thread() not hold it any longer than it strictly has to. */
 207 static pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
 208
 209 /** @brief Condition variable signalled when @ref received_packets is updated
 210  *
 211  * Used by listen_thread() to notify queue_thread() that it has added another
 212  * packet to @ref received_packets. */
 213 static pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
 214
 215 /** @brief Length of @ref received_packets */
 216 static uint32_t nreceived;
 217
 218 /** @struct pheap
 219  * @brief Binary heap of packets ordered by timestamp */
 220 HEAP_TYPE(pheap, struct packet *, lt_packet);
 221
 222 /** @brief Binary heap of received packets */
 223 static struct pheap packets;
 224
 225 /** @brief Total number of samples available
 226  *
 227  * We make this volatile because we inspect it without a protecting lock,
 228  * so the usual pthread_* guarantees aren't available.
 229  */
 230 static volatile uint32_t nsamples;
 231
 232 /** @brief Timestamp of next packet to play.
 233  *
 234  * This is set to the timestamp of the last packet, plus the number of
 235  * samples it contained.  Only valid if @ref active is nonzero.
 236  */
 237 static uint32_t next_timestamp;
 238
 239 /** @brief True if actively playing
 240  *
 241  * This is true when playing and false when just buffering. */
 242 static int active;
 243
 244 /** @brief Lock protecting @ref packets */
 245 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
 246
 247 /** @brief Condition variable signalled whenever @ref packets is changed */
 248 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 249
 250 /** @brief Structure of free packet list */
 251 union free_packet {
 252   struct packet p;
 253   union free_packet *next;
 254 };
 255
 256 /** @brief Linked list of free packets
 257  *
 258  * This is a linked list of formerly used packets.  For preference we re-use
 259  * packets that have already been used rather than unused ones, to limit the
 260  * size of the program's working set.  If there are no free packets in the list
 261  * we try @ref next_free_packet instead.
 262  *
 263  * Must hold @ref lock when accessing this.
 264  */
 265 static union free_packet *free_packets;
 266
 267 /** @brief Array of new free packets
 268  *
 269  * There are @ref count_free_packets ready to use at this address.  If there
 270  * are none left we allocate more memory.
 271  *
 272  * Must hold @ref lock when accessing this.
 273  */
 274 static union free_packet *next_free_packet;
 275
 276 /** @brief Count of new free packets at @ref next_free_packet
 277  *
 278  * Must hold @ref lock when accessing this.
 279  */
 280 static size_t count_free_packets;
 281
 282 /** @brief Lock protecting packet allocator */
 283 static pthread_mutex_t mem_lock = PTHREAD_MUTEX_INITIALIZER;
 284
 285 static const struct option options[] = {
 286   { "help", no_argument, 0, 'h' },
 287   { "version", no_argument, 0, 'V' },
 288   { "debug", no_argument, 0, 'd' },
 289   { "device", required_argument, 0, 'D' },
 290   { "min", required_argument, 0, 'm' },
 291   { "max", required_argument, 0, 'x' },
 292   { "buffer", required_argument, 0, 'b' },
 293   { "rcvbuf", required_argument, 0, 'R' },
 294   { "multicast", required_argument, 0, 'M' },
 295   { 0, 0, 0, 0 }
 296 };
 297
 298 /** @brief Return a new packet */
 299 static struct packet *new_packet(void) {
 300   struct packet *p;
 301
 302   pthread_mutex_lock(&mem_lock);
 303   if(free_packets) {
 304     p = &free_packets->p;
 305     free_packets = free_packets->next;
 306   } else {
 307     if(!count_free_packets) {
 308       next_free_packet = xcalloc(1024, sizeof (union free_packet));
 309       count_free_packets = 1024;
 310     }
 311     p = &(next_free_packet++)->p;
 312     --count_free_packets;
 313   }
 314   pthread_mutex_unlock(&mem_lock);
 315   return p;
 316 }
 317
 318 /** @brief Free a packet */
 319 static void free_packet(struct packet *p) {
 320   union free_packet *u = (union free_packet *)p;
 321   pthread_mutex_lock(&mem_lock);
 322   u->next = free_packets;
 323   free_packets = u;
 324   pthread_mutex_unlock(&mem_lock);
 325 }
 326
 327 /** @brief Drop the first packet
 328  *
 329  * Assumes that @ref lock is held.
 330  */
 331 static void drop_first_packet(void) {
 332   if(pheap_count(&packets)) {
 333     struct packet *const p = pheap_remove(&packets);
 334     nsamples -= p->nsamples;
 335     free_packet(p);
 336     pthread_cond_broadcast(&cond);
 337   }
 338 }
 339
 340 /** @brief Background thread adding packets to heap
 341  *
 342  * This just transfers packets from @ref received_packets to @ref packets.  It
 343  * is important that it holds @ref receive_lock for as little time as possible,
 344  * in order to minimize the interval between calls to read() in
 345  * listen_thread().
 346  */
 347 static void *queue_thread(void attribute((unused)) *arg) {
 348   struct packet *p;
 349
 350   for(;;) {
 351     /* Get the next packet */
 352     pthread_mutex_lock(&receive_lock);
 353     while(!received_packets)
 354       pthread_cond_wait(&receive_cond, &receive_lock);
 355     p = received_packets;
 356     received_packets = p->next;
 357     if(!received_packets)
 358       received_tail = &received_packets;
 359     --nreceived;
 360     pthread_mutex_unlock(&receive_lock);
 361     /* Add it to the heap */
 362     pthread_mutex_lock(&lock);
 363     pheap_insert(&packets, p);
 364     nsamples += p->nsamples;
 365     pthread_cond_broadcast(&cond);
 366     pthread_mutex_unlock(&lock);
 367   }
 368 }
 369
 370 /** @brief Background thread collecting samples
 371  *
 372  * This function collects samples, perhaps converts them to the target format,
 373  * and adds them to the packet list.
 374  *
 375  * It is crucial that the gap between successive calls to read() is as small as
 376  * possible: otherwise packets will be dropped.
 377  *
 378  * We use a binary heap to ensure that the unavoidable effort is at worst
 379  * logarithmic in the total number of packets - in fact if packets are mostly
 380  * received in order then we will largely do constant work per packet since the
 381  * newest packet will always be last.
 382  *
 383  * Of more concern is that we must acquire the lock on the heap to add a packet
 384  * to it.  If this proves a problem in practice then the answer would be
 385  * (probably doubly) linked list with new packets added the end and a second
 386  * thread which reads packets off the list and adds them to the heap.
 387  *
 388  * We keep memory allocation (mostly) very fast by keeping pre-allocated
 389  * packets around; see @ref new_packet().
 390  */
 391 static void *listen_thread(void attribute((unused)) *arg) {
 392   struct packet *p = 0;
 393   int n;
 394   struct rtp_header header;
 395   uint16_t seq;
 396   uint32_t timestamp;
 397   struct iovec iov[2];
 398
 399   for(;;) {
 400     if(!p)
 401       p = new_packet();
 402     iov[0].iov_base = &header;
 403     iov[0].iov_len = sizeof header;
 404     iov[1].iov_base = p->samples_raw;
 405     iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
 406     n = readv(rtpfd, iov, 2);
 407     if(n < 0) {
 408       switch(errno) {
 409       case EINTR:
 410         continue;
 411       default:
 412         fatal(errno, "error reading from socket");
 413       }
 414     }
 415     /* Ignore too-short packets */
 416     if((size_t)n <= sizeof (struct rtp_header)) {
 417       info("ignored a short packet");
 418       continue;
 419     }
 420     timestamp = htonl(header.timestamp);
 421     seq = htons(header.seq);
 422     /* Ignore packets in the past */
 423     if(active && lt(timestamp, next_timestamp)) {
 424       info("dropping old packet, timestamp=%"PRIx32" < %"PRIx32,
 425            timestamp, next_timestamp);
 426       continue;
 427     }
 428     p->next = 0;
 429     p->flags = 0;
 430     p->timestamp = timestamp;
 431     /* Convert to target format */
 432     if(header.mpt & 0x80)
 433       p->flags |= IDLE;
 434     switch(header.mpt & 0x7F) {
 435     case 10:
 436       p->nsamples = (n - sizeof header) / sizeof(uint16_t);
 437       break;
 438       /* TODO support other RFC3551 media types (when the speaker does) */
 439     default:
 440       fatal(0, "unsupported RTP payload type %d",
 441             header.mpt & 0x7F);
 442     }
 443     if(logfp)
 444       fprintf(logfp, "sequence %u timestamp %"PRIx32" length %"PRIx32" end %"PRIx32"\n",
 445               seq, timestamp, p->nsamples, timestamp + p->nsamples);
 446     /* Stop reading if we've reached the maximum.
 447      *
 448      * This is rather unsatisfactory: it means that if packets get heavily
 449      * out of order then we guarantee dropouts.  But for now... */
 450     if(nsamples >= maxbuffer) {
 451       pthread_mutex_lock(&lock);
 452       while(nsamples >= maxbuffer)
 453         pthread_cond_wait(&cond, &lock);
 454       pthread_mutex_unlock(&lock);
 455     }
 456     /* Add the packet to the receive queue */
 457     pthread_mutex_lock(&receive_lock);
 458     *received_tail = p;
 459     received_tail = &p->next;
 460     ++nreceived;
 461     pthread_cond_signal(&receive_cond);
 462     pthread_mutex_unlock(&receive_lock);
 463     /* We'll need a new packet */
 464     p = 0;
 465   }
 466 }
 467
 468 /** @brief Return true if @p p contains @p timestamp
 469  *
 470  * Containment implies that a sample @p timestamp exists within the packet.
 471  */
 472 static inline int contains(const struct packet *p, uint32_t timestamp) {
 473   const uint32_t packet_start = p->timestamp;
 474   const uint32_t packet_end = p->timestamp + p->nsamples;
 475
 476   return (ge(timestamp, packet_start)
 477           && lt(timestamp, packet_end));
 478 }
 479
 480 /** @brief Wait until the buffer is adequately full
 481  *
 482  * Must be called with @ref lock held.
 483  */
 484 static void fill_buffer(void) {
 485   while(nsamples)
 486     drop_first_packet();
 487   info("Buffering...");
 488   while(nsamples < readahead)
 489     pthread_cond_wait(&cond, &lock);
 490   next_timestamp = pheap_first(&packets)->timestamp;
 491   active = 1;
 492 }
 493
 494 /** @brief Find next packet
 495  * @return Packet to play or NULL if none found
 496  *
 497  * The return packet is merely guaranteed not to be in the past: it might be
 498  * the first packet in the future rather than one that is actually suitable to
 499  * play.
 500  *
 501  * Must be called with @ref lock held.
 502  */
 503 static struct packet *next_packet(void) {
 504   while(pheap_count(&packets)) {
 505     struct packet *const p = pheap_first(&packets);
 506     if(le(p->timestamp + p->nsamples, next_timestamp)) {
 507       /* This packet is in the past.  Drop it and try another one. */
 508       drop_first_packet();
 509     } else
 510       /* This packet is NOT in the past.  (It might be in the future
 511        * however.) */
 512       return p;
 513   }
 514   return 0;
 515 }
 516
 517 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
 518 /** @brief Callback from Core Audio */
 519 static OSStatus adioproc
 520     (AudioDeviceID attribute((unused)) inDevice,
 521      const AudioTimeStamp attribute((unused)) *inNow,
 522      const AudioBufferList attribute((unused)) *inInputData,
 523      const AudioTimeStamp attribute((unused)) *inInputTime,
 524      AudioBufferList *outOutputData,
 525      const AudioTimeStamp attribute((unused)) *inOutputTime,
 526      void attribute((unused)) *inClientData) {
 527   UInt32 nbuffers = outOutputData->mNumberBuffers;
 528   AudioBuffer *ab = outOutputData->mBuffers;
 529   uint32_t samples_available;
 530
 531   pthread_mutex_lock(&lock);
 532   while(nbuffers > 0) {
 533     float *samplesOut = ab->mData;
 534     size_t samplesOutLeft = ab->mDataByteSize / sizeof (float);
 535
 536     while(samplesOutLeft > 0) {
 537       const struct packet *p = next_packet();
 538       if(p && contains(p, next_timestamp)) {
 539         /* This packet is ready to play */
 540         const uint32_t packet_end = p->timestamp + p->nsamples;
 541         const uint32_t offset = next_timestamp - p->timestamp;
 542         const uint16_t *ptr = (void *)(p->samples_raw + offset);
 543
 544         samples_available = packet_end - next_timestamp;
 545         if(samples_available > samplesOutLeft)
 546           samples_available = samplesOutLeft;
 547         next_timestamp += samples_available;
 548         samplesOutLeft -= samples_available;
 549         while(samples_available-- > 0)
 550           *samplesOut++ = (int16_t)ntohs(*ptr++) * (0.5 / 32767);
 551         /* We don't bother junking the packet - that'll be dealt with next time
 552          * round */
 553       } else {
 554         /* No packet is ready to play (and there might be no packet at all) */
 555         samples_available = p ? p->timestamp - next_timestamp
 556                               : samplesOutLeft;
 557         if(samples_available > samplesOutLeft)
 558           samples_available = samplesOutLeft;
 559         //info("infill by %"PRIu32, samples_available);
 560         /* Conveniently the buffer is 0 to start with */
 561         next_timestamp += samples_available;
 562         samplesOut += samples_available;
 563         samplesOutLeft -= samples_available;
 564       }
 565     }
 566     ++ab;
 567     --nbuffers;
 568   }
 569   pthread_mutex_unlock(&lock);
 570   return 0;
 571 }
 572 #endif
 573
 574
 575 #if API_ALSA
 576 /** @brief PCM handle */
 577 static snd_pcm_t *pcm;
 578
 579 /** @brief True when @ref pcm is up and running */
 580 static int alsa_prepared = 1;
 581
 582 /** @brief Initialize @ref pcm */
 583 static void setup_alsa(void) {
 584   snd_pcm_hw_params_t *hwparams;
 585   snd_pcm_sw_params_t *swparams;
 586   /* Only support one format for now */
 587   const int sample_format = SND_PCM_FORMAT_S16_BE;
 588   unsigned rate = 44100;
 589   const int channels = 2;
 590   const int samplesize = channels * sizeof(uint16_t);
 591   snd_pcm_uframes_t pcm_bufsize = MAXSAMPLES * samplesize * 3;
 592   /* If we can write more than this many samples we'll get a wakeup */
 593   const int avail_min = 256;
 594   int err;
 595
 596   /* Open ALSA */
 597   if((err = snd_pcm_open(&pcm,
 598                          device ? device : "default",
 599                          SND_PCM_STREAM_PLAYBACK,
 600                          SND_PCM_NONBLOCK)))
 601     fatal(0, "error from snd_pcm_open: %d", err);
 602   /* Set up 'hardware' parameters */
 603   snd_pcm_hw_params_alloca(&hwparams);
 604   if((err = snd_pcm_hw_params_any(pcm, hwparams)) < 0)
 605     fatal(0, "error from snd_pcm_hw_params_any: %d", err);
 606   if((err = snd_pcm_hw_params_set_access(pcm, hwparams,
 607                                          SND_PCM_ACCESS_RW_INTERLEAVED)) < 0)
 608     fatal(0, "error from snd_pcm_hw_params_set_access: %d", err);
 609   if((err = snd_pcm_hw_params_set_format(pcm, hwparams,
 610                                          sample_format)) < 0)
 611
 612     fatal(0, "error from snd_pcm_hw_params_set_format (%d): %d",
 613           sample_format, err);
 614   if((err = snd_pcm_hw_params_set_rate_near(pcm, hwparams, &rate, 0)) < 0)
 615     fatal(0, "error from snd_pcm_hw_params_set_rate (%d): %d",
 616           rate, err);
 617   if((err = snd_pcm_hw_params_set_channels(pcm, hwparams,
 618                                            channels)) < 0)
 619     fatal(0, "error from snd_pcm_hw_params_set_channels (%d): %d",
 620           channels, err);
 621   if((err = snd_pcm_hw_params_set_buffer_size_near(pcm, hwparams,
 622                                                    &pcm_bufsize)) < 0)
 623     fatal(0, "error from snd_pcm_hw_params_set_buffer_size (%d): %d",
 624           MAXSAMPLES * samplesize * 3, err);
 625   if((err = snd_pcm_hw_params(pcm, hwparams)) < 0)
 626     fatal(0, "error calling snd_pcm_hw_params: %d", err);
 627   /* Set up 'software' parameters */
 628   snd_pcm_sw_params_alloca(&swparams);
 629   if((err = snd_pcm_sw_params_current(pcm, swparams)) < 0)
 630     fatal(0, "error calling snd_pcm_sw_params_current: %d", err);
 631   if((err = snd_pcm_sw_params_set_avail_min(pcm, swparams, avail_min)) < 0)
 632     fatal(0, "error calling snd_pcm_sw_params_set_avail_min %d: %d",
 633           avail_min, err);
 634   if((err = snd_pcm_sw_params(pcm, swparams)) < 0)
 635     fatal(0, "error calling snd_pcm_sw_params: %d", err);
 636 }
 637
 638 /** @brief Wait until ALSA wants some audio */
 639 static void wait_alsa(void) {
 640   struct pollfd fds[64];
 641   int nfds, err;
 642   unsigned short events;
 643
 644   for(;;) {
 645     do {
 646       if((nfds = snd_pcm_poll_descriptors(pcm,
 647                                           fds, sizeof fds / sizeof *fds)) < 0)
 648         fatal(0, "error calling snd_pcm_poll_descriptors: %d", nfds);
 649     } while(poll(fds, nfds, -1) < 0 && errno == EINTR);
 650     if((err = snd_pcm_poll_descriptors_revents(pcm, fds, nfds, &events)))
 651       fatal(0, "error calling snd_pcm_poll_descriptors_revents: %d", err);
 652     if(events & POLLOUT)
 653       return;
 654   }
 655 }
 656
 657 /** @brief Play some sound via ALSA
 658  * @param s Pointer to sample data
 659  * @param n Number of samples
 660  * @return 0 on success, -1 on non-fatal error
 661  */
 662 static int alsa_writei(const void *s, size_t n) {
 663   /* Do the write */
 664   const snd_pcm_sframes_t frames_written = snd_pcm_writei(pcm, s, n / 2);
 665   if(frames_written < 0) {
 666     /* Something went wrong */
 667     switch(frames_written) {
 668     case -EAGAIN:
 669       return 0;
 670     case -EPIPE:
 671       error(0, "error calling snd_pcm_writei: %ld",
 672             (long)frames_written);
 673       return -1;
 674     default:
 675       fatal(0, "error calling snd_pcm_writei: %ld",
 676             (long)frames_written);
 677     }
 678   } else {
 679     /* Success */
 680     next_timestamp += frames_written * 2;
 681     return 0;
 682   }
 683 }
 684
 685 /** @brief Play the relevant part of a packet
 686  * @param p Packet to play
 687  * @return 0 on success, -1 on non-fatal error
 688  */
 689 static int alsa_play(const struct packet *p) {
 690   return alsa_writei(p->samples_raw + next_timestamp - p->timestamp,
 691                      (p->timestamp + p->nsamples) - next_timestamp);
 692 }
 693
 694 /** @brief Play some silence
 695  * @param p Next packet or NULL
 696  * @return 0 on success, -1 on non-fatal error
 697  */
 698 static int alsa_infill(const struct packet *p) {
 699   static const uint16_t zeros[INFILL_SAMPLES];
 700   size_t samples_available = INFILL_SAMPLES;
 701
 702   if(p && samples_available > p->timestamp - next_timestamp)
 703     samples_available = p->timestamp - next_timestamp;
 704   return alsa_writei(zeros, samples_available);
 705 }
 706
 707 /** @brief Reset ALSA state after we lost synchronization */
 708 static void alsa_reset(int hard_reset) {
 709   int err;
 710
 711   if((err = snd_pcm_nonblock(pcm, 0)))
 712     fatal(0, "error calling snd_pcm_nonblock: %d", err);
 713   if(hard_reset) {
 714     if((err = snd_pcm_drop(pcm)))
 715       fatal(0, "error calling snd_pcm_drop: %d", err);
 716   } else
 717     if((err = snd_pcm_drain(pcm)))
 718       fatal(0, "error calling snd_pcm_drain: %d", err);
 719   if((err = snd_pcm_nonblock(pcm, 1)))
 720     fatal(0, "error calling snd_pcm_nonblock: %d", err);
 721   alsa_prepared = 0;
 722 }
 723 #endif
 724
 725 /** @brief Play an RTP stream
 726  *
 727  * This is the guts of the program.  It is responsible for:
 728  * - starting the listening thread
 729  * - opening the audio device
 730  * - reading ahead to build up a buffer
 731  * - arranging for audio to be played
 732  * - detecting when the buffer has got too small and re-buffering
 733  */
 734 static void play_rtp(void) {
 735   pthread_t ltid;
 736
 737   /* We receive and convert audio data in a background thread */
 738   pthread_create(&ltid, 0, listen_thread, 0);
 739   /* We have a second thread to add received packets to the queue */
 740   pthread_create(&ltid, 0, queue_thread, 0);
 741 #if API_ALSA
 742   {
 743     struct packet *p;
 744     int escape, err;
 745
 746     /* Open the sound device */
 747     setup_alsa();
 748     pthread_mutex_lock(&lock);
 749     for(;;) {
 750       /* Wait for the buffer to fill up a bit */
 751       fill_buffer();
 752       if(!alsa_prepared) {
 753         if((err = snd_pcm_prepare(pcm)))
 754           fatal(0, "error calling snd_pcm_prepare: %d", err);
 755         alsa_prepared = 1;
 756       }
 757       escape = 0;
 758       info("Playing...");
 759       /* Keep playing until the buffer empties out, or ALSA tells us to get
 760        * lost */
 761       while((nsamples >= minbuffer
 762              || (nsamples > 0
 763                  && contains(pheap_first(&packets), next_timestamp)))
 764             && !escape) {
 765         /* Wait for ALSA to ask us for more data */
 766         pthread_mutex_unlock(&lock);
 767         wait_alsa();
 768         pthread_mutex_lock(&lock);
 769         /* ALSA is ready for more data, find something to play */
 770         p = next_packet();
 771         /* Play it or play some silence */
 772         if(contains(p, next_timestamp))
 773           escape = alsa_play(p);
 774         else
 775           escape = alsa_infill(p);
 776       }
 777       active = 0;
 778       /* We stop playing for a bit until the buffer re-fills */
 779       pthread_mutex_unlock(&lock);
 780       alsa_reset(escape);
 781       pthread_mutex_lock(&lock);
 782     }
 783
 784   }
 785 #elif HAVE_COREAUDIO_AUDIOHARDWARE_H
 786   {
 787     OSStatus status;
 788     UInt32 propertySize;
 789     AudioDeviceID adid;
 790     AudioStreamBasicDescription asbd;
 791
 792     /* If this looks suspiciously like libao's macosx driver there's an
 793      * excellent reason for that... */
 794
 795     /* TODO report errors as strings not numbers */
 796     propertySize = sizeof adid;
 797     status = AudioHardwareGetProperty(kAudioHardwarePropertyDefaultOutputDevice,
 798                                       &propertySize, &adid);
 799     if(status)
 800       fatal(0, "AudioHardwareGetProperty: %d", (int)status);
 801     if(adid == kAudioDeviceUnknown)
 802       fatal(0, "no output device");
 803     propertySize = sizeof asbd;
 804     status = AudioDeviceGetProperty(adid, 0, false,
 805                                     kAudioDevicePropertyStreamFormat,
 806                                     &propertySize, &asbd);
 807     if(status)
 808       fatal(0, "AudioHardwareGetProperty: %d", (int)status);
 809     D(("mSampleRate       %f", asbd.mSampleRate));
 810     D(("mFormatID         %08lx", asbd.mFormatID));
 811     D(("mFormatFlags      %08lx", asbd.mFormatFlags));
 812     D(("mBytesPerPacket   %08lx", asbd.mBytesPerPacket));
 813     D(("mFramesPerPacket  %08lx", asbd.mFramesPerPacket));
 814     D(("mBytesPerFrame    %08lx", asbd.mBytesPerFrame));
 815     D(("mChannelsPerFrame %08lx", asbd.mChannelsPerFrame));
 816     D(("mBitsPerChannel   %08lx", asbd.mBitsPerChannel));
 817     D(("mReserved         %08lx", asbd.mReserved));
 818     if(asbd.mFormatID != kAudioFormatLinearPCM)
 819       fatal(0, "audio device does not support kAudioFormatLinearPCM");
 820     status = AudioDeviceAddIOProc(adid, adioproc, 0);
 821     if(status)
 822       fatal(0, "AudioDeviceAddIOProc: %d", (int)status);
 823     pthread_mutex_lock(&lock);
 824     for(;;) {
 825       /* Wait for the buffer to fill up a bit */
 826       fill_buffer();
 827       /* Start playing now */
 828       info("Playing...");
 829       next_timestamp = pheap_first(&packets)->timestamp;
 830       active = 1;
 831       status = AudioDeviceStart(adid, adioproc);
 832       if(status)
 833         fatal(0, "AudioDeviceStart: %d", (int)status);
 834       /* Wait until the buffer empties out */
 835       while(nsamples >= minbuffer
 836             || (nsamples > 0
 837                 && contains(pheap_first(&packets), next_timestamp)))
 838         pthread_cond_wait(&cond, &lock);
 839       /* Stop playing for a bit until the buffer re-fills */
 840       status = AudioDeviceStop(adid, adioproc);
 841       if(status)
 842         fatal(0, "AudioDeviceStop: %d", (int)status);
 843       active = 0;
 844       /* Go back round */
 845     }
 846   }
 847 #else
 848 # error No known audio API
 849 #endif
 850 }
 851
 852 /* display usage message and terminate */
 853 static void help(void) {
 854   xprintf("Usage:\n"
 855           "  disorder-playrtp [OPTIONS] ADDRESS [PORT]\n"
 856           "Options:\n"
 857           "  --device, -D DEVICE     Output device\n"
 858           "  --min, -m FRAMES        Buffer low water mark\n"
 859           "  --buffer, -b FRAMES     Buffer high water mark\n"
 860           "  --max, -x FRAMES        Buffer maximum size\n"
 861           "  --rcvbuf, -R BYTES      Socket receive buffer size\n"
 862           "  --multicast, -M GROUP   Join multicast group\n"
 863           "  --help, -h              Display usage message\n"
 864           "  --version, -V           Display version number\n"
 865           );
 866   xfclose(stdout);
 867   exit(0);
 868 }
 869
 870 /* display version number and terminate */
 871 static void version(void) {
 872   xprintf("disorder-playrtp version %s\n", disorder_version_string);
 873   xfclose(stdout);
 874   exit(0);
 875 }
 876
 877 int main(int argc, char **argv) {
 878   int n;
 879   struct addrinfo *res;
 880   struct stringlist sl;
 881   char *sockname;
 882   int rcvbuf, target_rcvbuf = 131072;
 883   socklen_t len;
 884   char *multicast_group = 0;
 885   struct ip_mreq mreq;
 886   struct ipv6_mreq mreq6;
 887
 888   static const struct addrinfo prefs = {
 889     AI_PASSIVE,
 890     PF_INET,
 891     SOCK_DGRAM,
 892     IPPROTO_UDP,
 893     0,
 894     0,
 895     0,
 896     0
 897   };
 898
 899   mem_init();
 900   if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale");
 901   while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:", options, 0)) >= 0) {
 902     switch(n) {
 903     case 'h': help();
 904     case 'V': version();
 905     case 'd': debugging = 1; break;
 906     case 'D': device = optarg; break;
 907     case 'm': minbuffer = 2 * atol(optarg); break;
 908     case 'b': readahead = 2 * atol(optarg); break;
 909     case 'x': maxbuffer = 2 * atol(optarg); break;
 910     case 'L': logfp = fopen(optarg, "w"); break;
 911     case 'R': target_rcvbuf = atoi(optarg); break;
 912     case 'M': multicast_group = optarg; break;
 913     default: fatal(0, "invalid option");
 914     }
 915   }
 916   if(!maxbuffer)
 917     maxbuffer = 4 * readahead;
 918   argc -= optind;
 919   argv += optind;
 920   if(argc < 1 || argc > 2)
 921     fatal(0, "usage: disorder-playrtp [OPTIONS] ADDRESS [PORT]");
 922   sl.n = argc;
 923   sl.s = argv;
 924   /* Listen for inbound audio data */
 925   if(!(res = get_address(&sl, &prefs, &sockname)))
 926     exit(1);
 927   if((rtpfd = socket(res->ai_family,
 928                      res->ai_socktype,
 929                      res->ai_protocol)) < 0)
 930     fatal(errno, "error creating socket");
 931   if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
 932     fatal(errno, "error binding socket to %s", sockname);
 933   if(multicast_group) {
 934     if((n = getaddrinfo(multicast_group, 0, &prefs, &res)))
 935       fatal(0, "getaddrinfo %s: %s", multicast_group, gai_strerror(n));
 936     switch(res->ai_family) {
 937     case PF_INET:
 938       mreq.imr_multiaddr = ((struct sockaddr_in *)res->ai_addr)->sin_addr;
 939       mreq.imr_interface.s_addr = 0;      /* use primary interface */
 940       if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 941                     &mreq, sizeof mreq) < 0)
 942         fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
 943       break;
 944     case PF_INET6:
 945       mreq6.ipv6mr_multiaddr = ((struct sockaddr_in6 *)res->ai_addr)->sin6_addr;
 946       memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
 947       if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
 948                     &mreq6, sizeof mreq6) < 0)
 949         fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
 950       break;
 951     default:
 952       fatal(0, "unsupported address family %d", res->ai_family);
 953     }
 954   }
 955   len = sizeof rcvbuf;
 956   if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
 957     fatal(errno, "error calling getsockopt SO_RCVBUF");
 958   if(target_rcvbuf > rcvbuf) {
 959     if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
 960                   &target_rcvbuf, sizeof target_rcvbuf) < 0)
 961       error(errno, "error calling setsockopt SO_RCVBUF %d",
 962             target_rcvbuf);
 963       /* We try to carry on anyway */
 964     else
 965       info("changed socket receive buffer from %d to %d",
 966            rcvbuf, target_rcvbuf);
 967   } else
 968     info("default socket receive buffer %d", rcvbuf);
 969   if(logfp)
 970     info("WARNING: -L option can impact performance");
 971   play_rtp();
 972   return 0;
 973 }
 974
 975 /*
 976 Local Variables:
 977 c-basic-offset:2
 978 comment-column:40
 979 fill-column:79
 980 indent-tabs-mode:nil
 981 End:
 982 */