X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/05b75f8d50b83e943af3be4071449304d82dbdcd..8d251217b5cfeae9d277054a456c4064eabde569:/clients/playrtp.c diff --git a/clients/playrtp.c b/clients/playrtp.c index cbc24ae..c6f0ad9 100644 --- a/clients/playrtp.c +++ b/clients/playrtp.c @@ -2,20 +2,18 @@ * This file is part of DisOrder. * Copyright (C) 2007, 2008 Richard Kettlewell * - * This program is free software; you can redistribute it and/or modify + * 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 2 of the License, or + * 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. - * + * 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, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 - * USA + * along with this program. If not, see . */ /** @file clients/playrtp.c * @brief RTP player @@ -26,20 +24,20 @@ * 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 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. * - * 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. + * 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). * - * 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. + * control_thread() accepts commands from Disobedience (or anything else). + * + * The main thread activates and deactivates audio playing via the @ref + * lib/uaudio.h API (which probably implies at least one further thread). * * 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 @@ -81,6 +79,7 @@ #include "playrtp.h" #include "inputline.h" #include "version.h" +#include "uaudio.h" #define readahead linux_headers_are_borked @@ -96,7 +95,6 @@ static int rtpfd; static FILE *logfp; /** @brief Output device */ -const char *device; /** @brief Minimum low watermark * @@ -106,7 +104,7 @@ unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */ /** @brief Buffer high watermark * * We'll only start playing when this many samples are available. */ -static unsigned readahead = 2 * 2 * 44100; +static unsigned readahead = 44100; /* 0.5 seconds */ /** @brief Maximum buffer size * @@ -170,16 +168,8 @@ pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; /** @brief Condition variable signalled whenever @ref packets is changed */ pthread_cond_t cond = PTHREAD_COND_INITIALIZER; -#if DEFAULT_BACKEND == BACKEND_ALSA -# define DEFAULT_PLAYRTP_BACKEND playrtp_alsa -#elif DEFAULT_BACKEND == BACKEND_OSS -# define DEFAULT_PLAYRTP_BACKEND playrtp_oss -#elif DEFAULT_BACKEND == BACKEND_COREAUDIO -# define DEFAULT_PLAYRTP_BACKEND playrtp_coreaudio -#endif - /** @brief Backend to play with */ -static void (*backend)(void) = DEFAULT_PLAYRTP_BACKEND; +static const struct uaudio *backend; HEAP_DEFINE(pheap, struct packet *, lt_packet); @@ -226,6 +216,8 @@ static const struct option options[] = { { "core-audio", no_argument, 0, 'c' }, #endif { "dump", required_argument, 0, 'r' }, + { "command", required_argument, 0, 'e' }, + { "pause-mode", required_argument, 0, 'P' }, { "socket", required_argument, 0, 's' }, { "config", required_argument, 0, 'C' }, { 0, 0, 0, 0 } @@ -399,6 +391,9 @@ static void *listen_thread(void attribute((unused)) *arg) { timestamp, next_timestamp); continue; } + /* Ignore packets with the extension bit set. */ + if(header.vpxcc & 0x10) + continue; p->next = 0; p->flags = 0; p->timestamp = timestamp; @@ -406,7 +401,7 @@ static void *listen_thread(void attribute((unused)) *arg) { if(header.mpt & 0x80) p->flags |= IDLE; switch(header.mpt & 0x7F) { - case 10: + case 10: /* L16 */ p->nsamples = (n - sizeof header) / sizeof(uint16_t); break; /* TODO support other RFC3551 media types (when the speaker does) */ @@ -478,33 +473,10 @@ struct packet *playrtp_next_packet(void) { return 0; } -/** @brief Play an RTP stream - * - * This is the guts of the program. It is responsible for: - * - starting the listening thread - * - opening the audio device - * - reading ahead to build up a buffer - * - arranging for audio to be played - * - detecting when the buffer has got too small and re-buffering - */ -static void play_rtp(void) { - pthread_t ltid; - int err; - - /* We receive and convert audio data in a background thread */ - if((err = pthread_create(<id, 0, listen_thread, 0))) - fatal(err, "pthread_create listen_thread"); - /* We have a second thread to add received packets to the queue */ - if((err = pthread_create(<id, 0, queue_thread, 0))) - fatal(err, "pthread_create queue_thread"); - /* The rest of the work is backend-specific */ - backend(); -} - /* display usage message and terminate */ static void help(void) { xprintf("Usage:\n" - " disorder-playrtp [OPTIONS] ADDRESS [PORT]\n" + " disorder-playrtp [OPTIONS] [[ADDRESS] PORT]\n" "Options:\n" " --device, -D DEVICE Output device\n" " --min, -m FRAMES Buffer low water mark\n" @@ -521,6 +493,9 @@ static void help(void) { #if HAVE_COREAUDIO_AUDIOHARDWARE_H " --core-audio, -c Use Core Audio to play audio\n" #endif + " --command, -e COMMAND Pipe audio to command.\n" + " --pause-mode, -P silence For -e: pauses send silence (default)\n" + " --pause-mode, -P suspend For -e: pauses suspend writes\n" " --help, -h Display usage message\n" " --version, -V Display version number\n" ); @@ -528,6 +503,66 @@ static void help(void) { exit(0); } +static size_t playrtp_callback(void *buffer, + size_t max_samples, + void attribute((unused)) *userdata) { + size_t samples; + + pthread_mutex_lock(&lock); + /* Get the next packet, junking any that are now in the past */ + const struct packet *p = playrtp_next_packet(); + if(p && contains(p, next_timestamp)) { + /* This packet is ready to play; the desired next timestamp points + * somewhere into it. */ + + /* Timestamp of end of packet */ + const uint32_t packet_end = p->timestamp + p->nsamples; + + /* Offset of desired next timestamp into current packet */ + const uint32_t offset = next_timestamp - p->timestamp; + + /* Pointer to audio data */ + const uint16_t *ptr = (void *)(p->samples_raw + offset); + + /* Compute number of samples left in packet, limited to output buffer + * size */ + samples = packet_end - next_timestamp; + if(samples > max_samples) + samples = max_samples; + + /* Copy into buffer, converting to native endianness */ + size_t i = samples; + int16_t *bufptr = buffer; + while(i > 0) { + *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 + * comparison with max_samples deals with the otherwise troubling overflow + * case. */ + samples = p ? p->timestamp - next_timestamp : max_samples; + if(samples > max_samples) + samples = max_samples; + //info("infill by %zu", samples); + memset(buffer, 0, samples * uaudio_sample_size); + } + /* Debug dump */ + if(dump_buffer) { + for(size_t i = 0; i < samples; ++i) { + dump_buffer[dump_index++] = ((int16_t *)buffer)[i]; + dump_index %= dump_size; + } + } + /* Advance timestamp */ + next_timestamp += samples; + pthread_mutex_unlock(&lock); + return samples; +} + int main(int argc, char **argv) { int n, err; struct addrinfo *res; @@ -547,6 +582,8 @@ int main(int argc, char **argv) { }; union any_sockaddr mgroup; const char *dumpfile = 0; + pthread_t ltid; + static const int one = 1; static const struct addrinfo prefs = { .ai_flags = AI_PASSIVE, @@ -557,29 +594,32 @@ int main(int argc, char **argv) { mem_init(); if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale"); - while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:r", options, 0)) >= 0) { + backend = uaudio_apis[0]; + while((n = getopt_long(argc, argv, "hVdD:m:b: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': device = optarg; 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; #if HAVE_ALSA_ASOUNDLIB_H - case 'a': backend = playrtp_alsa; break; + case 'a': backend = &uaudio_alsa; break; #endif #if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST - case 'o': backend = playrtp_oss; break; + case 'o': backend = &uaudio_oss; break; #endif #if HAVE_COREAUDIO_AUDIOHARDWARE_H - case 'c': backend = playrtp_coreaudio; break; + case 'c': backend = &uaudio_coreaudio; break; #endif case 'C': configfile = optarg; break; case 's': control_socket = optarg; break; case 'r': dumpfile = optarg; break; + case 'e': backend = &uaudio_command; uaudio_set("command", optarg); break; + case 'P': uaudio_set("pause-mode", optarg); break; default: fatal(0, "invalid option"); } } @@ -616,8 +656,14 @@ int main(int argc, char **argv) { res->ai_socktype, res->ai_protocol)) < 0) fatal(errno, "error creating socket"); - /* Stash the multicast group address */ - if((is_multicast = multicast(res->ai_addr))) { + /* 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: @@ -626,24 +672,13 @@ int main(int argc, char **argv) { case AF_INET6: mgroup.in6.sin6_port = 0; break; + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); } - } - /* Bind to 0/port */ - switch(res->ai_addr->sa_family) { - case AF_INET: - memset(&((struct sockaddr_in *)res->ai_addr)->sin_addr, 0, - sizeof (struct in_addr)); - break; - case AF_INET6: - memset(&((struct sockaddr_in6 *)res->ai_addr)->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", sockname); - if(is_multicast) { + /* 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; @@ -662,10 +697,35 @@ int main(int argc, char **argv) { 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 + } 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)); + } len = sizeof rcvbuf; if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0) fatal(errno, "error calling getsockopt SO_RCVBUF"); @@ -709,7 +769,41 @@ int main(int argc, char **argv) { fatal(errno, "mapping %s", dumpfile); info("dumping to %s", dumpfile); } - play_rtp(); + /* Set up output. Currently we only support L16 so there's no harm setting + * the format before we know what it is! */ + uaudio_set_format(44100/*Hz*/, 2/*channels*/, + 16/*bits/channel*/, 1/*signed*/); + backend->start(playrtp_callback, NULL); + /* We receive and convert audio data in a background thread */ + if((err = pthread_create(<id, 0, listen_thread, 0))) + fatal(err, "pthread_create listen_thread"); + /* We have a second thread to add received packets to the queue */ + if((err = pthread_create(<id, 0, queue_thread, 0))) + fatal(err, "pthread_create queue_thread"); + pthread_mutex_lock(&lock); + for(;;) { + /* Wait for the buffer to fill up a bit */ + playrtp_fill_buffer(); + /* Start playing now */ + info("Playing..."); + next_timestamp = pheap_first(&packets)->timestamp; + active = 1; + pthread_mutex_unlock(&lock); + backend->activate(); + pthread_mutex_lock(&lock); + /* Wait until the buffer empties out */ + while(nsamples >= minbuffer + || (nsamples > 0 + && contains(pheap_first(&packets), next_timestamp))) { + pthread_cond_wait(&cond, &lock); + } + /* Stop playing for a bit until the buffer re-fills */ + pthread_mutex_unlock(&lock); + backend->deactivate(); + pthread_mutex_lock(&lock); + active = 0; + /* Go back round */ + } return 0; }