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