/*
* This file is part of DisOrder.
- * Copyright (C) 2004-2009 Richard Kettlewell
+ * Copyright (C) 2004-2012 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
*/
#include "disorder-server.h"
-#include <ao/ao.h>
#define SPEAKER "disorder-speaker"
static int prepare_child(struct queue_entry *q,
const struct pbgc_params *params,
void attribute((unused)) *bgdata);
+static void ensure_next_scratch(ev_source *ev);
/** @brief File descriptor of our end of the socket to the speaker */
static int speaker_fd = -1;
int attribute((unused)) status,
const struct rusage attribute((unused)) *rusage,
void attribute((unused)) *u) {
- fatal(0, "speaker subprocess %s",
- wstat(status));
+ disorder_fatal(0, "speaker subprocess %s", wstat(status));
}
/** @brief Called when we get a message from the speaker process */
switch(sm.type) {
case SM_PAUSED:
/* track ID is paused, DATA seconds played */
- D(("SM_PAUSED %s %ld", sm.id, sm.data));
+ D(("SM_PAUSED %s %ld", sm.u.id, sm.data));
playing->sofar = sm.data;
break;
case SM_FINISHED: /* scratched the playing track */
case SM_STILLBORN: /* scratched too early */
case SM_UNKNOWN: /* scratched WAY too early */
- if(playing && !strcmp(sm.id, playing->id))
+ if(playing && !strcmp(sm.u.id, playing->id)) {
+ if((playing->state == playing_unplayed
+ || playing->state == playing_started)
+ && sm.type == SM_FINISHED)
+ playing->state = playing_ok;
finished(ev);
+ }
break;
case SM_PLAYING:
/* track ID is playing, DATA seconds played */
- D(("SM_PLAYING %s %ld", sm.id, sm.data));
+ D(("SM_PLAYING %s %ld", sm.u.id, sm.data));
playing->sofar = sm.data;
break;
+ case SM_ARRIVED: {
+ /* track ID is now prepared */
+ struct queue_entry *q;
+ for(q = qhead.next; q != &qhead && strcmp(q->id, sm.u.id); q = q->next)
+ ;
+ if(q && q->preparing) {
+ q->preparing = 0;
+ q->prepared = 1;
+ /* We might be waiting to play the now-prepared track */
+ play(ev);
+ }
+ break;
+ }
default:
- error(0, "unknown speaker message type %d", sm.type);
+ disorder_error(0, "unknown speaker message type %d", sm.type);
}
return 0;
}
struct speaker_message sm;
if(socketpair(PF_UNIX, SOCK_DGRAM, 0, sp) < 0)
- fatal(errno, "error calling socketpair");
+ disorder_fatal(errno, "error calling socketpair");
if(!(pid = xfork())) {
exitfn = _exit;
ev_signal_atfork(ev);
log_default == &log_syslog ? "--syslog" : "--no-syslog",
(char *)0);
#endif
- fatal(errno, "error invoking %s", SPEAKER);
+ disorder_fatal(errno, "error invoking %s", SPEAKER);
}
ev_child(ev, pid, 0, speaker_terminated, 0);
speaker_fd = sp[1];
speaker_recv(speaker_fd, &sm);
nonblock(speaker_fd);
if(ev_fd(ev, ev_read, speaker_fd, speaker_readable, 0, "speaker read") < 0)
- fatal(0, "error registering speaker socket fd");
+ disorder_fatal(0, "error registering speaker socket fd");
}
/** @brief Tell the speaker to reload its configuration */
* some time before the speaker reports it as finished) or when a non-raw
* (i.e. non-speaker) player terminates. In the latter case it's imaginable
* that the OS has buffered the last few samples.
- *
+ *
+ * NB. The finished track might NOT be in the queue (yet) - it might be a
+ * pre-chosen scratch.
*/
static int player_finished(ev_source *ev,
pid_t pid,
q->state = playing_ok;
break;
}
- /* Regardless we always report and record the status and do cleanup for
- * prefork calls. */
- if(status)
- error(0, "player for %s %s", q->track, wstat(status));
+ /* Report the status unless we killed it */
+ if(status) {
+ if(!(q->killed && WIFSIGNALED(status) && WTERMSIG(status) == q->killed))
+ disorder_error(0, "player for %s %s", q->track, wstat(status));
+ }
+ /* Clean up any prefork calls */
if(q->type & DISORDER_PLAYER_PREFORK)
play_cleanup(q->pl, q->data);
q->wstat = status;
* @return @ref START_OK, @ref START_HARDFAIL or @ref START_SOFTFAIL
*
* This makes @p actually start playing. It calls prepare() if necessary and
- * either sends an @ref SM_START command or invokes the player itself in a
+ * either sends an @ref SM_PLAY command or invokes the player itself in a
* subprocess.
*
* It's up to the caller to set @ref playing and @c playing->state (this might
D(("start %s", q->id));
/* Find the player plugin. */
- if(!(player = find_player(q)) < 0)
+ if(!(player = find_player(q)))
return START_HARDFAIL; /* No player */
if(!(q->pl = open_plugin(player->s[1], 0)))
return START_HARDFAIL;
* a subprocess. See speaker.c for further discussion. */
struct speaker_message sm[1];
memset(sm, 0, sizeof sm);
- strcpy(sm->id, q->id);
+ strcpy(sm->u.id, q->id);
sm->type = SM_PLAY;
speaker_send(speaker_fd, sm);
- D(("sent SM_PLAY for %s", sm->id));
+ D(("sent SM_PLAY for %s", sm->u.id));
/* Our caller will set playing and playing->state = playing_started */
return START_OK;
} else {
}
}
-/** @brief Child-process half of start() */
+/** @brief Child-process half of start()
+ * @return Process exit code
+ *
+ * Called in subprocess to execute non-raw-format players (via plugin).
+ */
static int start_child(struct queue_entry *q,
const struct pbgc_params *params,
void attribute((unused)) *bgdata) {
- int n;
-
- /* Wait for a device to clear. This ugliness is now deprecated and will
- * eventually be removed. */
- if(params->waitdevice) {
- ao_initialize();
- if(*params->waitdevice) {
- n = ao_driver_id(params->waitdevice);
- if(n == -1)
- fatal(0, "invalid libao driver: %s", params->waitdevice);
- } else
- n = ao_default_driver_id();
- /* Make up a format. */
- ao_sample_format format;
- memset(&format, 0, sizeof format);
- format.bits = 8;
- format.rate = 44100;
- format.channels = 1;
- format.byte_format = AO_FMT_NATIVE;
- int retries = 20;
- struct timespec ts;
- ts.tv_sec = 0;
- ts.tv_nsec = 100000000; /* 0.1s */
- ao_device *device;
- while((device = ao_open_live(n, &format, 0)) == 0 && retries-- > 0)
- nanosleep(&ts, 0);
- if(device)
- ao_close(device);
- }
/* Play the track */
play_track(q->pl,
params->argv, params->argc,
/** @brief Prepare a track for later play
* @return @ref START_OK, @ref START_HARDFAIL or @ref START_SOFTFAIL
*
+ * This can be called either when we want to play the track or slightly before
+ * so that some samples are decoded and available in a buffer.
+ *
* Only applies to raw-format (i.e. speaker-using) players; everything else
* gets @c START_OK.
*/
if(q->pid >= 0)
return START_OK;
/* If the track is already prepared, do nothing */
- if(q->prepared)
+ if(q->prepared || q->preparing)
return START_OK;
/* Find the player plugin */
- if(!(player = find_player(q)) < 0)
+ if(!(player = find_player(q)))
return START_HARDFAIL; /* No player */
q->pl = open_plugin(player->s[1], 0);
q->type = play_get_type(q->pl);
if((q->type & DISORDER_PLAYER_TYPEMASK) != DISORDER_PLAYER_RAW)
return START_OK; /* Not a raw player */
- const int rc = play_background(ev, player, q, prepare_child, NULL);
+ int rc = play_background(ev, player, q, prepare_child, NULL);
if(rc == START_OK) {
ev_child(ev, q->pid, 0, player_finished, q);
- q->prepared = 1;
+ q->preparing = 1;
+ /* Actually the track is still "in flight" */
+ rc = START_SOFTFAIL;
}
return rc;
}
-/** @brief Child-process half of prepare() */
+/** @brief Child-process half of prepare()
+ * @return Process exit code
+ *
+ * Called in subprocess to execute the decoder for a raw-format player.
+ *
+ * @todo We currently run the normalizer from here in a double-fork. This is
+ * unsatisfactory for many reasons: we can't prevent it outliving the main
+ * server and we don't adequately report its exit status.
+ */
static int prepare_child(struct queue_entry *q,
const struct pbgc_params *params,
void attribute((unused)) *bgdata) {
/* np will be the pipe to disorder-normalize */
int np[2];
if(socketpair(PF_UNIX, SOCK_STREAM, 0, np) < 0)
- fatal(errno, "error calling socketpair");
+ disorder_fatal(errno, "error calling socketpair");
/* Beware of the Leopard! On OS X 10.5.x, the order of the shutdown
* calls here DOES MATTER. If you do the SHUT_WR first then the SHUT_RD
* fails with "Socket is not connected". I think this is a bug but
memset(&addr, 0, sizeof addr);
addr.sun_family = AF_UNIX;
snprintf(addr.sun_path, sizeof addr.sun_path,
- "%s/speaker/socket", config->home);
+ "%s/private/speaker", config->home);
int sfd = xsocket(PF_UNIX, SOCK_STREAM, 0);
if(connect(sfd, (const struct sockaddr *)&addr, sizeof addr) < 0)
- fatal(errno, "connecting to %s", addr.sun_path);
+ disorder_fatal(errno, "connecting to %s", addr.sun_path);
/* Send the ID, with a NATIVE-ENDIAN 32 bit length */
uint32_t l = strlen(q->id);
if(write(sfd, &l, sizeof l) < 0
|| write(sfd, q->id, l) < 0)
- fatal(errno, "writing to %s", addr.sun_path);
+ disorder_fatal(errno, "writing to %s", addr.sun_path);
/* Await the ack */
if (read(sfd, &l, 1) < 0)
- fatal(errno, "reading ack from %s", addr.sun_path);
+ disorder_fatal(errno, "reading ack from %s", addr.sun_path);
/* Plumbing */
xdup2(np[0], 0);
xdup2(sfd, 1);
log_default == &log_syslog ? "--syslog" : "--no-syslog",
"--config", configfile,
(char *)0);
- fatal(errno, "executing disorder-normalize");
+ disorder_fatal(errno, "executing disorder-normalize");
/* End of the great-grandchild of disorderd */
}
/* Back in the grandchild of disorderd */
char buffer[64];
snprintf(buffer, sizeof buffer, "DISORDER_RAW_FD=%d", np[1]);
if(putenv(buffer) < 0)
- fatal(errno, "error calling putenv");
+ disorder_fatal(errno, "error calling putenv");
/* Close all the FDs we don't need */
xclose(np[0]);
/* Start the decoder itself */
return 0;
}
+/** @brief Kill a player
+ * @param q Queue entry corresponding to player
+ */
+static void kill_player(struct queue_entry *q) {
+ if(q->pid >= 0)
+ kill(-q->pid, config->signal);
+ q->killed = config->signal;
+}
+
/** @brief Abandon a queue entry
*
* Called from c_remove() (but NOT when scratching a track). Only does
if((q->type & DISORDER_PLAYER_TYPEMASK) != DISORDER_PLAYER_RAW)
return; /* Not a raw player. */
/* Terminate the player. */
- kill(-q->pid, config->signal);
+ kill_player(q);
/* Cancel the track. */
memset(&sm, 0, sizeof sm);
sm.type = SM_CANCEL;
- strcpy(sm.id, q->id);
+ strcpy(sm.u.id, q->id);
speaker_send(speaker_fd, &sm);
}
if(!track)
return;
/* Add the track to the queue */
- q = queue_add(track, 0, WHERE_END, origin_random);
+ q = queue_add(track, 0, WHERE_END, NULL, origin_random);
D(("picked %p (%s) at random", (void *)q, q->track));
queue_write();
/* Maybe a track can now be played */
}
/* It's become the playing track */
playing = q;
- time(&playing->played);
+ xtime(&playing->played);
playing->state = playing_started;
notify_play(playing->track, playing->submitter);
eventlog("playing", playing->track,
* potentially be a just-added random track. */
if(qhead.next != &qhead)
prepare(ev, qhead.next);
+ /* Make sure there is a prepared scratch */
+ ensure_next_scratch(ev);
break;
}
}
/* Miscelleneous ------------------------------------------------------------ */
+int flag_enabled(const char *s) {
+ return !s || !strcmp(s, "yes");
+}
+
/** @brief Return true if play is enabled */
int playing_is_enabled(void) {
- const char *s = trackdb_get_global("playing");
-
- return !s || !strcmp(s, "yes");
+ return flag_enabled(trackdb_get_global("playing"));
}
/** @brief Enable play */
}
/** @brief Disable play */
-void disable_playing(const char *who) {
+void disable_playing(const char *who, ev_source attribute((unused)) *ev) {
trackdb_set_global("playing", "no", who);
}
/** @brief Return true if random play is enabled */
int random_is_enabled(void) {
- const char *s = trackdb_get_global("random-play");
-
- return !s || !strcmp(s, "yes");
+ return flag_enabled(trackdb_get_global("random-play"));
}
/** @brief Enable random play */
}
/** @brief Disable random play */
-void disable_random(const char *who) {
+void disable_random(const char *who, ev_source attribute((unused)) *ev) {
trackdb_set_global("random-play", "no", who);
}
/* Scratching --------------------------------------------------------------- */
+/** @brief Track to play next time something is scratched */
+static struct queue_entry *next_scratch;
+
+/** @brief Ensure there isa prepared scratch */
+static void ensure_next_scratch(ev_source *ev) {
+ if(next_scratch) /* There's one already */
+ return;
+ if(!config->scratch.n) /* There are no scratches */
+ return;
+ int r = rand() * (double)config->scratch.n / (RAND_MAX + 1.0);
+ next_scratch = queue_add(config->scratch.s[r], NULL,
+ WHERE_NOWHERE, NULL, origin_scratch);
+ if(ev)
+ prepare(ev, next_scratch);
+}
+
/** @brief Scratch a track
- * @param User responsible (or NULL)
- * @param Track ID (or NULL for current)
+ * @param who User responsible (or NULL)
+ * @param id Track ID (or NULL for current)
*/
void scratch(const char *who, const char *id) {
- struct queue_entry *q;
struct speaker_message sm;
D(("scratch playing=%p state=%d id=%s playing->id=%s",
playing->state = playing_scratched;
playing->scratched = who ? xstrdup(who) : 0;
/* Find the player and kill the whole process group */
- if(playing->pid >= 0) {
- D(("kill -%d -%lu", config->signal, (unsigned long)playing->pid));
- kill(-playing->pid, config->signal);
- }
+ if(playing->pid >= 0)
+ kill_player(playing);
/* Tell the speaker, if we think it'll care */
if((playing->type & DISORDER_PLAYER_TYPEMASK) == DISORDER_PLAYER_RAW) {
memset(&sm, 0, sizeof sm);
sm.type = SM_CANCEL;
- strcpy(sm.id, playing->id);
+ strcpy(sm.u.id, playing->id);
speaker_send(speaker_fd, &sm);
D(("sending SM_CANCEL for %s", playing->id));
}
- /* put a scratch track onto the front of the queue (but don't
- * bother if playing is disabled) */
- if(playing_is_enabled() && config->scratch.n) {
- int r = rand() * (double)config->scratch.n / (RAND_MAX + 1.0);
- q = queue_add(config->scratch.s[r], who, WHERE_START, origin_scratch);
+ /* If playing is enabled then add a scratch to the queue. Having a scratch
+ * appear in the queue when further play is disabled is weird and
+ * contradicts implicit assumptions made elsewhere, so we try to avoid
+ * it. */
+ if(playing_is_enabled()) {
+ /* Try to make sure there is a scratch */
+ ensure_next_scratch(NULL);
+ /* Insert it at the head of the queue */
+ if(next_scratch){
+ next_scratch->submitter = who;
+ queue_insert_entry(&qhead, next_scratch);
+ eventlog_raw("queue", queue_marshall(next_scratch), (const char *)0);
+ next_scratch = NULL;
+ }
}
notify_scratch(playing->track, playing->submitter, who,
- time(0) - playing->played);
+ xtime(0) - playing->played);
}
}
shutting_down = 1;
/* Shut down the current player */
if(playing) {
- if(playing->pid >= 0)
- kill(-playing->pid, config->signal);
+ kill_player(playing);
playing->state = playing_quitting;
finished(0);
}
/* Zap any background decoders that are going */
for(q = qhead.next; q != &qhead; q = q->next)
- if(q->pid >= 0) {
- D(("kill -%d %lu", config->signal, (unsigned long)q->pid));
- kill(-q->pid, config->signal);
- }
+ kill_player(q);
/* Don't need the speaker any more */
ev_fd_cancel(ev, ev_read, speaker_fd);
xclose(speaker_fd);
case DISORDER_PLAYER_STANDALONE:
if(!(playing->type & DISORDER_PLAYER_PAUSES)) {
default:
- error(0, "cannot pause because player is not powerful enough");
+ disorder_error(0, "cannot pause because player is not powerful enough");
return -1;
}
if(play_pause(playing->pl, &played, playing->data)) {
- error(0, "player indicates it cannot pause");
+ disorder_error(0, "player indicates it cannot pause");
return -1;
}
- time(&playing->lastpaused);
+ xtime(&playing->lastpaused);
playing->uptopause = played;
playing->lastresumed = 0;
break;
speaker_send(speaker_fd, &sm);
break;
}
- if(who) info("paused by %s", who);
+ if(who)
+ disorder_info("paused by %s", who);
notify_pause(playing->track, who);
paused = 1;
if(playing->state == playing_started)
if(!playing) return;
switch(playing->type & DISORDER_PLAYER_TYPEMASK) {
case DISORDER_PLAYER_STANDALONE:
- if(!playing->type & DISORDER_PLAYER_PAUSES) {
+ if(!(playing->type & DISORDER_PLAYER_PAUSES)) {
default:
/* Shouldn't happen */
return;
}
play_resume(playing->pl, playing->data);
- time(&playing->lastresumed);
+ xtime(&playing->lastresumed);
break;
case DISORDER_PLAYER_RAW:
memset(&sm, 0, sizeof sm);
speaker_send(speaker_fd, &sm);
break;
}
- if(who) info("resumed by %s", who);
+ if(who) disorder_info("resumed by %s", who);
notify_resume(playing->track, who);
if(playing->state == playing_paused)
playing->state = playing_started;
eventlog("state", "resume", (char *)0);
}
+/** @brief Request an RTP stream */
+void rtp_request(const struct sockaddr_storage *sa) {
+ struct speaker_message sm;
+ memset(&sm, 0, sizeof sm);
+ sm.type = SM_RTP_REQUEST;
+ sm.u.address = *sa;
+ speaker_send(speaker_fd, &sm);
+}
+
+/** @brief Cancel an RTP stream */
+void rtp_request_cancel(const struct sockaddr_storage *sa) {
+ struct speaker_message sm;
+ memset(&sm, 0, sizeof sm);
+ sm.type = SM_RTP_CANCEL;
+ sm.u.address = *sa;
+ speaker_send(speaker_fd, &sm);
+}
+
/*
Local Variables:
c-basic-offset:2