* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
* USA
*/
+/** @file clients/playrtp.c
+ * @brief RTP player
+ *
+ * This player supports Linux (<a href="http://www.alsa-project.org/">ALSA</a>)
+ * and Apple Mac (<a
+ * href="http://developer.apple.com/audio/coreaudio.html">Core Audio</a>)
+ * systems. There is no support for Microsoft Windows yet, and that will in
+ * fact probably an entirely separate program.
+ *
+ * The program runs (at least) three threads. listen_thread() is responsible
+ * for reading RTP packets off the wire and adding them to the linked list @ref
+ * received_packets, assuming they are basically sound. queue_thread() takes
+ * packets off this linked list and adds them to @ref packets (an operation
+ * which might be much slower due to contention for @ref lock).
+ *
+ * The main thread is responsible for actually playing audio. In ALSA this
+ * means it waits until ALSA says it's ready for more audio which it then
+ * plays. See @ref clients/playrtp-alsa.c.
+ *
+ * In Core Audio the main thread is only responsible for starting and stopping
+ * play: the system does the actual playback in its own private thread, and
+ * calls adioproc() to fetch the audio data. See @ref
+ * clients/playrtp-coreaudio.c.
+ *
+ * Sometimes it happens that there is no audio available to play. This may
+ * because the server went away, or a packet was dropped, or the server
+ * deliberately did not send any sound because it encountered a silence.
+ *
+ * Assumptions:
+ * - it is safe to read uint32_t values without a lock protecting them
+ */
#include <config.h>
#include "types.h"
#include <pthread.h>
#include <locale.h>
#include <sys/uio.h>
+#include <string.h>
+#include <assert.h>
+#include <errno.h>
#include "log.h"
#include "mem.h"
#include "syscalls.h"
#include "rtp.h"
#include "defs.h"
-
-#if HAVE_COREAUDIO_AUDIOHARDWARE_H
-# include <CoreAudio/AudioHardware.h>
-#endif
-#if API_ALSA
-#include <alsa/asoundlib.h>
-#endif
+#include "vector.h"
+#include "heap.h"
+#include "timeval.h"
+#include "playrtp.h"
#define readahead linux_headers_are_borked
static FILE *logfp;
/** @brief Output device */
-static const char *device;
-
-/** @brief Maximum samples per packet we'll support
- *
- * NB that two channels = two samples in this program.
- */
-#define MAXSAMPLES 2048
+const char *device;
/** @brief Minimum low watermark
*
* We'll stop playing if there's only this many samples in the buffer. */
-static unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */
-
-/** @brief Maximum sample size
- *
- * The maximum supported size (in bytes) of one sample. */
-#define MAXSAMPLESIZE 2
+unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */
/** @brief Buffer high watermark
*
* We'll stop reading from the network if we have this many samples. */
static unsigned maxbuffer;
-/** @brief Number of samples to infill by in one go */
-#define INFILL_SAMPLES (44100 * 2) /* 1s */
+/** @brief Received packets
+ * Protected by @ref receive_lock
+ *
+ * Received packets are added to this list, and queue_thread() picks them off
+ * it and adds them to @ref packets. Whenever a packet is added to it, @ref
+ * receive_cond is signalled.
+ */
+struct packet *received_packets;
-/** @brief Received packet */
-struct packet {
- /** @brief Number of samples in this packet */
- uint32_t nsamples;
- /** @brief Timestamp from RTP packet
- *
- * NB that "timestamps" are really sample counters.*/
- uint32_t timestamp;
- /** @brief Raw sample data */
- unsigned char samples_raw[MAXSAMPLES * MAXSAMPLESIZE];
-};
+/** @brief Tail of @ref received_packets
+ * Protected by @ref receive_lock
+ */
+struct packet **received_tail = &received_packets;
-/** @brief Total number of samples available */
-static unsigned long nsamples;
+/** @brief Lock protecting @ref received_packets
+ *
+ * Only listen_thread() and queue_thread() ever hold this lock. It is vital
+ * that queue_thread() not hold it any longer than it strictly has to. */
+pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER;
-/** @brief Mapping of sequence numbers to packets
+/** @brief Condition variable signalled when @ref received_packets is updated
*
- * This isn't very efficient - 256KB on 32-bit machines, 512KB if you do a
- * 64-bit build for some reason. It can be optimized later if need be. */
-static struct packet *packets[65536];
+ * Used by listen_thread() to notify queue_thread() that it has added another
+ * packet to @ref received_packets. */
+pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER;
-/** @brief Total number of packets */
-static unsigned npackets;
+/** @brief Length of @ref received_packets */
+uint32_t nreceived;
+
+/** @brief Binary heap of received packets */
+struct pheap packets;
+
+/** @brief Total number of samples available
+ *
+ * We make this volatile because we inspect it without a protecting lock,
+ * so the usual pthread_* guarantees aren't available.
+ */
+volatile uint32_t nsamples;
/** @brief Timestamp of next packet to play.
*
* This is set to the timestamp of the last packet, plus the number of
* samples it contained. Only valid if @ref active is nonzero.
*/
-static uint32_t next_timestamp;
+uint32_t next_timestamp;
/** @brief True if actively playing
*
* This is true when playing and false when just buffering. */
-static int active;
-
-/** @brief Sequence number of next packet we expxect to play */
-static uint16_t sequence;
-
-/** @brief Structure of free packet list */
-union free_packet {
- struct packet p;
- union free_packet *next;
-};
+int active;
-/** @brief Linked list of free packets */
-static union free_packet *free_packets;
+/** @brief Lock protecting @ref packets */
+pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
-/** @brief Array of new free packets */
-static union free_packet *next_free_packet;
+/** @brief Condition variable signalled whenever @ref packets is changed */
+pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
-/** @brief Count of new free packets */
-static size_t count_free_packets;
+#if HAVE_ALSA_ASOUNDLIB_H
+# define DEFAULT_BACKEND playrtp_alsa
+#elif HAVE_SYS_SOUNDCARD_H
+# define DEFAULT_BACKEND playrtp_oss
+#elif HAVE_COREAUDIO_AUDIOHARDWARE_H
+# define DEFAULT_BACKEND playrtp_coreaudio
+#else
+# error No known backend
+#endif
-/** @brief Lock protecting @ref packets */
-static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
+/** @brief Backend to play with */
+static void (*backend)(void) = &DEFAULT_BACKEND;
-/** @brief Condition variable signalled whenever @ref packets is changed */
-static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
+HEAP_DEFINE(pheap, struct packet *, lt_packet);
static const struct option options[] = {
{ "help", no_argument, 0, 'h' },
{ "min", required_argument, 0, 'm' },
{ "max", required_argument, 0, 'x' },
{ "buffer", required_argument, 0, 'b' },
+ { "rcvbuf", required_argument, 0, 'R' },
+ { "multicast", required_argument, 0, 'M' },
+#if HAVE_SYS_SOUNDCARD_H
+ { "oss", no_argument, 0, 'o' },
+#endif
+#if HAVE_ALSA_ASOUNDLIB_H
+ { "alsa", no_argument, 0, 'a' },
+#endif
+#if HAVE_COREAUDIO_AUDIOHARDWARE_H
+ { "core-audio", no_argument, 0, 'c' },
+#endif
{ 0, 0, 0, 0 }
};
-/** @brief Return a new packet
+/** @brief Drop the first packet
*
- * Assumes that @ref lock is held. */
-static struct packet *new_packet(void) {
- struct packet *p;
-
- if(free_packets) {
- p = &free_packets->p;
- free_packets = free_packets->next;
- } else {
- if(!count_free_packets) {
- next_free_packet = xcalloc(1024, sizeof (union free_packet));
- count_free_packets = 1024;
- }
- p = &(next_free_packet++)->p;
- --count_free_packets;
+ * Assumes that @ref lock is held.
+ */
+static void drop_first_packet(void) {
+ if(pheap_count(&packets)) {
+ struct packet *const p = pheap_remove(&packets);
+ nsamples -= p->nsamples;
+ playrtp_free_packet(p);
+ pthread_cond_broadcast(&cond);
}
- return p;
}
-/** @brief Free a packet
+/** @brief Background thread adding packets to heap
*
- * Assumes that @ref lock is held. */
-static void free_packet(struct packet *p) {
- union free_packet *u = (union free_packet *)p;
- u->next = free_packets;
- free_packets = u;
-}
-
-/** @brief Return true iff a < b in sequence-space arithmetic */
-static inline int lt(uint32_t a, uint32_t b) {
- return (uint32_t)(a - b) & 0x80000000;
-}
-
-/** @brief Return true iff a >= b in sequence-space arithmetic */
-static inline int ge(uint32_t a, uint32_t b) {
- return !lt(a, b);
-}
-
-/** @brief Return true iff a > b in sequence-space arithmetic */
-static inline int gt(uint32_t a, uint32_t b) {
- return lt(b, a);
-}
-
-/** @brief Return true iff a <= b in sequence-space arithmetic */
-static inline int le(uint32_t a, uint32_t b) {
- return !lt(b, a);
-}
+ * This just transfers packets from @ref received_packets to @ref packets. It
+ * is important that it holds @ref receive_lock for as little time as possible,
+ * in order to minimize the interval between calls to read() in
+ * listen_thread().
+ */
+static void *queue_thread(void attribute((unused)) *arg) {
+ struct packet *p;
-/** @brief Drop the packet at the head of the queue */
-static void drop_packet(unsigned sequence) {
- if(packets[sequence]) {
- nsamples -= packets[sequence]->nsamples;
- free_packet(packets[sequence]);
- packets[sequence] = 0;
+ for(;;) {
+ /* Get the next packet */
+ pthread_mutex_lock(&receive_lock);
+ while(!received_packets)
+ pthread_cond_wait(&receive_cond, &receive_lock);
+ p = received_packets;
+ received_packets = p->next;
+ if(!received_packets)
+ received_tail = &received_packets;
+ --nreceived;
+ pthread_mutex_unlock(&receive_lock);
+ /* Add it to the heap */
+ pthread_mutex_lock(&lock);
+ pheap_insert(&packets, p);
+ nsamples += p->nsamples;
pthread_cond_broadcast(&cond);
- --npackets;
+ pthread_mutex_unlock(&lock);
}
}
/** @brief Background thread collecting samples
*
* This function collects samples, perhaps converts them to the target format,
- * and adds them to the packet list. */
+ * and adds them to the packet list.
+ *
+ * It is crucial that the gap between successive calls to read() is as small as
+ * possible: otherwise packets will be dropped.
+ *
+ * We use a binary heap to ensure that the unavoidable effort is at worst
+ * logarithmic in the total number of packets - in fact if packets are mostly
+ * received in order then we will largely do constant work per packet since the
+ * newest packet will always be last.
+ *
+ * Of more concern is that we must acquire the lock on the heap to add a packet
+ * to it. If this proves a problem in practice then the answer would be
+ * (probably doubly) linked list with new packets added the end and a second
+ * thread which reads packets off the list and adds them to the heap.
+ *
+ * We keep memory allocation (mostly) very fast by keeping pre-allocated
+ * packets around; see @ref playrtp_new_packet().
+ */
static void *listen_thread(void attribute((unused)) *arg) {
struct packet *p = 0;
int n;
struct iovec iov[2];
for(;;) {
- if(!p) {
- pthread_mutex_lock(&lock);
- p = new_packet();
- pthread_mutex_unlock(&lock);
- }
+ if(!p)
+ p = playrtp_new_packet();
iov[0].iov_base = &header;
iov[0].iov_len = sizeof header;
iov[1].iov_base = p->samples_raw;
- iov[1].iov_len = sizeof p->samples_raw;
+ iov[1].iov_len = sizeof p->samples_raw / sizeof *p->samples_raw;
n = readv(rtpfd, iov, 2);
if(n < 0) {
switch(errno) {
timestamp, next_timestamp);
continue;
}
- pthread_mutex_lock(&lock);
- p = new_packet();
+ p->next = 0;
+ p->flags = 0;
p->timestamp = timestamp;
/* Convert to target format */
+ if(header.mpt & 0x80)
+ p->flags |= IDLE;
switch(header.mpt & 0x7F) {
case 10:
p->nsamples = (n - sizeof header) / sizeof(uint16_t);
- /* ALSA can do any necessary conversion itself (though it might be better
- * to do any necessary conversion in the background) */
- /* TODO we could readv into the buffer */
break;
/* TODO support other RFC3551 media types (when the speaker does) */
default:
* This is rather unsatisfactory: it means that if packets get heavily
* out of order then we guarantee dropouts. But for now... */
if(nsamples >= maxbuffer) {
- info("buffer full");
+ pthread_mutex_lock(&lock);
while(nsamples >= maxbuffer)
pthread_cond_wait(&cond, &lock);
+ pthread_mutex_unlock(&lock);
}
- /* If there's a packet there already we overwrite it; perhaps it is left
- * over from an earlier stage. */
- drop_packet(seq);
- /* Record this packet */
- packets[seq] = p;
- /* If we currently have no idea where to start playing, this is it */
- if(!npackets)
- sequence = seq;
- ++npackets;
- nsamples += p->nsamples;
- pthread_cond_broadcast(&cond);
- pthread_mutex_unlock(&lock);
+ /* Add the packet to the receive queue */
+ pthread_mutex_lock(&receive_lock);
+ *received_tail = p;
+ received_tail = &p->next;
+ ++nreceived;
+ pthread_cond_signal(&receive_cond);
+ pthread_mutex_unlock(&receive_lock);
+ /* We'll need a new packet */
+ p = 0;
}
}
-/** @brief Return true if @p p contains @p timestamp */
-static inline int contains(const struct packet *p, uint32_t timestamp) {
- const uint32_t packet_start = p->timestamp;
- const uint32_t packet_end = p->timestamp + p->nsamples;
-
- return (ge(timestamp, packet_start)
- && lt(timestamp, packet_end));
+/** @brief Wait until the buffer is adequately full
+ *
+ * Must be called with @ref lock held.
+ */
+void playrtp_fill_buffer(void) {
+ while(nsamples)
+ drop_first_packet();
+ info("Buffering...");
+ while(nsamples < readahead)
+ pthread_cond_wait(&cond, &lock);
+ next_timestamp = pheap_first(&packets)->timestamp;
+ active = 1;
}
-#if HAVE_COREAUDIO_AUDIOHARDWARE_H
-/** @brief Callback from Core Audio */
-static OSStatus adioproc
- (AudioDeviceID attribute((unused)) inDevice,
- const AudioTimeStamp attribute((unused)) *inNow,
- const AudioBufferList attribute((unused)) *inInputData,
- const AudioTimeStamp attribute((unused)) *inInputTime,
- AudioBufferList *outOutputData,
- const AudioTimeStamp attribute((unused)) *inOutputTime,
- void attribute((unused)) *inClientData) {
- UInt32 nbuffers = outOutputData->mNumberBuffers;
- AudioBuffer *ab = outOutputData->mBuffers;
- const struct packet *p;
-
- pthread_mutex_lock(&lock);
- while(nbuffers > 0) {
- float *samplesOut = ab->mData;
- size_t samplesOutLeft = ab->mDataByteSize / sizeof (float);
-
- while(samplesOutLeft > 0) {
- /* Look for a suitable packet, dropping any unsuitable ones along the
- * way. Unsuitable packets are ones that are in the past. */
- while(npackets
- && (!packets[sequence]
- || le(packets[sequence]->timestamp
- + packets[sequence]->nsamples,
- next_timestamp)))
- drop_packet(sequence++);
- p = packets[sequence];
- if(p) {
- if(contains(p, next_timestamp)) {
- /* This packet is suitable */
- const uint32_t packet_end = p->timestamp + p->nsamples;
- const uint32_t offset = next_timestamp - p->timestamp;
- const uint16_t *ptr =
- (void *)(p->samples_raw + offset * sizeof (uint16_t));
- uint32_t samples_available = packet_end - next_timestamp;
- if(samples_available > samplesOutLeft)
- samples_available = samplesOutLeft;
- next_timestamp += samples_available;
- samplesOutLeft -= samples_available;
- while(samples_available-- > 0)
- *samplesOut++ = (int16_t)ntohs(*ptr++) * (0.5 / 32767);
- /* We don't bother junking the packet or advancing sequence - that'll
- * be dealt with next time round */
- continue;
- }
- }
- /* We didn't find a suitable packet (though there might still be
- * unsuitable ones). We infill with 0s. */
- if(p) {
- /* There is a next packet, only infill up to that point */
- uint32_t samples_available = p->timestamp - next_timestamp;
-
- if(samples_available > samplesOutLeft)
- samples_available = samplesOutLeft;
- info("infill by %"PRIu32, samples_available);
- /* Convniently the buffer is 0 to start with */
- next_timestamp += samples_available;
- samplesOut += samples_available;
- samplesOutLeft -= samples_available;
- } else {
- /* There's no next packet at all */
- info("infilled by %zu", samplesOutLeft);
- next_timestamp += samplesOutLeft;
- samplesOut += samplesOutLeft;
- samplesOutLeft = 0;
- }
- }
- ++ab;
- --nbuffers;
+/** @brief Find next packet
+ * @return Packet to play or NULL if none found
+ *
+ * The return packet is merely guaranteed not to be in the past: it might be
+ * the first packet in the future rather than one that is actually suitable to
+ * play.
+ *
+ * Must be called with @ref lock held.
+ */
+struct packet *playrtp_next_packet(void) {
+ while(pheap_count(&packets)) {
+ struct packet *const p = pheap_first(&packets);
+ if(le(p->timestamp + p->nsamples, next_timestamp)) {
+ /* This packet is in the past. Drop it and try another one. */
+ drop_first_packet();
+ } else
+ /* This packet is NOT in the past. (It might be in the future
+ * however.) */
+ return p;
}
- pthread_mutex_unlock(&lock);
return 0;
}
-#endif
/** @brief Play an RTP stream
*
/* We receive and convert audio data in a background thread */
pthread_create(<id, 0, listen_thread, 0);
-#if API_ALSA
- {
- snd_pcm_t *pcm;
- snd_pcm_hw_params_t *hwparams;
- snd_pcm_sw_params_t *swparams;
- /* Only support one format for now */
- const int sample_format = SND_PCM_FORMAT_S16_BE;
- unsigned rate = 44100;
- const int channels = 2;
- const int samplesize = channels * sizeof(uint16_t);
- snd_pcm_uframes_t pcm_bufsize = MAXSAMPLES * samplesize * 3;
- /* If we can write more than this many samples we'll get a wakeup */
- const int avail_min = 256;
- snd_pcm_sframes_t frames_written;
- size_t samples_written;
- int prepared = 1;
- int err;
- int infilling = 0, escape = 0;
- time_t logged, now;
- uint32_t packet_start, packet_end;
-
- /* Open ALSA */
- if((err = snd_pcm_open(&pcm,
- device ? device : "default",
- SND_PCM_STREAM_PLAYBACK,
- SND_PCM_NONBLOCK)))
- fatal(0, "error from snd_pcm_open: %d", err);
- /* Set up 'hardware' parameters */
- snd_pcm_hw_params_alloca(&hwparams);
- if((err = snd_pcm_hw_params_any(pcm, hwparams)) < 0)
- fatal(0, "error from snd_pcm_hw_params_any: %d", err);
- if((err = snd_pcm_hw_params_set_access(pcm, hwparams,
- SND_PCM_ACCESS_RW_INTERLEAVED)) < 0)
- fatal(0, "error from snd_pcm_hw_params_set_access: %d", err);
- if((err = snd_pcm_hw_params_set_format(pcm, hwparams,
- sample_format)) < 0)
- fatal(0, "error from snd_pcm_hw_params_set_format (%d): %d",
- sample_format, err);
- if((err = snd_pcm_hw_params_set_rate_near(pcm, hwparams, &rate, 0)) < 0)
- fatal(0, "error from snd_pcm_hw_params_set_rate (%d): %d",
- rate, err);
- if((err = snd_pcm_hw_params_set_channels(pcm, hwparams,
- channels)) < 0)
- fatal(0, "error from snd_pcm_hw_params_set_channels (%d): %d",
- channels, err);
- if((err = snd_pcm_hw_params_set_buffer_size_near(pcm, hwparams,
- &pcm_bufsize)) < 0)
- fatal(0, "error from snd_pcm_hw_params_set_buffer_size (%d): %d",
- MAXSAMPLES * samplesize * 3, err);
- if((err = snd_pcm_hw_params(pcm, hwparams)) < 0)
- fatal(0, "error calling snd_pcm_hw_params: %d", err);
- /* Set up 'software' parameters */
- snd_pcm_sw_params_alloca(&swparams);
- if((err = snd_pcm_sw_params_current(pcm, swparams)) < 0)
- fatal(0, "error calling snd_pcm_sw_params_current: %d", err);
- if((err = snd_pcm_sw_params_set_avail_min(pcm, swparams, avail_min)) < 0)
- fatal(0, "error calling snd_pcm_sw_params_set_avail_min %d: %d",
- avail_min, err);
- if((err = snd_pcm_sw_params(pcm, swparams)) < 0)
- fatal(0, "error calling snd_pcm_sw_params: %d", err);
-
- /* Ready to go */
-
- time(&logged);
- pthread_mutex_lock(&lock);
- for(;;) {
- /* Wait for the buffer to fill up a bit */
- logged = now;
- info("%lu samples in buffer (%lus)", nsamples,
- nsamples / (44100 * 2));
- info("Buffering...");
- while(nsamples < readahead)
- pthread_cond_wait(&cond, &lock);
- if(!prepared) {
- if((err = snd_pcm_prepare(pcm)))
- fatal(0, "error calling snd_pcm_prepare: %d", err);
- prepared = 1;
- }
- assert(sequence != -1);
- /* Start at the first available packet */
- next_timestamp = packets[sequence]->timestamp;
- active = 1;
- infilling = 0;
- escape = 0;
- logged = now;
- info("%lu samples in buffer (%lus)", nsamples,
- nsamples / (44100 * 2));
- info("Playing...");
- /* Wait until the buffer empties out */
- while(nsamples >= minbuffer && !escape) {
- time(&now);
- if(now > logged + 10) {
- logged = now;
- info("%lu samples in buffer (%lus)", nsamples,
- nsamples / (44100 * 2));
- }
- if(packets
- && ge(next_timestamp, packets->timestamp + packets->nsamples)) {
- info("dropping buffered past packet %"PRIx32" < %"PRIx32,
- packets->timestamp, next_timestamp);
- drop_first_packet();
- continue;
- }
- /* Wait for ALSA to ask us for more data */
- pthread_mutex_unlock(&lock);
- write(2, ".", 1); /* TODO remove me sometime */
- switch(err = snd_pcm_wait(pcm, -1)) {
- case 0:
- info("snd_pcm_wait timed out");
- break;
- case 1:
- break;
- default:
- fatal(0, "snd_pcm_wait returned %d", err);
- }
- pthread_mutex_lock(&lock);
- /* ALSA is ready for more data */
- packet_start = packets->timestamp;
- packet_end = packets->timestamp + packets->nsamples;
- if(ge(next_timestamp, packet_start)
- && lt(next_timestamp, packet_end)) {
- /* The target timestamp is somewhere in this packet */
- const uint32_t offset = next_timestamp - packets->timestamp;
- const uint32_t samples_available = (packets->timestamp + packets->nsamples) - next_timestamp;
- const size_t frames_available = samples_available / 2;
-
- frames_written = snd_pcm_writei(pcm,
- packets->samples_raw + offset,
- frames_available);
- if(frames_written < 0) {
- switch(frames_written) {
- case -EAGAIN:
- info("snd_pcm_wait() returned but we got -EAGAIN!");
- break;
- case -EPIPE:
- error(0, "error calling snd_pcm_writei: %ld",
- (long)frames_written);
- escape = 1;
- break;
- default:
- fatal(0, "error calling snd_pcm_writei: %ld",
- (long)frames_written);
- }
- } else {
- samples_written = frames_written * 2;
- next_timestamp += samples_written;
- if(ge(next_timestamp, packet_end))
- drop_first_packet();
- infilling = 0;
- }
- } else {
- /* We don't have anything to play! We'd better play some 0s. */
- static const uint16_t zeros[INFILL_SAMPLES];
- size_t samples_available = INFILL_SAMPLES, frames_available;
-
- /* If the maximum infill would take us past the start of the next
- * packet then we truncate the infill to the right amount. */
- if(lt(packets->timestamp,
- next_timestamp + samples_available))
- samples_available = packets->timestamp - next_timestamp;
- if((int)samples_available < 0) {
- info("packets->timestamp: %"PRIx32" next_timestamp: %"PRIx32" next+max: %"PRIx32" available: %"PRIx32,
- packets->timestamp, next_timestamp,
- next_timestamp + INFILL_SAMPLES, samples_available);
- }
- frames_available = samples_available / 2;
- if(!infilling) {
- info("Infilling %d samples, next=%"PRIx32" packet=[%"PRIx32",%"PRIx32"]",
- samples_available, next_timestamp,
- packets->timestamp, packets->timestamp + packets->nsamples);
- //infilling++;
- }
- frames_written = snd_pcm_writei(pcm,
- zeros,
- frames_available);
- if(frames_written < 0) {
- switch(frames_written) {
- case -EAGAIN:
- info("snd_pcm_wait() returned but we got -EAGAIN!");
- break;
- case -EPIPE:
- error(0, "error calling snd_pcm_writei: %ld",
- (long)frames_written);
- escape = 1;
- break;
- default:
- fatal(0, "error calling snd_pcm_writei: %ld",
- (long)frames_written);
- }
- } else {
- samples_written = frames_written * 2;
- next_timestamp += samples_written;
- }
- }
- }
- active = 0;
- /* We stop playing for a bit until the buffer re-fills */
- pthread_mutex_unlock(&lock);
- if((err = snd_pcm_nonblock(pcm, 0)))
- fatal(0, "error calling snd_pcm_nonblock: %d", err);
- if(escape) {
- if((err = snd_pcm_drop(pcm)))
- fatal(0, "error calling snd_pcm_drop: %d", err);
- escape = 0;
- } else
- if((err = snd_pcm_drain(pcm)))
- fatal(0, "error calling snd_pcm_drain: %d", err);
- if((err = snd_pcm_nonblock(pcm, 1)))
- fatal(0, "error calling snd_pcm_nonblock: %d", err);
- prepared = 0;
- pthread_mutex_lock(&lock);
- }
-
- }
-#elif HAVE_COREAUDIO_AUDIOHARDWARE_H
- {
- OSStatus status;
- UInt32 propertySize;
- AudioDeviceID adid;
- AudioStreamBasicDescription asbd;
-
- /* If this looks suspiciously like libao's macosx driver there's an
- * excellent reason for that... */
-
- /* TODO report errors as strings not numbers */
- propertySize = sizeof adid;
- status = AudioHardwareGetProperty(kAudioHardwarePropertyDefaultOutputDevice,
- &propertySize, &adid);
- if(status)
- fatal(0, "AudioHardwareGetProperty: %d", (int)status);
- if(adid == kAudioDeviceUnknown)
- fatal(0, "no output device");
- propertySize = sizeof asbd;
- status = AudioDeviceGetProperty(adid, 0, false,
- kAudioDevicePropertyStreamFormat,
- &propertySize, &asbd);
- if(status)
- fatal(0, "AudioHardwareGetProperty: %d", (int)status);
- D(("mSampleRate %f", asbd.mSampleRate));
- D(("mFormatID %08lx", asbd.mFormatID));
- D(("mFormatFlags %08lx", asbd.mFormatFlags));
- D(("mBytesPerPacket %08lx", asbd.mBytesPerPacket));
- D(("mFramesPerPacket %08lx", asbd.mFramesPerPacket));
- D(("mBytesPerFrame %08lx", asbd.mBytesPerFrame));
- D(("mChannelsPerFrame %08lx", asbd.mChannelsPerFrame));
- D(("mBitsPerChannel %08lx", asbd.mBitsPerChannel));
- D(("mReserved %08lx", asbd.mReserved));
- if(asbd.mFormatID != kAudioFormatLinearPCM)
- fatal(0, "audio device does not support kAudioFormatLinearPCM");
- status = AudioDeviceAddIOProc(adid, adioproc, 0);
- if(status)
- fatal(0, "AudioDeviceAddIOProc: %d", (int)status);
- pthread_mutex_lock(&lock);
- for(;;) {
- /* Wait for the buffer to fill up a bit */
- info("Buffering...");
- while(nsamples < readahead)
- pthread_cond_wait(&cond, &lock);
- /* Start playing now */
- info("Playing...");
- next_timestamp = packets[sequence]->timestamp;
- active = 1;
- status = AudioDeviceStart(adid, adioproc);
- if(status)
- fatal(0, "AudioDeviceStart: %d", (int)status);
- /* Wait until the buffer empties out */
- while(nsamples >= minbuffer)
- pthread_cond_wait(&cond, &lock);
- /* Stop playing for a bit until the buffer re-fills */
- status = AudioDeviceStop(adid, adioproc);
- if(status)
- fatal(0, "AudioDeviceStop: %d", (int)status);
- active = 0;
- /* Go back round */
- }
- }
-#else
-# error No known audio API
-#endif
+ /* We have a second thread to add received packets to the queue */
+ pthread_create(<id, 0, queue_thread, 0);
+ /* The rest of the work is backend-specific */
+ backend();
}
/* display usage message and terminate */
" --min, -m FRAMES Buffer low water mark\n"
" --buffer, -b FRAMES Buffer high water mark\n"
" --max, -x FRAMES Buffer maximum size\n"
+ " --rcvbuf, -R BYTES Socket receive buffer size\n"
+ " --multicast, -M GROUP Join multicast group\n"
+#if HAVE_ALSA_ASOUNDLIB_H
+ " --alsa, -a Use ALSA to play audio\n"
+#endif
+#if HAVE_SYS_SOUNDCARD_H
+ " --oss, -o Use OSS to play audio\n"
+#endif
+#if HAVE_COREAUDIO_AUDIOHARDWARE_H
+ " --core-audio, -c Use Core Audio to play audio\n"
+#endif
" --help, -h Display usage message\n"
" --version, -V Display version number\n"
);
struct addrinfo *res;
struct stringlist sl;
char *sockname;
+ int rcvbuf, target_rcvbuf = 131072;
+ socklen_t len;
+ char *multicast_group = 0;
+ struct ip_mreq mreq;
+ struct ipv6_mreq mreq6;
static const struct addrinfo prefs = {
AI_PASSIVE,
mem_init();
if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale");
- while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:", options, 0)) >= 0) {
+ while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aoc", options, 0)) >= 0) {
switch(n) {
case 'h': help();
case 'V': version();
case 'b': readahead = 2 * atol(optarg); break;
case 'x': maxbuffer = 2 * atol(optarg); break;
case 'L': logfp = fopen(optarg, "w"); break;
+ case 'R': target_rcvbuf = atoi(optarg); break;
+ case 'M': multicast_group = optarg; break;
+#if HAVE_ALSA_ASOUNDLIB_H
+ case 'a': backend = playrtp_alsa; break;
+#endif
+#if HAVE_SYS_SOUNDCARD_H
+ case 'o': backend = playrtp_oss; break;
+#endif
+#if HAVE_COREAUDIO_AUDIOHARDWARE_H
+ case 'c': backend = playrtp_coreaudio; break;
+#endif
default: fatal(0, "invalid option");
}
}
fatal(errno, "error creating socket");
if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
fatal(errno, "error binding socket to %s", sockname);
+ if(multicast_group) {
+ if((n = getaddrinfo(multicast_group, 0, &prefs, &res)))
+ fatal(0, "getaddrinfo %s: %s", multicast_group, gai_strerror(n));
+ switch(res->ai_family) {
+ case PF_INET:
+ mreq.imr_multiaddr = ((struct sockaddr_in *)res->ai_addr)->sin_addr;
+ mreq.imr_interface.s_addr = 0; /* use primary interface */
+ if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
+ &mreq, sizeof mreq) < 0)
+ fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
+ break;
+ case PF_INET6:
+ mreq6.ipv6mr_multiaddr = ((struct sockaddr_in6 *)res->ai_addr)->sin6_addr;
+ memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
+ if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
+ &mreq6, sizeof mreq6) < 0)
+ fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
+ break;
+ default:
+ fatal(0, "unsupported address family %d", res->ai_family);
+ }
+ }
+ len = sizeof rcvbuf;
+ if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0)
+ fatal(errno, "error calling getsockopt SO_RCVBUF");
+ if(target_rcvbuf > rcvbuf) {
+ if(setsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF,
+ &target_rcvbuf, sizeof target_rcvbuf) < 0)
+ error(errno, "error calling setsockopt SO_RCVBUF %d",
+ target_rcvbuf);
+ /* We try to carry on anyway */
+ else
+ info("changed socket receive buffer from %d to %d",
+ rcvbuf, target_rcvbuf);
+ } else
+ info("default socket receive buffer %d", rcvbuf);
+ if(logfp)
+ info("WARNING: -L option can impact performance");
play_rtp();
return 0;
}
~mdw / disorder / blobdiff