struct timeval clock_diff_barrier;
/** Number of the audio format as announced by para_server. */
int current_audio_format_num;
+ /* The status task btrn is the child of the client task. */
+ struct btr_node *btrn;
};
/** The array of status items sent by para_server. */
#define FOR_EACH_AUDIO_FORMAT(af) for (af = 0; af < NUM_AUDIO_FORMATS; af++)
/**
- * get the audio format number
- * \param name the name of the audio format
+ * Get the audio format number.
+ *
+ * \param name The name of the audio format.
*
* \return The audio format number on success, -E_UNSUPPORTED_AUDIO_FORMAT if
* \a name is not a supported audio format.
return -E_UNSUPPORTED_AUDIO_FORMAT;
}
+/**
+ * Compute the play time based on information of the given slot.
+ *
+ * \param slot_num The slot number (negative means: no slot).
+ *
+ * This computes a string of the form "0:07 [3:33] (3%/3:40)" using information
+ * from the status items received from para_server and the start time of the
+ * (first) writer of the given slot.
+ *
+ * It has to to take into account that probably the stream was not started at
+ * the beginning of the file, that the clock between the server and the client
+ * host may differ and that playback of the stream was delayed, e.g. because
+ * the prebuffer filter is used in the filter chain of the given slot.
+ *
+ * If no writer is active in the given slot, or \a slot_num is negative
+ * (indicating that para_audiod runs in standby mode), an approximation based
+ * only on the status items is computed and the returned string is prefixed
+ * with "~".
+ *
+ * \return A string that must be freed by the caller.
+ */
char *get_time_string(int slot_num)
{
int ret, seconds = 0, length;
}
if (audiod_status == AUDIOD_ON && !s)
goto empty;
- /* valid status items and playing */
+ /*
+ * Valid status items and playing, set length and tmp to the stream
+ * start. We use the slot info and fall back to the info from current
+ * status items if no slot info is available.
+ */
+ length = stat_task->length_seconds;
+ tmp = &stat_task->server_stream_start;
if (s && s->wns) { /* writer active in this slot */
- length = s->seconds_total;
- tmp = &s->server_stream_start;
- } else { /* standby mode, rely on status items */
- length = stat_task->length_seconds;
- tmp = &stat_task->server_stream_start;
+ btr_get_node_start(s->wns[0].btrn, &wstime);
+ if (wstime.tv_sec != 0) { /* writer wrote something */
+ if (s->server_stream_start.tv_sec == 0) {
+ /* copy status info to slot */
+ s->server_stream_start = stat_task->server_stream_start;
+ s->offset_seconds = stat_task->offset_seconds;
+ s->seconds_total = stat_task->length_seconds;
+ }
+ length = s->seconds_total;
+ tmp = &s->server_stream_start;
+ }
}
if (stat_task->sa_time_diff_sign > 0)
tv_diff(tmp, &stat_task->sa_time_diff, &sss);
seconds = diff.tv_sec + stat_task->offset_seconds;
goto out;
}
- btr_get_node_start(s->wns[0].btrn, &wstime);
tv_diff(now, &wstime, &wtime);
//PARA_CRIT_LOG("offset %d\n", s->offset_seconds);
seconds = s->offset_seconds;
{
struct slot_info *s = &slot[slot_num];
struct audio_format_info *a;
+ struct timeval restart_delay = {0, 200 * 1000};
if (s->format < 0 || !s->receiver_node)
return;
btr_free_node(s->receiver_node->btrn);
free(s->receiver_node);
s->receiver_node = NULL;
- stat_task->current_audio_format_num = -1;
+ tv_add(now, &restart_delay, &afi[s->format].restart_barrier);
}
static void writer_cleanup(struct writer_node *wn)
{
- struct writer *w = writers + wn->writer_num;
+ struct writer *w;
+ if (!wn)
+ return;
+ w = writers + wn->writer_num;
+ PARA_INFO_LOG("closing %s\n", writer_names[wn->writer_num]);
w->close(wn);
btr_free_node(wn->btrn);
}
if (s->format < 0)
return;
+ assert(s->wns);
a = afi + s->format;
if (a->num_writers == 0)
writer_cleanup(s->wns);
s->wns = NULL;
}
-static void _close_filters(struct slot_info *s)
+static void close_filters(struct slot_info *s)
{
int i;
struct audio_format_info *a = afi + s->format;
return;
for (i = 0; i < a->num_filters; i++) {
struct filter_node *fn = s->fns + i;
- struct filter *f = filters + fn->filter_num;
+ struct filter *f;
- f->close(fn);
+ if (!fn)
+ continue;
+ f = filters + fn->filter_num;
+ if (f->close)
+ f->close(fn);
btr_free_node(fn->btrn);
}
free(s->fns);
s->fns = NULL;
}
+/*
+ * Whenever a task commits suicide by returning from post_select with t->error
+ * < 0, it also removes its btr node. We do exactly that to kill a running
+ * task. Note that the scheduler checks t->error also _before_ each pre/post
+ * select call, so the victim will never be scheduled again.
+ */
+static void kill_btrn(struct btr_node *btrn, struct task *t, int error)
+{
+ if (t->error < 0)
+ return;
+ t->error = error;
+ btr_remove_node(btrn);
+}
+
static void kill_all_decoders(int error)
{
int i, j;
a = afi + s->format;
if (s->wns)
for (j = 0; j < a->num_writers; j++)
- s->wns[j].task.error = error;
+ kill_btrn(s->wns[j].btrn, &s->wns[j].task, error);
if (s->fns)
for (j = 0; j < a->num_writers; j++)
- s->fns[j].task.error = error;
+ kill_btrn(s->fns[j].btrn, &s->wns[j].task, error);
if (s->receiver_node)
- s->receiver_node->task.error = error;
+ kill_btrn(s->receiver_node->btrn, &s->receiver_node->task,
+ error);
}
}
fn->conf = a->filter_conf[i];
fn->task.pre_select = f->pre_select;
fn->task.post_select = f->post_select;
- fn->btrn = btr_new_node(f->name, parent, f->execute, fn);
+
+ fn->btrn = btr_new_node(&(struct btr_node_description)
+ EMBRACE(.name = f->name, .parent = parent,
+ .handler = f->execute, .context = fn));
+
f->open(fn);
register_task(&fn->task);
parent = fn->btrn;
PARA_NOTICE_LOG("%s filter %d/%d (%s) started in slot %d\n",
- audio_formats[s->format], i, nf, f->name, s - slot);
- sprintf(fn->task.status, "%s (slot %d)", f->name, s - slot);
+ audio_formats[s->format], i, nf, f->name, (int)(s - slot));
+ sprintf(fn->task.status, "%s (slot %d)", f->name, (int)(s - slot));
}
}
register_writer_node(wn, parent);
}
}
- s->server_stream_start = stat_task->server_stream_start.tv_sec?
- stat_task->server_stream_start : *now;
- s->offset_seconds = stat_task->offset_seconds;
- s->seconds_total = stat_task->length_seconds;
}
/* returns slot num on success */
rn = s->receiver_node;
rn->receiver = r;
rn->conf = a->receiver_conf;
- rn->btrn = btr_new_node(r->name, NULL, NULL, rn);
+ rn->btrn = btr_new_node(&(struct btr_node_description)
+ EMBRACE(.name = r->name, .context = rn));
ret = r->open(rn);
if (ret < 0) {
btr_free_node(rn->btrn);
return ret;
}
-/* returns slot num on success. */
-static int open_current_receiver(struct sched *s)
+/**
+ * Return the root node of the current buffer tree.
+ *
+ * This is only used for stream grabbing.
+ *
+ * \return \p NULL if no slot is currently active. If more than one buffer tree
+ * exists, the node corresponding to the most recently started receiver is
+ * returned.
+ */
+struct btr_node *audiod_get_btr_root(void)
+{
+ int i, newest_slot = -1;
+ struct timeval newest_rstime = {0, 0};
+
+ FOR_EACH_SLOT(i) {
+ struct slot_info *s = &slot[i];
+ struct timeval rstime;
+ if (!s->receiver_node)
+ continue;
+ if (s->receiver_node->task.error < 0)
+ continue;
+ btr_get_node_start(s->receiver_node->btrn, &rstime);
+ if (newest_slot >= 0 && tv_diff(&rstime, &newest_rstime, NULL) < 0)
+ continue;
+ newest_rstime = rstime;
+ newest_slot = i;
+ }
+ if (newest_slot == -1)
+ return NULL;
+ return slot[newest_slot].receiver_node->btrn;
+}
+
+/* whether a new instance of a decoder should be started. */
+static bool must_start_decoder(void)
{
- struct timeval diff;
int ret, cafn = stat_task->current_audio_format_num;
if (cafn < 0 || !stat_task->ct)
- return -1;
+ return false;
/* Do nothing if the 'N' flag is set or the 'P' flag is unset */
if (stat_task->vss_status != VSS_STATUS_FLAG_PLAYING)
- return -1;
+ return false;
ret = receiver_running(cafn);
- if (ret > 0) /* already running and not eof */
- return -1;
- if (ret < 0) { /* eof */
- /*
- * para_server uses a zero start time during the announcement
- * period, i.e. before it sends the first chunk. Wait until
- * this period begins to avoid restarting the receiver that
- * belongs to the file just completed.
- */
- PARA_CRIT_LOG("->>>>>>>>>>>>>>>>>> no delay\n");
- if (stat_task->server_stream_start.tv_sec != 0) {
- PARA_CRIT_LOG("->>>>>>>>>>>>>>>>>><delay\n");
- return -1;
- }
- }
- if (tv_diff(now, &afi[cafn].restart_barrier, &diff) < 0) {
- /* avoid busy loop */
- s->timeout = diff;
- return -1;
- }
- /* start a new receiver */
- return open_receiver(cafn);
+ if (ret != 0) /* already running */
+ return false;
+ if (tv_diff(now, &afi[cafn].restart_barrier, NULL) < 0)
+ return false;
+ return true;
}
static unsigned compute_time_diff(const struct timeval *status_time)
&tmp);
stat_task->sa_time_diff = tmp;
PARA_INFO_LOG("time diff (cur/avg): %s%lums/%s%lums\n",
- sign > 0? "+" : "-",
+ sign < 0? "-" : "+",
tv2ms(&diff),
- sa_time_diff_sign ? "+" : "-",
+ sa_time_diff_sign < 0? "-" : "+",
tv2ms(&stat_task->sa_time_diff)
);
out:
{
if (!stat_task->ct)
return;
+ btr_free_node(stat_task->ct->btrn);
client_close(stat_task->ct);
stat_task->ct = NULL;
clear_and_dump_items();
for (i = 0; i < a->num_filters; i++)
if (s->fns && s->fns[i].task.error != -E_TASK_UNREGISTERED)
return;
- for (i = 0; i < a->num_writers; i++)
- if (s->wns && s->wns[i].task.error != -E_TASK_UNREGISTERED)
+ if (a->num_writers > 0) {
+ for (i = 0; i < a->num_writers; i++)
+ if (s->wns && s->wns[i].task.error != -E_TASK_UNREGISTERED)
+ return;
+ } else {
+ if (s->wns && s->wns[0].task.error != -E_TASK_UNREGISTERED)
return;
+ }
PARA_INFO_LOG("closing slot %d\n", slot_num);
close_writers(s);
- _close_filters(s);
+ close_filters(s);
close_receiver(slot_num);
clear_slot(slot_num);
}
* Check if any receivers/filters/writers need to be started and do so if
* necessary.
*/
-static void start_stop_decoders(struct sched *s)
+static void start_stop_decoders(void)
{
int i, ret;
struct slot_info *sl;
FOR_EACH_SLOT(i)
try_to_close_slot(i);
-// if (audiod_status != AUDIOD_ON ||
-// !(stat_task->vss_status & VSS_STATUS_FLAG_PLAYING))
-// return kill_all_decoders(-E_NOT_PLAYING);
- ret = open_current_receiver(s);
+ if (audiod_status != AUDIOD_ON ||
+ !(stat_task->vss_status & VSS_STATUS_FLAG_PLAYING))
+ return kill_all_decoders(-E_NOT_PLAYING);
+ if (!must_start_decoder())
+ return;
+ ret = open_receiver(stat_task->current_audio_format_num);
if (ret < 0)
return;
sl = slot + ret;
if (a->num_filters)
open_filters(sl);
open_writers(sl);
+ activate_grab_clients();
btr_log_tree(sl->receiver_node->btrn, LL_NOTICE);
- s->timeout.tv_sec = 0;
- s->timeout.tv_usec = 1;
}
-/* restart the client task if necessary */
static void status_pre_select(struct sched *s, struct task *t)
+{
+ struct status_task *st = container_of(t, struct status_task, task);
+ int ret, cafn = stat_task->current_audio_format_num;
+
+ if (must_start_decoder())
+ goto min_delay;
+ ret = btr_node_status(st->btrn, 0, BTR_NT_LEAF);
+ if (ret > 0)
+ goto min_delay;
+ if (!st->ct)
+ sched_request_barrier_or_min_delay(&st->restart_barrier, s);
+ if (cafn >= 0)
+ sched_request_barrier(&afi[cafn].restart_barrier, s);
+ return;
+min_delay:
+ sched_min_delay(s);
+}
+
+/* restart the client task if necessary */
+static void status_post_select(__a_unused struct sched *s, struct task *t)
{
struct status_task *st = container_of(t, struct status_task, task);
if (!st->ct)
goto out;
if (st->ct->task.error >= 0) {
- st->ct->task.error = -E_AUDIOD_OFF;
+ kill_btrn(st->ct->btrn, &st->ct->task, -E_AUDIOD_OFF);
goto out;
}
if (st->ct->task.error != -E_TASK_UNREGISTERED)
goto out;
}
if (st->ct) {
+ char *buf;
+ size_t sz;
int ret;
if (st->ct->task.error < 0) {
if (st->ct->task.error != -E_TASK_UNREGISTERED)
}
if (st->ct->status != CL_RECEIVING)
goto out;
- ret = for_each_stat_item(st->ct->buf, st->ct->loaded,
- update_item);
+ ret = btr_node_status(st->btrn, 0, BTR_NT_LEAF);
+ if (ret <= 0)
+ goto out;
+ sz = btr_next_buffer(st->btrn, &buf);
+ ret = for_each_stat_item(buf, sz, update_item);
if (ret < 0) {
- st->ct->task.error = ret;
+ kill_btrn(st->ct->btrn, &st->ct->task, ret);
goto out;
}
- if (st->ct->loaded != ret) {
+ if (sz != ret)
st->last_status_read = *now;
- st->ct->loaded = ret;
- } else {
+ else {
struct timeval diff;
tv_diff(now, &st->last_status_read, &diff);
if (diff.tv_sec > 61)
- st->ct->task.error = -E_STATUS_TIMEOUT;
+ kill_btrn(st->ct->btrn, &st->ct->task,
+ -E_STATUS_TIMEOUT);
}
+ btr_consume(st->btrn, sz - ret);
goto out;
}
if (tv_diff(now, &st->restart_barrier, NULL) < 0)
goto out;
if (st->clock_diff_count) { /* get status only one time */
- char *argv[] = {"audiod", "--", "stat", "-p", "1", NULL};
+ char *argv[] = {"audiod", "--", "stat", "-p", "-n=1", NULL};
int argc = 5;
PARA_INFO_LOG("clock diff count: %d\n", st->clock_diff_count);
st->clock_diff_count--;
- client_open(argc, argv, &st->ct, NULL);
+ client_open(argc, argv, &st->ct, NULL, NULL, st->btrn);
set_stat_task_restart_barrier(2);
} else {
char *argv[] = {"audiod", "--", "stat", "-p", NULL};
int argc = 4;
- client_open(argc, argv, &st->ct, NULL);
+ client_open(argc, argv, &st->ct, NULL, NULL, st->btrn);
set_stat_task_restart_barrier(5);
}
free(stat_item_values[SI_BASENAME]);
stat_client_write_item(SI_BASENAME);
st->last_status_read = *now;
out:
- start_stop_decoders(s);
+ start_stop_decoders();
}
static void init_status_task(struct status_task *st)
{
memset(st, 0, sizeof(struct status_task));
st->task.pre_select = status_pre_select;
+ st->task.post_select = status_post_select;
st->sa_time_diff_sign = 1;
st->clock_diff_count = conf.clock_diff_count_arg;
st->current_audio_format_num = -1;
- sprintf(st->task.status, "status task");
+ sprintf(st->task.status, "stat");
+ st->btrn = btr_new_node(&(struct btr_node_description)
+ EMBRACE(.name = "stat"));
}
static void set_initial_status(void)
set_initial_status();
FOR_EACH_SLOT(i)
clear_slot(i);
- init_grabbing();
setup_signal_handling();
signal_setup_default(sig_task);
projects / paraslash.git / blobdiff