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