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