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
  21 #include <config.h>
  22 #include "types.h"
  23
  24 #include <getopt.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <sys/socket.h>
  28 #include <sys/types.h>
  29 #include <sys/socket.h>
  30 #include <netdb.h>
  31 #include <pthread.h>
  32 #include <locale.h>
  33
  34 #include "log.h"
  35 #include "mem.h"
  36 #include "configuration.h"
  37 #include "addr.h"
  38 #include "syscalls.h"
  39 #include "rtp.h"
  40 #include "defs.h"
  41
  42 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
  43 # include <CoreAudio/AudioHardware.h>
  44 #endif
  45 #if API_ALSA
  46 #include <alsa/asoundlib.h>
  47 #endif
  48
  49 /** @brief RTP socket */
  50 static int rtpfd;
  51
  52 /** @brief Output device */
  53 static const char *device;
  54
  55 /** @brief Maximum samples per packet we'll support
  56  *
  57  * NB that two channels = two samples in this program.
  58  */
  59 #define MAXSAMPLES 2048
  60
  61 /** @brief Minimum buffer size
  62  *
  63  * We'll stop playing if there's only this many samples in the buffer. */
  64 #define MINBUFFER 8820
  65
  66 /** @brief Maximum sample size
  67  *
  68  * The maximum supported size (in bytes) of one sample. */
  69 #define MAXSAMPLESIZE 2
  70
  71 #define READAHEAD 88200                 /* how far to read ahead */
  72
  73 #define MAXBUFFER (3 * 88200)           /* maximum buffer contents */
  74
  75 /** @brief Received packet
  76  *
  77  * Packets are recorded in an ordered linked list. */
  78 struct packet {
  79   /** @brief Pointer to next packet
  80    * The next packet might not be immediately next: if packets are dropped
  81    * or mis-ordered there may be gaps at any given moment. */
  82   struct packet *next;
  83   /** @brief Number of samples in this packet */
  84   int nsamples;
  85   /** @brief Number of samples used from this packet */
  86   int nused;
  87   /** @brief Timestamp from RTP packet
  88    *
  89    * NB that "timestamps" are really sample counters.*/
  90   uint32_t timestamp;
  91 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
  92   /** @brief Converted sample data */
  93   float samples_float[MAXSAMPLES];
  94 #else
  95   /** @brief Raw sample data */
  96   unsigned char samples_raw[MAXSAMPLES * MAXSAMPLESIZE];
  97 #endif
  98 };
  99
 100 /** @brief Total number of samples available */
 101 static unsigned long nsamples;
 102
 103 /** @brief Linked list of packets
 104  *
 105  * In ascending order of timestamp. */
 106 static struct packet *packets;
 107
 108 /** @brief Timestamp of next packet to play.
 109  *
 110  * This is set to the timestamp of the last packet, plus the number of
 111  * samples it contained.
 112  */
 113 static uint32_t next_timestamp;
 114
 115 /** @brief Lock protecting @ref packets */
 116 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
 117
 118 /** @brief Condition variable signalled whenever @ref packets is changed */
 119 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 120
 121 static const struct option options[] = {
 122   { "help", no_argument, 0, 'h' },
 123   { "version", no_argument, 0, 'V' },
 124   { "debug", no_argument, 0, 'd' },
 125   { "device", required_argument, 0, 'D' },
 126   { 0, 0, 0, 0 }
 127 };
 128
 129 /** @brief Return true iff a < b in sequence-space arithmetic */
 130 static inline int lt(const struct packet *a, const struct packet *b) {
 131   return (uint32_t)(a->timestamp - b->timestamp) & 0x80000000;
 132 }
 133
 134 /** Background thread collecting samples
 135  *
 136  * This function collects samples, perhaps converts them to the target format,
 137  * and adds them to the packet list. */
 138 static void *listen_thread(void attribute((unused)) *arg) {
 139   struct packet *f = 0, **ff;
 140   int n;
 141   union {
 142     struct rtp_header header;
 143     uint8_t bytes[sizeof(uint16_t) * MAXSAMPLES + sizeof (struct rtp_header)];
 144   } packet;
 145   const uint16_t *const samples = (uint16_t *)(packet.bytes
 146                                                + sizeof (struct rtp_header));
 147
 148   for(;;) {
 149     if(!f)
 150       f = xmalloc(sizeof *f);
 151     n = read(rtpfd, packet.bytes, sizeof packet.bytes);
 152     if(n < 0) {
 153       switch(errno) {
 154       case EINTR:
 155         continue;
 156       default:
 157         fatal(errno, "error reading from socket");
 158       }
 159     }
 160     /* Ignore too-short packets */
 161     if((size_t)n <= sizeof (struct rtp_header))
 162       continue;
 163     /* Convert to target format */
 164     switch(packet.header.mpt & 0x7F) {
 165     case 10:
 166       f->nsamples = (n - sizeof (struct rtp_header)) / sizeof(uint16_t);
 167 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
 168       /* Convert to what Core Audio expects */
 169       for(n = 0; n < f->nsamples; ++n)
 170         f->samples_float[n] = (int16_t)ntohs(samples[n]) * (0.5f / 32767);
 171 #else
 172       /* ALSA can do any necessary conversion itself (though it might be better
 173        * to do any necessary conversion in the background) */
 174       memcpy(f->samples_raw, samples, n - sizeof (struct rtp_header));
 175 #endif
 176       break;
 177       /* TODO support other RFC3551 media types (when the speaker does) */
 178     default:
 179       fatal(0, "unsupported RTP payload type %d",
 180             packet.header.mpt & 0x7F);
 181     }
 182     f->nused = 0;
 183     f->timestamp = ntohl(packet.header.timestamp);
 184     pthread_mutex_lock(&lock);
 185     /* Stop reading if we've reached the maximum.
 186      *
 187      * This is rather unsatisfactory: it means that if packets get heavily
 188      * out of order then we guarantee dropouts.  But for now... */
 189     while(nsamples >= MAXBUFFER)
 190       pthread_cond_wait(&cond, &lock);
 191     for(ff = &packets; *ff && lt(*ff, f); ff = &(*ff)->next)
 192       ;
 193     /* So now either !*ff or *ff >= f */
 194     if(*ff && f->timestamp == (*ff)->timestamp) {
 195       /* *ff == f; a duplicate.  Ideally we avoid the translation step here,
 196        * but we'll worry about that another time. */
 197       free(f);
 198     } else {
 199       f->next = *ff;
 200       *ff = f;
 201       nsamples += f->nsamples;
 202       pthread_cond_broadcast(&cond);
 203     }
 204     pthread_mutex_unlock(&lock);
 205     f = 0;
 206   }
 207 }
 208
 209 #if HAVE_COREAUDIO_AUDIOHARDWARE_H
 210 static OSStatus adioproc(AudioDeviceID inDevice,
 211                          const AudioTimeStamp *inNow,
 212                          const AudioBufferList *inInputData,
 213                          const AudioTimeStamp *inInputTime,
 214                          AudioBufferList *outOutputData,
 215                          const AudioTimeStamp *inOutputTime,
 216                          void *inClientData) {
 217   UInt32 nbuffers = outOutputData->mNumberBuffers;
 218   AudioBuffer *ab = outOutputData->mBuffers;
 219   float *samplesOut;                    /* where to write samples to */
 220   size_t samplesOutLeft;                /* space left */
 221   size_t samplesInLeft;
 222   size_t samplesToCopy;
 223
 224   pthread_mutex_lock(&lock);
 225   samplesOut = ab->data;
 226   samplesOutLeft = ab->mDataByteSize / sizeof (float);
 227   while(packets && nbuffers > 0) {
 228     if(packets->used == packets->nsamples) {
 229       /* TODO if we dropped a packet then we should introduce a gap here */
 230       struct packet *const f = packets;
 231       packets = f->next;
 232       free(f);
 233       pthread_cond_broadcast(&cond);
 234       continue;
 235     }
 236     if(samplesOutLeft == 0) {
 237       --nbuffers;
 238       ++ab;
 239       samplesOut = ab->data;
 240       samplesOutLeft = ab->mDataByteSize / sizeof (float);
 241       continue;
 242     }
 243     /* Now: (1) there is some data left to read
 244      *      (2) there is some space to put it */
 245     samplesInLeft = packets->nsamples - packets->used;
 246     samplesToCopy = (samplesInLeft < samplesOutLeft
 247                      ? samplesInLeft : samplesOutLeft);
 248     memcpy(samplesOut, packet->samples + packets->used, samplesToCopy);
 249     packets->used += samplesToCopy;
 250     samplesOut += samplesToCopy;
 251     samesOutLeft -= samplesToCopy;
 252   }
 253   pthread_mutex_unlock(&lock);
 254   return 0;
 255 }
 256 #endif
 257
 258 static void play_rtp(void) {
 259   pthread_t ltid;
 260
 261   /* We receive and convert audio data in a background thread */
 262   pthread_create(&ltid, 0, listen_thread, 0);
 263 #if API_ALSA
 264   {
 265     snd_pcm_t *pcm;
 266     snd_pcm_hw_params_t *hwparams;
 267     snd_pcm_sw_params_t *swparams;
 268     /* Only support one format for now */
 269     const int sample_format = SND_PCM_FORMAT_S16_BE;
 270     unsigned rate = 44100;
 271     const int channels = 2;
 272     const int samplesize = channels * sizeof(uint16_t);
 273     snd_pcm_uframes_t pcm_bufsize = MAXSAMPLES * samplesize * 3;
 274     /* If we can write more than this many samples we'll get a wakeup */
 275     const int avail_min = 256;
 276     snd_pcm_sframes_t frames_written;
 277     size_t samples_written;
 278     int prepared = 1;
 279     int err;
 280
 281     /* Open ALSA */
 282     if((err = snd_pcm_open(&pcm,
 283                            device ? device : "default",
 284                            SND_PCM_STREAM_PLAYBACK,
 285                            SND_PCM_NONBLOCK)))
 286       fatal(0, "error from snd_pcm_open: %d", err);
 287     /* Set up 'hardware' parameters */
 288     snd_pcm_hw_params_alloca(&hwparams);
 289     if((err = snd_pcm_hw_params_any(pcm, hwparams)) < 0)
 290       fatal(0, "error from snd_pcm_hw_params_any: %d", err);
 291     if((err = snd_pcm_hw_params_set_access(pcm, hwparams,
 292                                            SND_PCM_ACCESS_RW_INTERLEAVED)) < 0)
 293       fatal(0, "error from snd_pcm_hw_params_set_access: %d", err);
 294     if((err = snd_pcm_hw_params_set_format(pcm, hwparams,
 295                                            sample_format)) < 0)
 296       fatal(0, "error from snd_pcm_hw_params_set_format (%d): %d",
 297             sample_format, err);
 298     if((err = snd_pcm_hw_params_set_rate_near(pcm, hwparams, &rate, 0)) < 0)
 299       fatal(0, "error from snd_pcm_hw_params_set_rate (%d): %d",
 300             rate, err);
 301     if((err = snd_pcm_hw_params_set_channels(pcm, hwparams,
 302                                              channels)) < 0)
 303       fatal(0, "error from snd_pcm_hw_params_set_channels (%d): %d",
 304             channels, err);
 305     if((err = snd_pcm_hw_params_set_buffer_size_near(pcm, hwparams,
 306                                                      &pcm_bufsize)) < 0)
 307       fatal(0, "error from snd_pcm_hw_params_set_buffer_size (%d): %d",
 308             MAXSAMPLES * samplesize * 3, err);
 309     if((err = snd_pcm_hw_params(pcm, hwparams)) < 0)
 310       fatal(0, "error calling snd_pcm_hw_params: %d", err);
 311     /* Set up 'software' parameters */
 312     snd_pcm_sw_params_alloca(&swparams);
 313     if((err = snd_pcm_sw_params_current(pcm, swparams)) < 0)
 314       fatal(0, "error calling snd_pcm_sw_params_current: %d", err);
 315     if((err = snd_pcm_sw_params_set_avail_min(pcm, swparams, avail_min)) < 0)
 316       fatal(0, "error calling snd_pcm_sw_params_set_avail_min %d: %d",
 317             avail_min, err);
 318     if((err = snd_pcm_sw_params(pcm, swparams)) < 0)
 319       fatal(0, "error calling snd_pcm_sw_params: %d", err);
 320
 321     /* Ready to go */
 322
 323     pthread_mutex_lock(&lock);
 324     for(;;) {
 325       /* Wait for the buffer to fill up a bit */
 326       while(nsamples < READAHEAD)
 327         pthread_cond_wait(&cond, &lock);
 328       if(!prepared) {
 329         if((err = snd_pcm_prepare(pcm)))
 330           fatal(0, "error calling snd_pcm_prepare: %d", err);
 331         prepared = 1;
 332       }
 333       /* Wait until the buffer empties out */
 334       while(nsamples >= MINBUFFER) {
 335         /* Wait for ALSA to ask us for more data */
 336         pthread_mutex_unlock(&lock);
 337         snd_pcm_wait(pcm, -1);
 338         pthread_mutex_lock(&lock);
 339         /* ALSA wants more data */
 340         if(packets && packets->timestamp + packets->nused == next_timestamp) {
 341           /* Hooray, we have a packet we can play */
 342           const size_t samples_available = packets->nsamples - packets->nused;
 343           const size_t frames_available = samples_available / 2;
 344
 345           frames_written = snd_pcm_writei(pcm,
 346                                           packets->samples_raw + packets->nused,
 347                                           frames_available);
 348           if(frames_written < 0)
 349             fatal(0, "error calling snd_pcm_writei: %d", err);
 350           samples_written = frames_written * 2;
 351           packets->nused += samples_written;
 352           next_timestamp += samples_written;
 353           if(packets->nused == packets->nsamples) {
 354             struct packet *f = packets;
 355
 356             packets = f->next;
 357             nsamples -= f->nsamples;
 358             free(f);
 359             pthread_cond_broadcast(&cond);
 360           }
 361         } else {
 362           /* We don't have anything to play!  We'd better play some 0s. */
 363           static const uint16_t zeros[1024];
 364           size_t samples_available = 1024, frames_available;
 365           if(packets && next_timestamp + samples_available > packets->timestamp)
 366             samples_available = packets->timestamp - next_timestamp;
 367           frames_available = samples_available / 2;
 368           frames_written = snd_pcm_writei(pcm,
 369                                           zeros,
 370                                           frames_available);
 371           if(frames_written < 0)
 372             fatal(0, "error calling snd_pcm_writei: %d", err);
 373           next_timestamp += samples_written;
 374         }
 375       }
 376       /* We stop playing for a bit until the buffer re-fills */
 377       pthread_mutex_unlock(&lock);
 378       if((err = snd_pcm_drain(pcm)))
 379         fatal(0, "error calling snd_pcm_drain: %d", err);
 380       prepared = 0;
 381       pthread_mutex_lock(&lock);
 382     }
 383
 384   }
 385 #elif HAVE_COREAUDIO_AUDIOHARDWARE_H
 386   {
 387     OSStatus status;
 388     UInt32 propertySize;
 389     AudioDeviceID adid;
 390     AudioStreamBasicDescription asbd;
 391
 392     /* If this looks suspiciously like libao's macosx driver there's an
 393      * excellent reason for that... */
 394
 395     /* TODO report errors as strings not numbers */
 396     propertySize = sizeof adid;
 397     status = AudioHardwareGetProperty(kAudioHardwarePropertyDefaultOutputDevice,
 398                                       &propertySize, &adid);
 399     if(status)
 400       fatal(0, "AudioHardwareGetProperty: %d", (int)status);
 401     if(adid == kAudioDeviceUnknown)
 402       fatal(0, "no output device");
 403     propertySize = sizeof asbd;
 404     status = AudioDeviceGetProperty(adid, 0, false,
 405                                     kAudioDevicePropertyStreamFormat,
 406                                     &propertySize, &asbd);
 407     if(status)
 408       fatal(0, "AudioHardwareGetProperty: %d", (int)status);
 409     D(("mSampleRate       %f", asbd.mSampleRate));
 410     D(("mFormatID         %08"PRIx32, asbd.mFormatID));
 411     D(("mFormatFlags      %08"PRIx32, asbd.mFormatFlags));
 412     D(("mBytesPerPacket   %08"PRIx32, asbd.mBytesPerPacket));
 413     D(("mFramesPerPacket  %08"PRIx32, asbd.mFramesPerPacket));
 414     D(("mBytesPerFrame    %08"PRIx32, asbd.mBytesPerFrame));
 415     D(("mChannelsPerFrame %08"PRIx32, asbd.mChannelsPerFrame));
 416     D(("mBitsPerChannel   %08"PRIx32, asbd.mBitsPerChannel));
 417     D(("mReserved         %08"PRIx32, asbd.mReserved));
 418     if(asbd.mFormatID != kAudioFormatLinearPCM)
 419       fatal(0, "audio device does not support kAudioFormatLinearPCM");
 420     status = AudioDeviceAddIOProc(adid, adioproc, 0);
 421     if(status)
 422       fatal(0, "AudioDeviceAddIOProc: %d", (int)status);
 423     pthread_mutex_lock(&lock);
 424     for(;;) {
 425       /* Wait for the buffer to fill up a bit */
 426       while(nsamples < READAHEAD)
 427         pthread_cond_wait(&cond, &lock);
 428       /* Start playing now */
 429       status = AudioDeviceStart(adid, adioproc);
 430       if(status)
 431         fatal(0, "AudioDeviceStart: %d", (int)status);
 432       /* Wait until the buffer empties out */
 433       while(nsamples >= MINBUFFER)
 434         pthread_cond_wait(&cond, &lock);
 435       /* Stop playing for a bit until the buffer re-fills */
 436       status = AudioDeviceStop(adid, adioproc);
 437       if(status)
 438         fatal(0, "AudioDeviceStop: %d", (int)status);
 439       /* Go back round */
 440     }
 441   }
 442 #else
 443 # error No known audio API
 444 #endif
 445 }
 446
 447 /* display usage message and terminate */
 448 static void help(void) {
 449   xprintf("Usage:\n"
 450           "  disorder-playrtp [OPTIONS] ADDRESS [PORT]\n"
 451           "Options:\n"
 452           "  --help, -h              Display usage message\n"
 453           "  --version, -V           Display version number\n"
 454           "  --debug, -d             Turn on debugging\n"
 455           "  --device, -D DEVICE     Output device\n");
 456   xfclose(stdout);
 457   exit(0);
 458 }
 459
 460 /* display version number and terminate */
 461 static void version(void) {
 462   xprintf("disorder-playrtp version %s\n", disorder_version_string);
 463   xfclose(stdout);
 464   exit(0);
 465 }
 466
 467 int main(int argc, char **argv) {
 468   int n;
 469   struct addrinfo *res;
 470   struct stringlist sl;
 471   char *sockname;
 472
 473   static const struct addrinfo prefs = {
 474     AI_PASSIVE,
 475     PF_INET,
 476     SOCK_DGRAM,
 477     IPPROTO_UDP,
 478     0,
 479     0,
 480     0,
 481     0
 482   };
 483
 484   mem_init();
 485   if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale");
 486   while((n = getopt_long(argc, argv, "hVdD", options, 0)) >= 0) {
 487     switch(n) {
 488     case 'h': help();
 489     case 'V': version();
 490     case 'd': debugging = 1; break;
 491     case 'D': device = optarg; break;
 492     default: fatal(0, "invalid option");
 493     }
 494   }
 495   argc -= optind;
 496   argv += optind;
 497   if(argc < 1 || argc > 2)
 498     fatal(0, "usage: disorder-playrtp [OPTIONS] ADDRESS [PORT]");
 499   sl.n = argc;
 500   sl.s = argv;
 501   /* Listen for inbound audio data */
 502   if(!(res = get_address(&sl, &prefs, &sockname)))
 503     exit(1);
 504   if((rtpfd = socket(res->ai_family,
 505                      res->ai_socktype,
 506                      res->ai_protocol)) < 0)
 507     fatal(errno, "error creating socket");
 508   if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
 509     fatal(errno, "error binding socket to %s", sockname);
 510   play_rtp();
 511   return 0;
 512 }
 513
 514 /*
 515 Local Variables:
 516 c-basic-offset:2
 517 comment-column:40
 518 fill-column:79
 519 indent-tabs-mode:nil
 520 End:
 521 */