/*
* This file is part of DisOrder
- * Copyright (C) 2005-2009 Richard Kettlewell
+ * Copyright (C) 2005-2013 Richard Kettlewell
* Portions (C) 2007 Mark Wooding
*
* This program is free software: you can redistribute it and/or modify
* assumed that the main server won't start outrageously many decoders.
*
* Audio is supplied from this buffer to the uaudio play callback. Playback is
- * enabled when a track is to be played and disabled when the its last bytes
- * have been return by the callback; pause and resume is implemneted the
+ * enabled when a track is to be played and disabled when its last bytes
+ * have been returned by the callback; pause and resume is implemented the
* obvious way. If the callback finds itself required to play when there is no
* playing track it returns dead air.
*
#include <syslog.h>
#include <unistd.h>
#include <errno.h>
-#include <ao/ao.h>
#include <sys/select.h>
#include <sys/wait.h>
#include <time.h>
struct track *next;
/** @brief Input file descriptor */
- int fd; /* input FD */
+ int fd;
/** @brief Track ID */
char id[24];
*/
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
-/** @brief Linked list of all prepared tracks */
+/** @brief Linked list of all prepared tracks
+ *
+ * This includes @ref playing and @ref pending_playing.
+ */
static struct track *tracks;
/** @brief Playing track, or NULL
*
* This means the track the speaker process intends to play. It does not
* reflect any other state (e.g. activation of uaudio backend).
+ *
+ * This track remains on @ref track.
*/
static struct track *playing;
/** @brief Pending playing track, or NULL
*
* This means the track the server wants the speaker to play.
+ *
+ * This track remains on @p track.
*/
static struct track *pending_playing;
/** @brief Array of file descriptors for poll() */
static struct pollfd fds[NFDS];
-/** @brief Next free slot in @ref fds */
+/** @brief Next free slot in @ref fds
+ *
+ * This is used when filling in the @ref fds array each iteration through the
+ * event loop.
+ */
static int fdno;
/** @brief Listen socket */
/** @brief Find track @p id, maybe creating it if not found
* @param id Track ID to find
- * @param create If nonzero, create track structure of @p id not found
+ * @param create If nonzero, create track structure of @p id if not found
* @return Pointer to track structure or NULL
*/
static struct track *findtrack(const char *id, int create) {
* This is effectively the read callback on @c t->fd. It is called from the
* main loop whenever the track's file descriptor is readable, assuming the
* buffer has not reached the maximum allowed occupancy.
+ *
+ * Errors count as EOF.
*/
static int speaker_fill(struct track *t) {
size_t where, left;
n = read(t->fd, t->buffer + where, left);
} while(n < 0 && errno == EINTR);
pthread_mutex_lock(&lock);
- if(n < 0) {
- if(errno != EAGAIN)
- disorder_fatal(errno, "error reading sample stream");
+ if(n < 0 && errno == EAGAIN) {
+ /* EAGAIN means more later */
rc = 0;
- } else if(n == 0) {
- D(("fill %s: eof detected", t->id));
+ } else if(n <= 0) {
+ /* n=0 means EOF. n<0 means some error occurred. We log the error but
+ * otherwise treat it as identical to EOF. */
+ if(n < 0)
+ disorder_error(errno, "error reading sample stream for %s", t->id);
+ else
+ D(("fill %s: eof detected", t->id));
t->eof = 1;
/* A track always becomes playable at EOF; we're not going to see any
* more data. */
t->playable = 1;
rc = 0;
}
- }
+ } else
+ rc = 0;
return rc;
}
* We don't allow tracks to be paused if we've already told the server we've
* finished them; that would cause such tracks to survive much longer than the
* few samples they're supposed to, with report() remaining silent for the
- * duration.
+ * duration. The effect is that if you hit pause towards the end of a track,
+ * what should happen is that it finished but the next one is paused right at
+ * its start.
*/
static int playable(void) {
return playing
return;
memset(&sm, 0, sizeof sm);
sm.type = paused ? SM_PAUSED : SM_PLAYING;
- strcpy(sm.id, playing->id);
+ strcpy(sm.u.id, playing->id);
sm.data = playing->played / (uaudio_rate * uaudio_channels);
speaker_send(1, &sm);
xtime(&last_report);
static size_t speaker_callback(void *buffer,
size_t max_samples,
void attribute((unused)) *userdata) {
- const size_t max_bytes = max_samples * uaudio_sample_size;
+ size_t max_bytes = max_samples * uaudio_sample_size;
size_t provided_samples = 0;
+ /* Be sure to keep the amount of data in a buffer a whole number of frames:
+ * otherwise the playing threads can become stuck. */
+ max_bytes -= max_bytes % (uaudio_sample_size * uaudio_channels);
+
pthread_mutex_lock(&lock);
/* TODO perhaps we should immediately go silent if we've been asked to pause
* or cancel the playing track (maybe block in the cancel case and see what
/* Limit to what we were asked for */
if(bytes > max_bytes)
bytes = max_bytes;
+ /* And truncate to a whole number of frames. */
+ bytes -= bytes % (uaudio_sample_size * uaudio_channels);
/* Provide it */
memcpy(buffer, playing->buffer + playing->start, bytes);
playing->start += bytes;
/* Wrap around to start of buffer */
if(playing->start == sizeof playing->buffer)
playing->start = 0;
- /* See if we've reached the end of the track */
+ /* See if we've reached the end of the track; if so make sure the event
+ * loop wakes up. */
if(playing->used == 0 && playing->eof) {
int ignored = write(sigpipe[1], "", 1);
(void) ignored;
struct speaker_message sm;
int n, fd, stdin_slot, timeout, listen_slot, sigpipe_slot;
- /* Keep going while our parent process is alive */
pthread_mutex_lock(&lock);
+ /* Keep going while our parent process is alive */
while(getppid() != 1) {
int force_report = 0;
playing->slot = addfd(playing->fd, POLLIN);
else if(playing)
playing->slot = -1;
+ /* Allow the poll() to be interrupted at the end of a track */
+ sigpipe_slot = addfd(sigpipe[0], POLLIN);
/* If any other tracks don't have a full buffer, try to read sample data
* from them. We do this last of all, so that if we run out of slots,
* nothing important can't be monitored. */
} else
t->slot = -1;
}
- sigpipe_slot = addfd(sigpipe[0], POLLIN);
/* Wait for something interesting to happen */
pthread_mutex_unlock(&lock);
n = poll(fds, fdno, timeout);
char id[24];
if((fd = accept(listenfd, (struct sockaddr *)&addr, &addrlen)) >= 0) {
+ /* We do blocking reads for the header. In theory this means that the
+ * connecting process could wedge the speaker indefinitely. In
+ * practice that would mean that the main server was broken anyway.
+ * Still, this is ugly, and a rewrite would be nice. */
blocking(fd);
if(read(fd, &l, sizeof l) < 4) {
disorder_error(errno, "reading length from inbound connection");
D(("id %s fd %d", id, fd));
t = findtrack(id, 1/*create*/);
if (write(fd, "", 1) < 0) /* write an ack */
- disorder_error(errno, "writing ack to inbound connection");
+ disorder_error(errno, "writing ack to inbound connection for %s",
+ id);
if(t->fd != -1) {
disorder_error(0, "%s: already got a connection", id);
xclose(fd);
nonblock(fd);
t->fd = fd; /* yay */
}
+ /* Notify the server that the connection arrived */
+ sm.type = SM_ARRIVED;
+ strcpy(sm.u.id, id);
+ speaker_send(1, &sm);
}
} else
disorder_error(errno, "accept");
if(pending_playing)
disorder_fatal(0, "got SM_PLAY but have a pending playing track");
}
- t = findtrack(sm.id, 1);
+ t = findtrack(sm.u.id, 1);
D(("SM_PLAY %s fd %d", t->id, t->fd));
if(t->fd == -1)
disorder_error(0,
force_report = 1;
break;
case SM_CANCEL:
- D(("SM_CANCEL %s", sm.id));
- t = removetrack(sm.id);
+ D(("SM_CANCEL %s", sm.u.id));
+ t = removetrack(sm.u.id);
if(t) {
if(t == playing || t == pending_playing) {
/* Scratching the track that the server believes is playing,
* log more because there's been a bug here recently than because
* it's particularly interesting; the log message will be removed
* if no further problems show up. */
- disorder_info("SM_CANCEL for nonplaying track %s", sm.id);
+ disorder_info("SM_CANCEL for nonplaying track %s", sm.u.id);
sm.type = SM_STILLBORN;
}
- strcpy(sm.id, t->id);
+ strcpy(sm.u.id, t->id);
destroy(t);
} else {
/* Probably scratching the playing track well before it's got
* going, but could indicate a bug, so we log this as an error. */
sm.type = SM_UNKNOWN;
- disorder_error(0, "SM_CANCEL for unknown track %s", sm.id);
+ disorder_error(0, "SM_CANCEL for unknown track %s", sm.u.id);
}
speaker_send(1, &sm);
force_report = 1;
disorder_error(0, "cannot read configuration");
disorder_info("reloaded configuration");
break;
+ case SM_RTP_REQUEST:
+ /* TODO the error behavior here is really unhelpful */
+ if(rtp_add_recipient(&sm.u.address))
+ disorder_error(0, "unacceptable RTP destination");
+ break;
+ case SM_RTP_CANCEL:
+ if(rtp_remove_recipient(&sm.u.address))
+ disorder_error(0, "unacceptable RTP destination for removal");
+ break;
default:
disorder_error(0, "unknown message type %d", sm.type);
}
&& playing->used <= early_finish) {
memset(&sm, 0, sizeof sm);
sm.type = SM_FINISHED;
- strcpy(sm.id, playing->id);
+ strcpy(sm.u.id, playing->id);
speaker_send(1, &sm);
playing->finished = 1;
}
/* When the track is actually finished, deconfigure it */
if(playing && playing->eof && !playing->used) {
+ if(!playing->finished) {
+ /* should never happen but we'd like to know if it does */
+ disorder_fatal(0, "track finish state inconsistent");
+ }
removetrack(playing->id);
destroy(playing);
playing = 0;
/* backend-specific initialization */
if(backend->configure)
backend->configure();
+ uaudio_set("application", "disorder-speaker");
backend->start(speaker_callback, NULL);
- /* create the socket directory */
- byte_xasprintf(&dir, "%s/speaker", config->home);
+ /* create the private socket directory */
+ byte_xasprintf(&dir, "%s/private", config->home);
unlink(dir); /* might be a leftover socket */
if(mkdir(dir, 0700) < 0 && errno != EEXIST)
disorder_fatal(errno, "error creating %s", dir);
listenfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
memset(&addr, 0, sizeof addr);
addr.sun_family = AF_UNIX;
- snprintf(addr.sun_path, sizeof addr.sun_path, "%s/speaker/socket",
+ snprintf(addr.sun_path, sizeof addr.sun_path, "%s/private/speaker",
config->home);
if(unlink(addr.sun_path) < 0 && errno != ENOENT)
disorder_error(errno, "removing %s", addr.sun_path);
disorder_fatal(errno, "error binding socket to %s", addr.sun_path);
xlisten(listenfd, 128);
nonblock(listenfd);
+ disorder_info("version "VERSION" process ID %lu",
+ (unsigned long)getpid());
disorder_info("listening on %s", addr.sun_path);
memset(&sm, 0, sizeof sm);
sm.type = SM_READY;