From: Richard Kettlewell Date: Sat, 11 Apr 2009 19:59:14 +0000 (+0100) Subject: Merge audio timing fix branch. X-Git-Tag: 5.0~145 X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/commitdiff_plain/e99a9fd614f9bd1d205e00b47e3e669b7735b91f?hp=9d7a61297cb7263160d85a12d445fc1897fbfa80 Merge audio timing fix branch. --- diff --git a/.bzrignore b/.bzrignore index ec15e8e..3b13be6 100644 --- a/.bzrignore +++ b/.bzrignore @@ -202,3 +202,4 @@ doc/disorder-choose.8 doc/disorder-choose.8.html config.aux/compile server/endian +clients/rtpmon diff --git a/clients/Makefile.am b/clients/Makefile.am index 1ec55a9..b1399b2 100644 --- a/clients/Makefile.am +++ b/clients/Makefile.am @@ -17,7 +17,7 @@ # bin_PROGRAMS=disorder disorderfm disorder-playrtp -noinst_PROGRAMS=filename-bytes +noinst_PROGRAMS=filename-bytes rtpmon noinst_SCRIPTS=dump2wav AM_CPPFLAGS=-I${top_srcdir}/lib -I../lib @@ -39,6 +39,9 @@ disorder_playrtp_LDADD=$(LIBOBJS) ../lib/libdisorder.a \ $(LIBDB) $(LIBPTHREAD) disorder_playrtp_DEPENDENCIES=$(LIBOBJS) ../lib/libdisorder.a +rtpmon_SOURCES=rtpmon.c +rtpmon_LDADD=$(LIBOBJS) ../lib/libdisorder.a + filename_bytes_SOURCES=filename-bytes.c install-exec-hook: diff --git a/clients/playrtp.c b/clients/playrtp.c index f566409..f5f538d 100644 --- a/clients/playrtp.c +++ b/clients/playrtp.c @@ -81,8 +81,6 @@ #include "version.h" #include "uaudio.h" -#define readahead linux_headers_are_borked - /** @brief Obsolete synonym */ #ifndef IPV6_JOIN_GROUP # define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP @@ -96,19 +94,13 @@ static FILE *logfp; /** @brief Output device */ -/** @brief Minimum low watermark - * - * We'll stop playing if there's only this many samples in the buffer. */ -unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */ +/** @brief Buffer low watermark in samples */ +unsigned minbuffer = 4 * (2 * 44100) / 10; /* 0.4 seconds */ -/** @brief Buffer high watermark +/** @brief Maximum buffer size in samples * - * We'll only start playing when this many samples are available. */ -static unsigned readahead = 44100; /* 0.5 seconds */ - -/** @brief Maximum buffer size - * - * We'll stop reading from the network if we have this many samples. */ + * We'll stop reading from the network if we have this many samples. + */ static unsigned maxbuffer; /** @brief Received packets @@ -204,7 +196,6 @@ static const struct option options[] = { { "device", required_argument, 0, 'D' }, { "min", required_argument, 0, 'm' }, { "max", required_argument, 0, 'x' }, - { "buffer", required_argument, 0, 'b' }, { "rcvbuf", required_argument, 0, 'R' }, #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST { "oss", no_argument, 0, 'o' }, @@ -440,12 +431,18 @@ static void *listen_thread(void attribute((unused)) *arg) { * Must be called with @ref lock held. */ void playrtp_fill_buffer(void) { - while(nsamples) + /* Discard current buffer contents */ + while(nsamples) { + //fprintf(stderr, "%8u/%u (%u) DROPPING\n", nsamples, maxbuffer, minbuffer); drop_first_packet(); + } info("Buffering..."); - while(nsamples < readahead) { + /* Wait until there's at least minbuffer samples available */ + while(nsamples < minbuffer) { + //fprintf(stderr, "%8u/%u (%u) FILLING\n", nsamples, maxbuffer, minbuffer); pthread_cond_wait(&cond, &lock); } + /* Start from whatever is earliest */ next_timestamp = pheap_first(&packets)->timestamp; active = 1; } @@ -480,7 +477,6 @@ static void help(void) { "Options:\n" " --device, -D DEVICE Output device\n" " --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" " --config, -C PATH Set configuration file\n" @@ -537,8 +533,6 @@ static size_t playrtp_callback(void *buffer, *bufptr++ = (int16_t)ntohs(*ptr++); --i; } - /* We don't junk the packet here; a subsequent call to - * playrtp_next_packet() will dispose of it (if it's actually done with). */ } else { /* There is no suitable packet. We introduce 0s up to the next packet, or * to fill the buffer if there's no next packet or that's too many. The @@ -559,6 +553,8 @@ static size_t playrtp_callback(void *buffer, } /* Advance timestamp */ next_timestamp += samples; + /* Junk obsolete packets */ + playrtp_next_packet(); pthread_mutex_unlock(&lock); return samples; } @@ -568,7 +564,7 @@ int main(int argc, char **argv) { struct addrinfo *res; struct stringlist sl; char *sockname; - int rcvbuf, target_rcvbuf = 131072; + int rcvbuf, target_rcvbuf = 0; socklen_t len; struct ip_mreq mreq; struct ipv6_mreq mreq6; @@ -598,14 +594,13 @@ int main(int argc, char **argv) { mem_init(); if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale"); backend = uaudio_apis[0]; - while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:re:P:", options, 0)) >= 0) { + while((n = getopt_long(argc, argv, "hVdD:m:x:L:R:M:aocC:re:P:", options, 0)) >= 0) { switch(n) { case 'h': help(); case 'V': version("disorder-playrtp"); case 'd': debugging = 1; break; case 'D': uaudio_set("device", optarg); break; case 'm': minbuffer = 2 * atol(optarg); break; - 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; @@ -628,7 +623,7 @@ int main(int argc, char **argv) { } if(config_read(0)) fatal(0, "cannot read configuration"); if(!maxbuffer) - maxbuffer = 4 * readahead; + maxbuffer = 2 * minbuffer; argc -= optind; argv += optind; switch(argc) { @@ -743,6 +738,7 @@ int main(int argc, char **argv) { rcvbuf, target_rcvbuf); } else info("default socket receive buffer %d", rcvbuf); + //info("minbuffer %u maxbuffer %u", minbuffer, maxbuffer); if(logfp) info("WARNING: -L option can impact performance"); if(control_socket) { @@ -794,12 +790,28 @@ int main(int argc, char **argv) { pthread_mutex_unlock(&lock); backend->activate(); pthread_mutex_lock(&lock); - /* Wait until the buffer empties out */ + /* Wait until the buffer empties out + * + * If there's a packet that we can play right now then we definitely + * continue. + * + * Also if there's at least minbuffer samples we carry on regardless and + * insert silence. The assumption is there's been a pause but more data + * is now available. + */ while(nsamples >= minbuffer || (nsamples > 0 && contains(pheap_first(&packets), next_timestamp))) { + //fprintf(stderr, "%8u/%u (%u) PLAYING\n", nsamples, maxbuffer, minbuffer); pthread_cond_wait(&cond, &lock); } +#if 0 + if(nsamples) { + struct packet *p = pheap_first(&packets); + fprintf(stderr, "nsamples=%u (%u) next_timestamp=%"PRIx32", first packet is [%"PRIx32",%"PRIx32")\n", + nsamples, minbuffer, next_timestamp,p->timestamp,p->timestamp+p->nsamples); + } +#endif /* Stop playing for a bit until the buffer re-fills */ pthread_mutex_unlock(&lock); backend->deactivate(); diff --git a/clients/rtpmon.c b/clients/rtpmon.c new file mode 100644 index 0000000..e64b08e --- /dev/null +++ b/clients/rtpmon.c @@ -0,0 +1,295 @@ +/* + * This file is part of DisOrder. + * Copyright (C) 2009 Richard Kettlewell + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +/** @file clients/rtpmon.c + * @brief RTP monitor + * + * This progam monitors the rate at which data arrives by RTP and + * constantly display it. It is intended for debugging only. + * + * TODO de-dupe with playrtp. + */ +#include "common.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "syscalls.h" +#include "timeval.h" +#include "mem.h" +#include "log.h" +#include "version.h" +#include "addr.h" +#include "configuration.h" +#include "rtp.h" + +/** @brief Record of one packet */ +struct entry { + /** @brief When packet arrived */ + struct timeval when; + + /** @brief Serial number of first sample */ + uint32_t serial; +}; + +/** @brief Bytes per frame */ +static unsigned bpf = 4; + +/** @brief Frame serial number */ +static uint32_t serial; + +/** @brief Size of ring buffer */ +#define RINGSIZE 131072 + +/** @brief Ring buffer */ +static struct entry ring[RINGSIZE]; + +/** @brief Where new packets join the ring */ +static unsigned ringtail; + +static const struct option options[] = { + { "help", no_argument, 0, 'h' }, + { "version", no_argument, 0, 'V' }, + { "bpf", required_argument, 0, 'b' }, + { 0, 0, 0, 0 } +}; + +static void help(void) { + xprintf("Usage:\n" + " rtpmon [OPTIONS] [ADDRESS] PORT\n" + "Options:\n" + " --bpf, -b Bytes/frame (default 4)\n" + " --help, -h Display usage message\n" + " --version, -V Display version number\n" + ); + xfclose(stdout); + exit(0); +} + +/** @brief Compute the rate by sampling at two points in the ring buffer */ +static double rate(unsigned earlier, unsigned later) { + const uint32_t frames = ring[later].serial - ring[earlier].serial; + const int64_t us = tvsub_us(ring[later].when, ring[earlier].when); + + if(us) + return 1000000.0 * frames / us; + else + return 0.0; +} + +/** @brief Called to say we received some bytes + * @param when When we received them + * @param n How many frames of audio data we received + */ +static void frames(const struct timeval *when, size_t n) { + const time_t prev = ring[(ringtail - 1) % RINGSIZE].when.tv_sec; + + ring[ringtail].when = *when; + ring[ringtail].serial = serial; + serial += n; + ringtail = (ringtail + 1) % RINGSIZE; + // Report once a second + if(prev != when->tv_sec) { + if(printf("%8.2f %8.2f %8.2f %8.2f %8.2f %8.2f %8.2f\n", + rate((ringtail - RINGSIZE / 128) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 64) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 32) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 16) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 8) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 4) % RINGSIZE, + (ringtail - 1) % RINGSIZE), + rate((ringtail - RINGSIZE / 2) % RINGSIZE, + (ringtail - 1) % RINGSIZE)) < 0 + || fflush(stdout) < 0) + fatal(errno, "stdout"); + } +} + +int main(int argc, char **argv) { + int n; + struct addrinfo *res; + struct stringlist sl; + struct ip_mreq mreq; + struct ipv6_mreq mreq6; + int rtpfd; + char *sockname; + int is_multicast; + union any_sockaddr { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; + }; + union any_sockaddr mgroup; + + static const struct addrinfo prefs = { + .ai_flags = AI_PASSIVE, + .ai_family = PF_INET, + .ai_socktype = SOCK_DGRAM, + .ai_protocol = IPPROTO_UDP + }; + static const int one = 1; + + mem_init(); + if(!setlocale(LC_CTYPE, "")) + fatal(errno, "error calling setlocale"); + while((n = getopt_long(argc, argv, "hVb:", options, 0)) >= 0) { + switch(n) { + case 'h': help(); + case 'V': version("rtpmon"); + case 'b': bpf = atoi(optarg); break; + default: fatal(0, "invalid option"); + } + } + argc -= optind; + argv += optind; + switch(argc) { + case 1: + case 2: + /* Use command-line ADDRESS+PORT or just PORT */ + sl.n = argc; + sl.s = argv; + break; + default: + fatal(0, "usage: rtpmon [OPTIONS] [ADDRESS] PORT"); + } + if(!(res = get_address(&sl, &prefs, &sockname))) + exit(1); + /* Create the socket */ + if((rtpfd = socket(res->ai_family, + res->ai_socktype, + res->ai_protocol)) < 0) + fatal(errno, "error creating socket"); + /* Allow multiple listeners */ + xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); + is_multicast = multicast(res->ai_addr); + /* The multicast and unicast/broadcast cases are different enough that they + * are totally split. Trying to find commonality between them causes more + * trouble that it's worth. */ + if(is_multicast) { + /* Stash the multicast group address */ + memcpy(&mgroup, res->ai_addr, res->ai_addrlen); + switch(res->ai_addr->sa_family) { + case AF_INET: + mgroup.in.sin_port = 0; + break; + case AF_INET6: + mgroup.in6.sin6_port = 0; + break; + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); + } + /* Bind to to the multicast group address */ + if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) + fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr)); + /* Add multicast group membership */ + switch(mgroup.sa.sa_family) { + case PF_INET: + mreq.imr_multiaddr = mgroup.in.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 = mgroup.in6.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); + } + /* Report what we did */ + info("listening on %s multicast group %s", + format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa)); + } else { + /* Bind to 0/port */ + switch(res->ai_addr->sa_family) { + case AF_INET: { + struct sockaddr_in *in = (struct sockaddr_in *)res->ai_addr; + + memset(&in->sin_addr, 0, sizeof (struct in_addr)); + if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) + fatal(errno, "error binding socket to 0.0.0.0 port %d", + ntohs(in->sin_port)); + break; + } + case AF_INET6: { + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)res->ai_addr; + + memset(&in6->sin6_addr, 0, sizeof (struct in6_addr)); + break; + } + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); + } + if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) + fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr)); + /* Report what we did */ + info("listening on %s", format_sockaddr(res->ai_addr)); + } + for(;;) { + struct rtp_header header; + char buffer[4096]; + struct iovec iov[2]; + struct timeval when; + + iov[0].iov_base = &header; + iov[0].iov_len = sizeof header; + iov[1].iov_base = buffer; + iov[1].iov_len = sizeof buffer; + n = readv(rtpfd, iov, 2); + gettimeofday(&when, 0); + if(n < 0) { + switch(errno) { + case EINTR: + continue; + default: + fatal(errno, "error reading from socket"); + } + } + if((size_t)n <= sizeof (struct rtp_header)) { + info("ignored a short packet"); + continue; + } + frames(&when, (n - sizeof header) / bpf); + } +} + +/* +Local Variables: +c-basic-offset:2 +comment-column:40 +fill-column:79 +indent-tabs-mode:nil +End: +*/ diff --git a/doc/disorder-playrtp.1.in b/doc/disorder-playrtp.1.in index c313ece..480a935 100644 --- a/doc/disorder-playrtp.1.in +++ b/doc/disorder-playrtp.1.in @@ -98,25 +98,21 @@ Display a usage message. Display version number. .SS "Buffer Control Options" You shouldn't need to use these options. +You should consult the source code for details of their effects. .TP .B \-\-min \fIFRAMES\fR, \fB\-m \fIFRAMES\fR Specifies the buffer low watermark in frames. -If the number of frames falls below this value then playing will be -stopped until the buffer fills up. -.TP -.B \-\-buffer \fIFRAMES\fR, \fB\-b \fIFRAMES\fR -Specifies the buffer high watermark in frames. -Once there are this many frames in the buffer, playing will be (re-)started. .TP .B \-\-max \fIFRAMES\fR, \fB\-x \fIFRAMES\fR Specifies the maximum buffer size in frames. If there are this many frames in the buffer then reading from the network socket will be suspended. -The default is four times the \fB\-\-buffer\fR value. +The default is twice the \fB\-\-min\fR value. .TP .B \-\-rcvbuf \fIBYTES\fR, \fB\-R \fIBYTES\fR Specifies socket receive buffer size. -The default is 131072 (128Kbytes). +The default is not to change the buffer size, i.e. you get whatever the +local operating system chooses. The buffer size will not be reduced below the operating system's default. .SH "REMOTE CONTROL" The diff --git a/lib/uaudio-alsa.c b/lib/uaudio-alsa.c index 721639c..3cec315 100644 --- a/lib/uaudio-alsa.c +++ b/lib/uaudio-alsa.c @@ -57,7 +57,24 @@ static long alsa_mixer_min; static long alsa_mixer_max; /** @brief Actually play sound via ALSA */ -static size_t alsa_play(void *buffer, size_t samples) { +static size_t alsa_play(void *buffer, size_t samples, unsigned flags) { + /* If we're paused we just pretend. We rely on snd_pcm_writei() blocking so + * we have to fake up a sleep here. However it doesn't have to be all that + * accurate - in particular it's quite acceptable to greatly underestimate + * the required wait time. For 'lengthy' waits we do this by the blunt + * instrument of halving it. */ + if(flags & UAUDIO_PAUSED) { + if(samples > 64) + samples /= 2; + const uint64_t ns = ((uint64_t)samples * 1000000000 + / (uaudio_rate * uaudio_channels)); + struct timespec ts[1]; + ts->tv_sec = ns / 1000000000; + ts->tv_nsec = ns % 1000000000; + while(nanosleep(ts, ts) < 0 && errno == EINTR) + ; + return samples; + } int err; /* ALSA wants 'frames', where frame = several concurrently played samples */ const snd_pcm_uframes_t frames = samples / uaudio_channels; @@ -117,14 +134,6 @@ static void alsa_open(void) { } -static void alsa_activate(void) { - uaudio_thread_activate(); -} - -static void alsa_deactivate(void) { - uaudio_thread_deactivate(); -} - static void alsa_start(uaudio_callback *callback, void *userdata) { if(uaudio_channels != 1 && uaudio_channels != 2) @@ -258,8 +267,8 @@ const struct uaudio uaudio_alsa = { .options = alsa_options, .start = alsa_start, .stop = alsa_stop, - .activate = alsa_activate, - .deactivate = alsa_deactivate, + .activate = uaudio_thread_activate, + .deactivate = uaudio_thread_deactivate, .open_mixer = alsa_open_mixer, .close_mixer = alsa_close_mixer, .get_volume = alsa_get_volume, diff --git a/lib/uaudio-command.c b/lib/uaudio-command.c index 2c610ff..57e8762 100644 --- a/lib/uaudio-command.c +++ b/lib/uaudio-command.c @@ -42,6 +42,9 @@ static int command_fd; /** @brief Child process ID */ static pid_t command_pid; +/** @brief Whether to suspend on pause */ +static int command_suspend_on_pause; + static const char *const command_options[] = { "command", "pause-mode", @@ -92,8 +95,12 @@ static void command_open(void) { } /** @brief Send audio data to subprocess */ -static size_t command_play(void *buffer, size_t nsamples) { - uaudio_schedule_synchronize(); +static size_t command_play(void *buffer, size_t nsamples, unsigned flags) { + uaudio_schedule_sync(); + /* If we're pausing and want that to be represented by stopping writing, we + * just pretend */ + if((flags & UAUDIO_PAUSED) && command_suspend_on_pause) + return nsamples; const size_t bytes = nsamples * uaudio_sample_size; int written = write(command_fd, buffer, bytes); if(written < 0) { @@ -109,8 +116,11 @@ static size_t command_play(void *buffer, size_t nsamples) { fatal(errno, "error writing to audio command subprocess"); } } + /* TODO what if we write a partial sample? Actually reasonably unlikely but + * not impossible. Maybe someone who actually uses this backend should sort + * it out. */ const size_t written_samples = written / uaudio_sample_size; - uaudio_schedule_update(written_samples); + uaudio_schedule_sent(written_samples); return written_samples; } @@ -120,9 +130,9 @@ static void command_start(uaudio_callback *callback, unsigned flags = 0; if(!strcmp(pausemode, "silence")) - flags |= UAUDIO_THREAD_FAKE_PAUSE; + command_suspend_on_pause = 0; else if(!strcmp(pausemode, "suspend")) - ; + command_suspend_on_pause = 1; else fatal(0, "unknown pause mode '%s'", pausemode); command_open(); @@ -140,15 +150,6 @@ static void command_stop(void) { command_wait(); } -static void command_activate(void) { - uaudio_schedule_reactivated = 1; - uaudio_thread_activate(); -} - -static void command_deactivate(void) { - uaudio_thread_deactivate(); -} - static void command_configure(void) { uaudio_set("command", config->speaker_command); uaudio_set("pause-mode", config->pause_mode); @@ -159,8 +160,8 @@ const struct uaudio uaudio_command = { .options = command_options, .start = command_start, .stop = command_stop, - .activate = command_activate, - .deactivate = command_deactivate, + .activate = uaudio_thread_activate, + .deactivate = uaudio_thread_deactivate, .configure = command_configure, }; diff --git a/lib/uaudio-oss.c b/lib/uaudio-oss.c index 6d902cb..6ebb6df 100644 --- a/lib/uaudio-oss.c +++ b/lib/uaudio-oss.c @@ -28,6 +28,7 @@ #include #include #include +#include #include "mem.h" #include "log.h" @@ -61,15 +62,6 @@ static const char *const oss_options[] = { NULL }; -/** @brief Actually play sound via OSS */ -static size_t oss_play(void *buffer, size_t samples) { - const size_t bytes = samples * uaudio_sample_size; - int rc = write(oss_fd, buffer, bytes); - if(rc < 0) - fatal(errno, "error writing to sound device"); - return rc / uaudio_sample_size; -} - /** @brief Open the OSS sound device */ static void oss_open(void) { const char *device = uaudio_get("device", NULL); @@ -105,17 +97,40 @@ static void oss_open(void) { #endif } -static void oss_activate(void) { - oss_open(); - uaudio_thread_activate(); +/** @brief Close the OSS sound device */ +static void oss_close(void) { + if(oss_fd != -1) { + close(oss_fd); + oss_fd = -1; + } } -static void oss_deactivate(void) { - uaudio_thread_deactivate(); - close(oss_fd); - oss_fd = -1; +/** @brief Actually play sound via OSS */ +static size_t oss_play(void *buffer, size_t samples, unsigned flags) { + /* cf uaudio-alsa.c:alsa-play() */ + if(flags & UAUDIO_PAUSED) { + if(flags & UAUDIO_PAUSE) + oss_close(); + if(samples > 64) + samples /= 2; + const uint64_t ns = ((uint64_t)samples * 1000000000 + / (uaudio_rate * uaudio_channels)); + struct timespec ts[1]; + ts->tv_sec = ns / 1000000000; + ts->tv_nsec = ns % 1000000000; + while(nanosleep(ts, ts) < 0 && errno == EINTR) + ; + return samples; + } + if(flags & UAUDIO_RESUME) + oss_open(); + const size_t bytes = samples * uaudio_sample_size; + int rc = write(oss_fd, buffer, bytes); + if(rc < 0) + fatal(errno, "error writing to sound device"); + return rc / uaudio_sample_size; } - + static void oss_start(uaudio_callback *callback, void *userdata) { if(uaudio_channels != 1 && uaudio_channels != 2) @@ -142,6 +157,7 @@ static void oss_start(uaudio_callback *callback, static void oss_stop(void) { uaudio_thread_stop(); + oss_close(); /* might not have been paused */ } /** @brief Channel names */ @@ -209,8 +225,8 @@ const struct uaudio uaudio_oss = { .options = oss_options, .start = oss_start, .stop = oss_stop, - .activate = oss_activate, - .deactivate = oss_deactivate, + .activate = uaudio_thread_activate, + .deactivate = uaudio_thread_deactivate, .open_mixer = oss_open_mixer, .close_mixer = oss_close_mixer, .get_volume = oss_get_volume, diff --git a/lib/uaudio-rtp.c b/lib/uaudio-rtp.c index bb03be4..e1768a8 100644 --- a/lib/uaudio-rtp.c +++ b/lib/uaudio-rtp.c @@ -59,6 +59,9 @@ static int rtp_fd; /** @brief RTP SSRC */ static uint32_t rtp_id; +/** @brief Base for timestamp */ +static uint32_t rtp_base; + /** @brief RTP sequence number */ static uint16_t rtp_sequence; @@ -68,11 +71,8 @@ static uint16_t rtp_sequence; */ static int rtp_errors; -/** @brief Delay threshold in microseconds - * - * rtp_play() never attempts to introduce a delay shorter than this. - */ -static int64_t rtp_delay_threshold; +/** @brief Set while paused */ +static volatile int rtp_paused; static const char *const rtp_options[] = { "rtp-destination", @@ -81,7 +81,6 @@ static const char *const rtp_options[] = { "rtp-source-port", "multicast-ttl", "multicast-loop", - "delay-threshold", NULL }; @@ -129,16 +128,28 @@ static void rtp_set_netconfig(const char *af, } } -static size_t rtp_play(void *buffer, size_t nsamples) { +static size_t rtp_play(void *buffer, size_t nsamples, unsigned flags) { struct rtp_header header; struct iovec vec[2]; - + +#if 0 + if(flags & (UAUDIO_PAUSE|UAUDIO_RESUME)) + fprintf(stderr, "rtp_play %zu samples%s%s%s%s\n", nsamples, + flags & UAUDIO_PAUSE ? " UAUDIO_PAUSE" : "", + flags & UAUDIO_RESUME ? " UAUDIO_RESUME" : "", + flags & UAUDIO_PLAYING ? " UAUDIO_PLAYING" : "", + flags & UAUDIO_PAUSED ? " UAUDIO_PAUSED" : ""); +#endif + /* We do as much work as possible before checking what time it is */ /* Fill out header */ header.vpxcc = 2 << 6; /* V=2, P=0, X=0, CC=0 */ header.seq = htons(rtp_sequence++); header.ssrc = rtp_id; - header.mpt = (uaudio_schedule_reactivated ? 0x80 : 0x00) | rtp_payload; + header.mpt = rtp_payload; + /* If we've come out of a pause, set the marker bit */ + if(flags & UAUDIO_RESUME) + header.mpt |= 0x80; #if !WORDS_BIGENDIAN /* Convert samples to network byte order */ uint16_t *u = buffer, *const limit = u + nsamples; @@ -151,8 +162,13 @@ static size_t rtp_play(void *buffer, size_t nsamples) { vec[0].iov_len = sizeof header; vec[1].iov_base = buffer; vec[1].iov_len = nsamples * uaudio_sample_size; - uaudio_schedule_synchronize(); - header.timestamp = htonl((uint32_t)uaudio_schedule_timestamp); + const uint32_t timestamp = uaudio_schedule_sync(); + header.timestamp = htonl(rtp_base + (uint32_t)timestamp); + /* If we're paused don't actually end a packet, we just pretend */ + if(flags & UAUDIO_PAUSED) { + uaudio_schedule_sent(nsamples); + return nsamples; + } int written_bytes; do { written_bytes = writev(rtp_fd, vec, 2); @@ -165,10 +181,10 @@ static size_t rtp_play(void *buffer, size_t nsamples) { return 0; } else rtp_errors /= 2; /* gradual decay */ - written_bytes -= sizeof (struct rtp_header); - const size_t written_samples = written_bytes / uaudio_sample_size; - uaudio_schedule_update(written_samples); - return written_samples; + /* TODO what can we sensibly do about short writes here? Really that's just + * an error and we ought to be using smaller packets. */ + uaudio_schedule_sent(nsamples); + return nsamples; } static void rtp_open(void) { @@ -187,7 +203,6 @@ static void rtp_open(void) { "rtp-source", "rtp-source-port", src); - rtp_delay_threshold = atoi(uaudio_get("rtp-delay-threshold", "1000")); /* ...microseconds */ /* Resolve addresses */ @@ -299,6 +314,7 @@ static void rtp_start(uaudio_callback *callback, * packet contents are highly public so there's no point asking for very * strong randomness. */ gcry_create_nonce(&rtp_id, sizeof rtp_id); + gcry_create_nonce(&rtp_base, sizeof rtp_base); gcry_create_nonce(&rtp_sequence, sizeof rtp_sequence); rtp_open(); uaudio_schedule_init(); @@ -317,15 +333,6 @@ static void rtp_stop(void) { rtp_fd = -1; } -static void rtp_activate(void) { - uaudio_schedule_reactivated = 1; - uaudio_thread_activate(); -} - -static void rtp_deactivate(void) { - uaudio_thread_deactivate(); -} - static void rtp_configure(void) { char buffer[64]; @@ -338,8 +345,6 @@ static void rtp_configure(void) { snprintf(buffer, sizeof buffer, "%ld", config->multicast_ttl); uaudio_set("multicast-ttl", buffer); uaudio_set("multicast-loop", config->multicast_loop ? "yes" : "no"); - snprintf(buffer, sizeof buffer, "%ld", config->rtp_delay_threshold); - uaudio_set("delay-threshold", buffer); } const struct uaudio uaudio_rtp = { @@ -347,8 +352,8 @@ const struct uaudio uaudio_rtp = { .options = rtp_options, .start = rtp_start, .stop = rtp_stop, - .activate = rtp_activate, - .deactivate = rtp_deactivate, + .activate = uaudio_thread_activate, + .deactivate = uaudio_thread_deactivate, .configure = rtp_configure, }; diff --git a/lib/uaudio-schedule.c b/lib/uaudio-schedule.c index 95fa0c0..95e4115 100644 --- a/lib/uaudio-schedule.c +++ b/lib/uaudio-schedule.c @@ -27,12 +27,24 @@ * * The sequence numbers are intended for RTP's use but it's more convenient to * maintain them here. + * + * The basic idea: + * - we maintain a base time + * - we calculate from this how many samples SHOULD have been sent by now + * - we compare this with the number of samples sent so far + * - we use this to wait until we're ready to send something + * - it's up to the caller to send nothing, or send 0s, if it's supposed to + * be paused + * + * An implication of this is that the caller must still call + * uaudio_schedule_sync() when deactivated (paused) and pretend to send 0s. */ #include "common.h" #include -#include +#include +#include #include "uaudio.h" #include "mem.h" @@ -50,113 +62,88 @@ * timestamp is maintained as a 64-bit integer, giving around six million years * before wrapping, and truncated to 32 bits when transmitting. */ -uint64_t uaudio_schedule_timestamp; - -/** @brief Actual time corresponding to @ref uaudio_schedule_timestamp - * - * This is the time, on this machine, at which the sample at @ref - * uaudio_schedule_timestamp ought to be sent, interpreted as the time the last - * packet was sent plus the time length of the packet. */ -static struct timeval uaudio_schedule_timeval; - -/** @brief Set when we (re-)activate, to provoke timestamp resync */ -int uaudio_schedule_reactivated; +static uint64_t timestamp; -/** @brief Delay threshold in microseconds +/** @brief Base time * - * uaudio_schedule_play() never attempts to introduce a delay shorter than this. + * This is the base time that corresponds to a timestamp of 0. */ -static int64_t uaudio_schedule_delay_threshold; - -/** @brief Time for current packet */ -static struct timeval uaudio_schedule_now; +struct timeval base; /** @brief Synchronize playback operations against real time + * @return Sample number * - * This function sleeps as necessary to rate-limit playback operations to match - * the actual playback rate. It also maintains @ref uaudio_schedule_timestamp - * as an arbitrarily-based sample counter, for use by RTP. - * - * You should call this in your API's @ref uaudio_playcallback before writing - * and call uaudio_schedule_update() afterwards. */ -void uaudio_schedule_synchronize(void) { -retry: - xgettimeofday(&uaudio_schedule_now, NULL); - if(uaudio_schedule_reactivated) { - /* We've been deactivated for some unknown interval. We need to advance - * rtp_timestamp to account for the dead air. */ - /* On the first run through we'll set the start time. */ - if(!uaudio_schedule_timeval.tv_sec) - uaudio_schedule_timeval = uaudio_schedule_now; - /* See how much time we missed. - * - * This will be 0 on the first run through, in which case we'll not modify - * anything. - * - * It'll be negative in the (rare) situation where the deactivation - * interval is shorter than the last packet we sent. In this case we wait - * for that much time and then return having sent no samples, which will - * cause uaudio_play_thread_fn() to retry. - * - * In the normal case it will be positive. - */ - const int64_t delay = tvsub_us(uaudio_schedule_now, - uaudio_schedule_timeval); /* microseconds */ - if(delay < 0) { - usleep(-delay); - goto retry; - } - /* Advance the RTP timestamp to the present. With 44.1KHz stereo this will - * overflow the intermediate value with a delay of a bit over 6 years. - * This seems acceptable. */ - uint64_t update = (delay * uaudio_rate * uaudio_channels) / 1000000; - /* Don't throw off channel synchronization */ - update -= update % uaudio_channels; - /* We log nontrivial changes */ - if(update) - info("advancing uaudio_schedule_timeval by %"PRIu64" samples", update); - uaudio_schedule_timestamp += update; - uaudio_schedule_timeval = uaudio_schedule_now; - uaudio_schedule_reactivated = 0; +uint32_t uaudio_schedule_sync(void) { + const unsigned rate = uaudio_rate * uaudio_channels; + struct timeval now; + + xgettimeofday(&now, NULL); + /* If we're just starting then we might as well send as much as possible + * straight away. */ + if(!base.tv_sec) { + base = now; + return timestamp; + } + /* Calculate how many microseconds ahead of the base time we are */ + uint64_t us = tvsub_us(now, base); + /* Calculate how many samples that is */ + uint64_t samples = us * rate / 1000000; + /* So... + * + * We've actually sent 'timestamp' samples so far. + * + * We OUGHT to have sent 'samples' samples so far. + * + * Suppose it's the SECOND call. timestamp will be (say) 716. 'samples' + * will be (say) 10 - there's been a bit of scheduling delay. So in that + * case we should wait for 716-10=706 samples worth of time before we can + * even send one sample. + * + * So we wait that long and send our 716 samples. + * + * On the next call we'll have timestamp=1432 and samples=726, say. So we + * wait and send again. + * + * On the next call there's been a bit of a delay. timestamp=2148 but + * samples=2200. So we send our 716 samples immediately. + * + * If the delay had been longer we might sent further packets back to back to + * make up for it. + * + * Now timestamp=2864 and samples=2210 (say). Now we're back to waiting. + */ + if(samples < timestamp) { + /* We should delay a bit */ + int64_t wait_samples = timestamp - samples; + int64_t wait_ns = wait_samples * 1000000000 / rate; + + struct timespec ts[1]; + ts->tv_sec = wait_ns / 1000000000; + ts->tv_nsec = wait_ns % 1000000000; +#if 0 + fprintf(stderr, + "samples=%8"PRIu64" timestamp=%8"PRIu64" wait=%"PRId64" (%"PRId64"ns)\n", + samples, timestamp, wait_samples, wait_ns); +#endif + while(nanosleep(ts, ts) < 0 && errno == EINTR) + ; } else { - /* Chances are we've been called right on the heels of the previous packet. - * If we just sent packets as fast as we got audio data we'd get way ahead - * of the player and some buffer somewhere would fill (or at least become - * unreasonably large). - * - * First find out how far ahead of the target time we are. - */ - const int64_t ahead = tvsub_us(uaudio_schedule_timeval, - uaudio_schedule_now); /* microseconds */ - /* Only delay at all if we are nontrivially ahead. */ - if(ahead > uaudio_schedule_delay_threshold) { - /* Don't delay by the full amount */ - usleep(ahead - uaudio_schedule_delay_threshold / 2); - /* Refetch time (so we don't get out of step with reality) */ - xgettimeofday(&uaudio_schedule_now, NULL); - } +#if 0 + fprintf(stderr, "samples=%8"PRIu64" timestamp=%8"PRIu64"\n", + samples, timestamp); +#endif } + /* If samples >= timestamp then it's time, or gone time, to play the + * timestamp'th sample. So we return immediately. */ + return timestamp; } -/** @brief Update schedule after writing - * - * Called by your API's @ref uaudio_playcallback after sending audio data (to a - * subprocess or network or whatever). A separate function so that the caller - * doesn't have to know how many samples they're going to write until they've - * done so. +/** @brief Report how many samples we actually sent + * @param nsamples Number of samples sent */ -void uaudio_schedule_update(size_t written_samples) { - /* uaudio_schedule_timestamp and uaudio_schedule_timestamp are supposed to - * refer to the first sample of the next packet */ - uaudio_schedule_timestamp += written_samples; - const unsigned usec = (uaudio_schedule_timeval.tv_usec - + 1000000 * written_samples / (uaudio_rate - * uaudio_channels)); - /* ...will only overflow 32 bits if one packet is more than about half an - * hour long, which is not plausible. */ - uaudio_schedule_timeval.tv_sec += usec / 1000000; - uaudio_schedule_timeval.tv_usec = usec % 1000000; +void uaudio_schedule_sent(size_t nsamples) { + timestamp += nsamples; } /** @brief Initialize audio scheduling @@ -164,10 +151,8 @@ void uaudio_schedule_update(size_t written_samples) { * Should be called from your API's @c start callback. */ void uaudio_schedule_init(void) { - gcry_create_nonce(&uaudio_schedule_timestamp, - sizeof uaudio_schedule_timestamp); /* uaudio_schedule_play() will spot this and choose an initial value */ - uaudio_schedule_timeval.tv_sec = 0; + base.tv_sec = 0; } /* diff --git a/lib/uaudio-thread.c b/lib/uaudio-thread.c index 30a3b8a..f04b84b 100644 --- a/lib/uaudio-thread.c +++ b/lib/uaudio-thread.c @@ -75,7 +75,6 @@ static uaudio_callback *uaudio_thread_collect_callback; static uaudio_playcallback *uaudio_thread_play_callback; static void *uaudio_thread_userdata; static int uaudio_thread_started; -static int uaudio_thread_activated; static int uaudio_thread_collecting; static pthread_mutex_t uaudio_thread_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t uaudio_thread_cond = PTHREAD_COND_INITIALIZER; @@ -86,6 +85,9 @@ static size_t uaudio_thread_min; /** @brief Maximum number of samples per chunk */ static size_t uaudio_thread_max; +/** @brief Set when activated, clear when paused */ +static int uaudio_thread_activated; + /** @brief Return number of buffers currently in use */ static int uaudio_buffers_used(void) { return (uaudio_collect_buffer - uaudio_play_buffer) % UAUDIO_THREAD_BUFFERS; @@ -107,8 +109,7 @@ static void *uaudio_collect_thread_fn(void attribute((unused)) *arg) { /* We are definitely active now */ uaudio_thread_collecting = 1; pthread_cond_broadcast(&uaudio_thread_cond); - while(uaudio_thread_activated - || (uaudio_thread_flags & UAUDIO_THREAD_FAKE_PAUSE)) { + while(uaudio_thread_activated) { if(uaudio_buffers_used() < UAUDIO_THREAD_BUFFERS - 1) { /* At least one buffer is available. We release the lock while * collecting data so that other already-filled buffers can be played @@ -127,9 +128,6 @@ static void *uaudio_collect_thread_fn(void attribute((unused)) *arg) { uaudio_thread_max - b->nsamples, uaudio_thread_userdata); } - } else if(uaudio_thread_flags & UAUDIO_THREAD_FAKE_PAUSE) { - memset(b->samples, 0, uaudio_thread_min * uaudio_sample_size); - b->nsamples += uaudio_thread_min; } pthread_mutex_lock(&uaudio_thread_lock); /* Advance to next buffer */ @@ -154,9 +152,23 @@ static void *uaudio_collect_thread_fn(void attribute((unused)) *arg) { */ static void *uaudio_play_thread_fn(void attribute((unused)) *arg) { int resync = 1; + unsigned last_flags = 0; + unsigned char zero[uaudio_thread_max * uaudio_sample_size]; + memset(zero, 0, sizeof zero); - pthread_mutex_lock(&uaudio_thread_lock); while(uaudio_thread_started) { + // If we're paused then just play silence + if(!uaudio_thread_activated) { + pthread_mutex_unlock(&uaudio_thread_lock); + unsigned flags = UAUDIO_PAUSED; + if(last_flags & UAUDIO_PLAYING) + flags |= UAUDIO_PAUSE; + uaudio_thread_play_callback(zero, uaudio_thread_max, + last_flags = flags); + /* We expect the play callback to block for a reasonable period */ + pthread_mutex_lock(&uaudio_thread_lock); + continue; + } const int used = uaudio_buffers_used(); int go; @@ -171,10 +183,15 @@ static void *uaudio_play_thread_fn(void attribute((unused)) *arg) { pthread_mutex_unlock(&uaudio_thread_lock); //fprintf(stderr, "P%d.", uaudio_play_buffer); size_t played = 0; - while(played < b->nsamples) + while(played < b->nsamples) { + unsigned flags = UAUDIO_PLAYING; + if(last_flags & UAUDIO_PAUSED) + flags |= UAUDIO_RESUME; played += uaudio_thread_play_callback((char *)b->samples + played * uaudio_sample_size, - b->nsamples - played); + b->nsamples - played, + last_flags = flags); + } pthread_mutex_lock(&uaudio_thread_lock); /* Move to next buffer */ uaudio_play_buffer = (1 + uaudio_play_buffer) % UAUDIO_THREAD_BUFFERS; @@ -198,15 +215,12 @@ static void *uaudio_play_thread_fn(void attribute((unused)) *arg) { * @param playcallback Callback to play audio data * @param min Minimum number of samples to play in a chunk * @param max Maximum number of samples to play in a chunk - * @param flags Flags + * @param flags Flags (not currently used) * * @p callback will be called multiple times in quick succession if necessary * to gather at least @p min samples. Equally @p playcallback may be called * repeatedly in quick succession to play however much was received in a single * chunk. - * - * Possible flags are: - * - @ref UAUDIO_THREAD_FAKE_PAUSE */ void uaudio_thread_start(uaudio_callback *callback, void *userdata, @@ -222,6 +236,7 @@ void uaudio_thread_start(uaudio_callback *callback, uaudio_thread_max = max; uaudio_thread_flags = flags; uaudio_thread_started = 1; + uaudio_thread_activated = 0; for(int n = 0; n < UAUDIO_THREAD_BUFFERS; ++n) uaudio_buffers[n].samples = xcalloc_noptr(uaudio_thread_max, uaudio_sample_size); @@ -258,20 +273,14 @@ void uaudio_thread_activate(void) { pthread_mutex_lock(&uaudio_thread_lock); uaudio_thread_activated = 1; pthread_cond_broadcast(&uaudio_thread_cond); - while(!uaudio_thread_collecting) - pthread_cond_wait(&uaudio_thread_cond, &uaudio_thread_lock); pthread_mutex_unlock(&uaudio_thread_lock); } /** @brief Deactivate audio output */ void uaudio_thread_deactivate(void) { pthread_mutex_lock(&uaudio_thread_lock); - uaudio_thread_activated = 0; + uaudio_thread_activated = 0; pthread_cond_broadcast(&uaudio_thread_cond); - if(!(uaudio_thread_flags & UAUDIO_THREAD_FAKE_PAUSE)) { - while(uaudio_thread_collecting || uaudio_buffers_used()) - pthread_cond_wait(&uaudio_thread_cond, &uaudio_thread_lock); - } pthread_mutex_unlock(&uaudio_thread_lock); } diff --git a/lib/uaudio.h b/lib/uaudio.h index d6376c8..37e0ae4 100644 --- a/lib/uaudio.h +++ b/lib/uaudio.h @@ -51,11 +51,33 @@ typedef size_t uaudio_callback(void *buffer, /** @brief Callback to play audio data * @param buffer Pointer to audio buffer * @param samples Number of samples to play + * @param flags Flags word * @return Number of samples played * * Used with uaudio_thread_start() etc. + * + * @p flags is a bitmap giving the current pause state and transitions: + * - @ref UAUDIO_PAUSE if this is the first call of a pause + * - @ref UAUDIO_RESUME if this is the first call of a resumse + * - @ref UAUDIO_PLAYING if this is outside a pause + * - @ref UAUDIO_PAUSED if this is in a pause + * + * During a pause, the sample data is guaranteed to be 0. */ -typedef size_t uaudio_playcallback(void *buffer, size_t samples); +typedef size_t uaudio_playcallback(void *buffer, size_t samples, + unsigned flags); + +/** @brief Start of a pause */ +#define UAUDIO_PAUSE 0x0001 + +/** @brief End of a pause */ +#define UAUDIO_RESUME 0x0002 + +/** @brief Currently playing */ +#define UAUDIO_PLAYING 0x0004 + +/** @brief Currently paused */ +#define UAUDIO_PAUSED 0x0008 /** @brief Audio API definition */ struct uaudio { @@ -139,19 +161,11 @@ void uaudio_thread_start(uaudio_callback *callback, size_t max, unsigned flags); -/** @brief Fake pauses - * - * This flag is used for audio backends that cannot sensibly be paused. - * The thread support code will supply silence while deactivated in this - * case. - */ -#define UAUDIO_THREAD_FAKE_PAUSE 0x00000001 - void uaudio_thread_stop(void); void uaudio_thread_activate(void); void uaudio_thread_deactivate(void); -void uaudio_schedule_synchronize(void); -void uaudio_schedule_update(size_t written_samples); +uint32_t uaudio_schedule_sync(void); +void uaudio_schedule_sent(size_t nsamples); void uaudio_schedule_init(void); const struct uaudio *uaudio_find(const char *name);