From: Andre Noll <maan@systemlinux.org>
Date: Wed, 6 Jan 2010 23:48:47 +0000 (+0100)
Subject: Switch audiod over to the buffer tree API.
X-Git-Tag: v0.4.2~181
X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=commitdiff_plain;h=c282c836791cedf57c128555af90af37c7c01c05

Switch audiod over to the buffer tree API.

Still a bit rough and there are too many btr merges.
---

diff --git a/alsa_write.c b/alsa_write.c
index 77621979..8e3688fa 100644
--- a/alsa_write.c
+++ b/alsa_write.c
@@ -66,6 +66,7 @@ static int alsa_init(struct private_alsa_write_data *pad,
 	snd_pcm_uframes_t period_size;
 	int err;
 
+	PARA_INFO_LOG("opening %s\n", conf->device_arg);
 	err = snd_pcm_open(&pad->handle, conf->device_arg,
 		SND_PCM_STREAM_PLAYBACK, SND_PCM_NONBLOCK);
 	if (err < 0)
@@ -290,7 +291,6 @@ static void alsa_close(struct writer_node *wn)
 		snd_pcm_close(pad->handle);
 		snd_config_update_free_global();
 	}
-	alsa_cmdline_parser_free(wn->conf);
 	free(pad);
 }
 
@@ -312,7 +312,7 @@ again:
 		return;
 	btr_merge(btrn, wn->min_iqs);
 	bytes = btr_next_buffer(btrn, &data);
-	if (bytes < pad->bytes_per_frame) { /* eof */
+	if (ret < 0 || bytes < pad->bytes_per_frame) { /* eof */
 		assert(btr_no_parent(btrn));
 		ret = -E_ALSA_EOF;
 		if (!pad->handle)
@@ -321,13 +321,13 @@ again:
 		avail = snd_pcm_avail_update(pad->handle);
 		if (avail + 1000 > pad->buffer_frames)
 			goto err;
+		PARA_DEBUG_LOG("waiting for device to drain\n");
 		return;
 	}
 	if (!pad->handle) {
 		struct alsa_write_args_info *conf = wn->conf;
 		if (bytes == 0) /* no data available */
 			return;
-		PARA_CRIT_LOG("alsa init\n");
 		/* defaults */
 		pad->samplerate = conf->samplerate_arg;
 		pad->channels = conf->channels_arg;
@@ -394,12 +394,17 @@ err_out:
 	return NULL;
 }
 
+static void alsa_free_config(void *conf)
+{
+	alsa_cmdline_parser_free(conf);
+}
+
 /**
- * the init function of the alsa writer
+ * The init function of the alsa writer.
  *
- * \param w pointer to the writer to initialize
+ * \param w Pointer to the writer to initialize.
  *
- * \sa struct writer
+ * \sa \ref struct writer.
  */
 void alsa_write_init(struct writer *w)
 {
@@ -414,6 +419,7 @@ void alsa_write_init(struct writer *w)
 	w->post_select_btr = alsa_write_post_select_btr;
 	w->parse_config = alsa_parse_config;
 	w->shutdown = NULL; /* nothing to do */
+	w->free_config = alsa_free_config;
 	w->help = (struct ggo_help) {
 		.short_help = alsa_write_args_info_help,
 		.detailed_help = alsa_write_args_info_detailed_help
diff --git a/amp_filter.c b/amp_filter.c
index d0a893f4..e91791bf 100644
--- a/amp_filter.c
+++ b/amp_filter.c
@@ -156,6 +156,11 @@ err:
 	btr_remove_node(btrn);
 }
 
+static void amp_free_config(void *conf)
+{
+	amp_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the amplify filter.
  *
@@ -172,6 +177,7 @@ void amp_filter_init(struct filter *f)
 	f->pre_select = generic_filter_pre_select;
 	f->post_select = amp_post_select;
 	f->parse_config = amp_parse_config;
+	f->free_config = amp_free_config;
 	f->help = (struct ggo_help) {
 		.short_help = amp_filter_args_info_help,
 		.detailed_help = amp_filter_args_info_detailed_help
diff --git a/audiod.c b/audiod.c
index d9460934..db7ae8a5 100644
--- a/audiod.c
+++ b/audiod.c
@@ -180,9 +180,12 @@ char *get_time_string(int slot_num)
 {
 	int ret, seconds = 0, length;
 	struct timeval *tmp, sum, sss, /* server stream start */
+		rstime, /* receiver start time */
+		wstime, /* writer start time */
 		wtime, /* now - writer start */
 		rskip; /* receiver start - sss */
 	struct slot_info *s = slot_num < 0? NULL : &slot[slot_num];
+	char *msg;
 
 	if (audiod_status == AUDIOD_OFF)
 		goto empty;
@@ -194,7 +197,7 @@ char *get_time_string(int slot_num)
 	if (audiod_status == AUDIOD_ON && !s)
 		goto empty;
 	/* valid status items and playing */
-	if (s) { /* writer active in this slot */
+	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 */
@@ -205,15 +208,18 @@ char *get_time_string(int slot_num)
 		tv_diff(tmp, &stat_task->sa_time_diff, &sss);
 	else
 		tv_add(tmp, &stat_task->sa_time_diff, &sss);
-	if (!s) {
+	if (!s || !s->wns) {
 		struct timeval diff;
 		tv_diff(now, &sss, &diff);
 		seconds = diff.tv_sec + stat_task->offset_seconds;
 		goto out;
 	}
-	tv_diff(now, &s->wstime, &wtime);
+	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;
-	ret = tv_diff(&s->rstime, &sss, &rskip);
+	btr_get_node_start(s->receiver_node->btrn, &rstime);
+	ret = tv_diff(&rstime, &sss, &rskip);
 	if (ret > 0) { /* audiod was started in the middle of the stream */
 		tv_add(&wtime, &rskip, &sum);
 		seconds += sum.tv_sec;
@@ -222,7 +228,7 @@ char *get_time_string(int slot_num)
 out:
 	seconds = PARA_MIN(seconds, length);
 	seconds = PARA_MAX(seconds, 0);
-	return make_message(
+	msg = make_message(
 		"%s%d:%02d [%d:%02d] (%d%%/%d:%02d)",
 		s? "" : "~",
 		seconds / 60,
@@ -233,6 +239,8 @@ out:
 		length / 60,
 		length % 60
 	);
+	PARA_DEBUG_LOG("slot %d: %s\n", slot_num, msg);
+	return msg;
 empty:
 	return para_strdup(NULL);
 }
@@ -312,30 +320,73 @@ static void close_receiver(int slot_num)
 	PARA_NOTICE_LOG("closing %s receiver in slot %d\n",
 		audio_formats[s->format], slot_num);
 	a->receiver->close(s->receiver_node);
+	btr_free_node(s->receiver_node->btrn);
 	free(s->receiver_node);
 	s->receiver_node = NULL;
 	stat_task->current_audio_format_num = -1;
 }
 
-static void kill_all_decoders(int error)
+static void writer_cleanup(struct writer_node *wn)
+{
+	struct writer *w = writers + wn->writer_num;
+
+	w->close(wn);
+	btr_free_node(wn->btrn);
+}
+
+static void close_writers(struct slot_info *s)
+{
+	struct audio_format_info *a;
+	int i;
+
+	if (s->format < 0)
+		return;
+	a = afi + s->format;
+	if (a->num_writers == 0)
+		writer_cleanup(s->wns);
+	else {
+		for (i = 0; i < a->num_writers; i++)
+			writer_cleanup(s->wns + i);
+	}
+	free(s->wns);
+	s->wns = NULL;
+}
+
+static void _close_filters(struct slot_info *s)
 {
 	int i;
+	struct audio_format_info *a = afi + s->format;
+	if (a->num_filters == 0)
+		return;
+	for (i = 0; i < a->num_filters; i++) {
+		struct filter_node *fn = s->fns + i;
+		struct filter *f = filters + fn->filter_num;
+
+		f->close(fn);
+		btr_free_node(fn->btrn);
+	}
+	free(s->fns);
+	s->fns = NULL;
+}
+
+static void kill_all_decoders(int error)
+{
+	int i, j;
 
 	FOR_EACH_SLOT(i) {
 		struct slot_info *s = &slot[i];
-		if (s->wng && s->wng->task.error >= 0) {
-			PARA_INFO_LOG("deactivating wng in slot %d\n",
-				i);
-			s->wng->task.error = error;
-		}
-		if (s->fc && s->fc->task.error >= 0) {
-			PARA_INFO_LOG("deactivatimg filter chain in slot %d\n", i);
-			s->fc->task.error = error;
-		}
-		if (s->receiver_node && s->receiver_node->task.error >= 0) {
-			PARA_INFO_LOG("deactivating receiver_node in slot %d\n", i);
+		struct audio_format_info *a;
+		if (s->format < 0)
+			continue;
+		a = afi + s->format;
+		if (s->wns)
+			for (j = 0; j < a->num_writers; j++)
+				s->wns[j].task.error = error;
+		if (s->fns)
+			for (j = 0; j < a->num_writers; j++)
+				s->fns[j].task.error = error;
+		if (s->receiver_node)
 			s->receiver_node->task.error = error;
-		}
 	}
 }
 
@@ -350,7 +401,7 @@ static int get_empty_slot(void)
 			clear_slot(i);
 			return i;
 		}
-		if (s->wng || s->receiver_node || s->fc)
+		if (s->wns || s->receiver_node || s->fns)
 			continue;
 		clear_slot(i);
 		return i;
@@ -372,90 +423,71 @@ int num_filters(int audio_format_num)
 	return afi[audio_format_num].num_filters;
 }
 
-static void open_filters(int slot_num)
+static void open_filters(struct slot_info *s)
 {
-	struct slot_info *s = &slot[slot_num];
-	struct audio_format_info *a = &afi[s->format];
+	struct audio_format_info *a = afi + s->format;
 	struct filter_node *fn;
 	int nf = a->num_filters;
+	struct btr_node *parent;
 	int i;
 
-	s->fc = NULL;
-	if (!nf)
+	if (nf == 0)
 		return;
 	PARA_INFO_LOG("opening %s filters\n", audio_formats[s->format]);
-	s->fc = para_calloc(sizeof(struct filter_chain));
-	s->fc->filter_nodes = para_malloc(nf * sizeof(struct filter_node));
-	s->fc->inbufp = &s->receiver_node->buf;
-	s->fc->in_loaded = &s->receiver_node->loaded;
-	s->fc->input_error = &s->receiver_node->task.error;
-	s->fc->task.pre_select = NULL;
-	s->fc->task.post_select = filter_post_select;
-	s->fc->task.error = 0;
-	s->fc->num_filters = nf;
-
-	s->receiver_node->output_error = &s->fc->task.error;
-	sprintf(s->fc->task.status, "filter chain");
-	FOR_EACH_FILTER_NODE(fn, s->fc, i) {
+	assert(s->fns == NULL);
+	s->fns = para_calloc(nf * sizeof(struct filter_node));
+	parent = s->receiver_node->btrn;
+	for (i = 0; i < nf; i++) {
 		struct filter *f = filters + a->filter_nums[i];
+		fn = s->fns + i;
 		fn->filter_num = a->filter_nums[i];
 		fn->conf = a->filter_conf[i];
-		fn->fc = s->fc;
-		fn->loaded = 0;
-		INIT_LIST_HEAD(&fn->callbacks);
+		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);
 		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, slot_num);
-		s->fc->outbufp = &fn->buf;
-		s->fc->out_loaded = &fn->loaded;
+			audio_formats[s->format], i,  nf, f->name, s - slot);
+		sprintf(fn->task.status, "%s (slot %d)", f->name, s - slot);
 	}
-	register_task(&s->fc->task);
 }
 
-static void open_writers(int slot_num)
+static void open_writers(struct slot_info *s)
 {
-	int ret, i;
-	struct slot_info *s = &slot[slot_num];
-	struct audio_format_info *a = &afi[s->format];
-
-	PARA_INFO_LOG("opening %s writers\n", audio_formats[s->format]);
-	if (!a->num_writers)
-		s->wng = setup_default_wng();
-	else
-		s->wng = wng_new(a->num_writers);
-	if (s->fc) {
-		s->wng->bufp = s->fc->outbufp;
-		s->wng->loaded = s->fc->out_loaded;
-		s->wng->input_error = &s->fc->task.error;
-		s->wng->channels = &s->fc->channels;
-		s->wng->samplerate = &s->fc->samplerate;
-		s->fc->output_error = &s->wng->task.error;
-		PARA_INFO_LOG("samplerate: %d\n", *s->wng->samplerate);
-	} else {
-		s->wng->bufp = &s->receiver_node->buf;
-		s->wng->loaded = &s->receiver_node->loaded;
-		s->wng->input_error = &s->receiver_node->task.error;
-	}
-	for (i = 0; i < a->num_writers; i++) {
-		s->wng->writer_nodes[i].conf = a->writer_conf[i];
-		s->wng->writer_nodes[i].writer_num = a->writer_nums[i];
+	int i;
+	struct audio_format_info *a = afi + s->format;
+	struct writer_node *wn;
+	struct btr_node *parent = s->fns[a->num_filters - 1].btrn;
+
+	assert(s->wns == NULL);
+	s->wns = para_calloc(PARA_MAX(1U, a->num_writers)
+		* sizeof(struct writer_node));
+	if (a->num_writers == 0)
+		setup_writer_node(NULL, parent, s->wns);
+	else {
+		PARA_INFO_LOG("opening %s writers\n", audio_formats[s->format]);
+		for (i = 0; i < a->num_writers; i++) {
+			wn = s->wns + i;
+			wn->conf = a->writer_conf[i];
+			wn->writer_num = a->writer_nums[i];
+			register_writer_node(wn, parent);
+		}
 	}
-	ret = wng_open(s->wng);
-	if (ret < 0)
-		return;
-	s->wstime = *now;
 	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;
-	activate_inactive_grab_clients(s->format, s->fc);
 }
 
+/* returns slot num on success */
 static int open_receiver(int format)
 {
 	struct audio_format_info *a = &afi[format];
 	struct slot_info *s;
 	int ret, slot_num;
+	struct receiver *r = a->receiver;
 	struct receiver_node *rn;
 	const struct timeval restart_delay = {2, 0};
 
@@ -467,22 +499,23 @@ static int open_receiver(int format)
 	s->format = format;
 	s->receiver_node = para_calloc(sizeof(struct receiver_node));
 	rn = s->receiver_node;
-	rn->receiver = a->receiver;
+	rn->receiver = r;
 	rn->conf = a->receiver_conf;
-	ret = a->receiver->open(s->receiver_node);
+	rn->btrn = btr_new_node(r->name, NULL, NULL, rn);
+	ret = r->open(rn);
 	if (ret < 0) {
-		free(s->receiver_node);
+		btr_free_node(rn->btrn);
+		free(rn);
 		s->receiver_node = NULL;
 		goto err;
 	}
 	PARA_NOTICE_LOG("started %s: %s receiver in slot %d\n",
-		audio_formats[s->format], a->receiver->name, slot_num);
-	rn->task.pre_select = a->receiver->pre_select;
-	rn->task.post_select = a->receiver->post_select;
-	s->rstime = *now;
-	sprintf(rn->task.status, "%s receiver node", rn->receiver->name);
+		audio_formats[s->format], r->name, slot_num);
+	rn->task.pre_select = r->pre_select;
+	rn->task.post_select = r->post_select;
+	sprintf(rn->task.status, "%s receiver node", r->name);
 	register_task(&rn->task);
-	ret = 1;
+	ret = slot_num;
 err:
 	if (ret < 0)
 		PARA_ERROR_LOG("%s\n", para_strerror(-ret));
@@ -508,19 +541,20 @@ static int receiver_running(int format)
 	return ret;
 }
 
-static void open_current_receiver(struct sched *s)
+/* returns slot num on success. */
+static int open_current_receiver(struct sched *s)
 {
 	struct timeval diff;
 	int ret, cafn = stat_task->current_audio_format_num;
 
 	if (cafn < 0 || !stat_task->ct)
-		return;
+		return -1;
 	/* Do nothing if the 'N' flag is set or the 'P' flag is unset */
 	if (stat_task->vss_status != VSS_STATUS_FLAG_PLAYING)
-		return;
+		return -1;
 	ret = receiver_running(cafn);
 	if (ret > 0) /* already running and not eof */
-		return;
+		return -1;
 	if (ret < 0) { /* eof */
 		/*
 		 * para_server uses a zero start time during the announcement
@@ -528,16 +562,19 @@ static void open_current_receiver(struct sched *s)
 		 * this period begins to avoid restarting the receiver that
 		 * belongs to the file just completed.
 		 */
-		if (stat_task->server_stream_start.tv_sec)
-			return;
+		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;
+		return -1;
 	}
 	/* start a new receiver */
-	open_receiver(cafn);
+	return open_receiver(cafn);
 }
 
 static unsigned compute_time_diff(const struct timeval *status_time)
@@ -906,9 +943,9 @@ static void command_post_select(struct sched *s, struct task *t)
 	int ret;
 	struct command_task *ct = container_of(t, struct command_task, task);
 	static struct timeval last_status_dump;
-	struct timeval tmp;
+	struct timeval tmp, delay = {0, 500 * 1000};
 
-	tv_add(&last_status_dump, &(struct timeval){0, 500 * 1000}, &tmp);
+	tv_add(&last_status_dump, &delay, &tmp);
 	if (tv_diff(&tmp, now, NULL) < 0) {
 		audiod_status_dump();
 		last_status_dump = *now;
@@ -975,19 +1012,22 @@ static void set_stat_task_restart_barrier(unsigned seconds)
 static void try_to_close_slot(int slot_num)
 {
 	struct slot_info *s = &slot[slot_num];
+	struct audio_format_info *a = afi + s->format;
+	int i;
 
 	if (s->format < 0)
 		return;
 	if (s->receiver_node && s->receiver_node->task.error != -E_TASK_UNREGISTERED)
 		return;
-	if (s->fc && s->fc->task.error != -E_TASK_UNREGISTERED)
-		return;
-	if (s->wng && s->wng->task.error != -E_TASK_UNREGISTERED)
-		return;
+	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)
+			return;
 	PARA_INFO_LOG("closing slot %d\n", slot_num);
-	wng_close(s->wng);
-	close_filters(s->fc);
-	free(s->fc);
+	close_writers(s);
+	_close_filters(s);
 	close_receiver(slot_num);
 	clear_slot(slot_num);
 }
@@ -998,50 +1038,28 @@ static void try_to_close_slot(int slot_num)
  */
 static void start_stop_decoders(struct sched *s)
 {
-	int i;
+	int i, ret;
+	struct slot_info *sl;
+	struct audio_format_info *a;
 
 	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);
-	open_current_receiver(s);
-	FOR_EACH_SLOT(i) {
-		struct slot_info *sl = &slot[i];
-		struct audio_format_info *a;
-		struct timeval diff;
-
-		if (sl->format < 0)
-			continue;
-		a = &afi[sl->format];
-		if (!sl->receiver_node)
-			continue;
-		if ((!a->num_filters || sl->fc) && sl->wng)
-			continue; /* everything already started */
-		if (!a->num_filters) {
-			if (sl->receiver_node->loaded && !sl->wng) {
-				open_writers(i);
-			}
-			continue;
-		}
-		if (sl->receiver_node->loaded && !sl->fc) {
-			open_filters(i);
-			continue;
-		}
-		if (sl->wng || !sl->fc || !*sl->fc->out_loaded)
-			continue;
-		if (tv_diff(now, &initial_delay_barrier, &diff) > 0) {
-			open_writers(i);
-			continue;
-		}
-		PARA_INFO_LOG("initial delay: %lu ms left\n", tv2ms(&diff));
-		if (tv_diff(&s->timeout, &diff, NULL) > 0) {
-			s->timeout = diff;
-		}
-	}
+//	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 (ret < 0)
+		return;
+	sl = slot + ret;
+	a = afi + sl->format;
+	if (a->num_filters)
+		open_filters(sl);
+	open_writers(sl);
+	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)
 {
@@ -1240,7 +1258,7 @@ int main(int argc, char *argv[])
 	register_task(&sig_task->task);
 	register_task(&cmd_task->task);
 	register_task(&stat_task->task);
-	s.default_timeout.tv_sec = 0;
+	s.default_timeout.tv_sec = 2;
 	s.default_timeout.tv_usec = 999 * 1000;
 	ret = schedule(&s);
 
diff --git a/audiod.h b/audiod.h
index 2d8fec1a..90ca22c4 100644
--- a/audiod.h
+++ b/audiod.h
@@ -51,10 +51,6 @@ struct audiod_command {
 struct slot_info {
 	/** Number of the audio format in this slot. */
 	int format;
-	/** Receiver start time. */
-	struct timeval rstime;
-	/** Writer start time. */
-	struct timeval wstime;
 	/** The stream_start status item announced by para_server.  */
 	struct timeval server_stream_start;
 	/** The offset status item announced by para_server. */
@@ -65,8 +61,10 @@ struct slot_info {
 	struct receiver_node *receiver_node;
 	/** The active filter chain. */
 	struct filter_chain *fc;
-	/** The active writer node group. */
-	struct writer_node_group *wng;
+	/** The array of filter nodes. */
+	struct filter_node *fns;
+	/** The array of writers attached to the last filter. */
+	struct writer_node *wns;
 };
 
 extern struct slot_info slot[MAX_STREAM_SLOTS];
diff --git a/audiod_command.c b/audiod_command.c
index b790fc15..a5f1bde6 100644
--- a/audiod_command.c
+++ b/audiod_command.c
@@ -24,6 +24,7 @@
 #include "net.h"
 #include "daemon.h"
 #include "string.h"
+#include "write.h"
 #include "fd.h"
 #include "audiod_command_list.h"
 
@@ -192,18 +193,22 @@ __malloc static char *audiod_status_string(void)
 
 static int get_play_time_slot_num(void)
 {
-	int i, oldest = -1;
+	int i, oldest_slot = -1;
+	struct timeval oldest_wstime = {0, 0};
 
 	FOR_EACH_SLOT(i) {
 		struct slot_info *s = &slot[i];
-		if (!s->wng)
+		struct timeval wstime;
+		if (!s->wns)
 			continue;
-		if (oldest >= 0 && tv_diff(&s->wstime, &slot[oldest].wstime,
-				NULL) > 0)
+		btr_get_node_start(s->wns[0].btrn, &wstime);
+		if (oldest_slot >= 0 && tv_diff(&wstime, &oldest_wstime, NULL) > 0)
 			continue;
-		oldest = i;
+		oldest_wstime = wstime;
+		oldest_slot = i;
 	}
-	return oldest;
+	//PARA_CRIT_LOG("oldest slot: %d\n", oldest_slot);
+	return oldest_slot;
 }
 
 __malloc static char *decoder_flags(void)
@@ -216,9 +221,9 @@ __malloc static char *decoder_flags(void)
 		char flag = '0';
 		if (s->receiver_node)
 			flag += 1;
-		if (s->fc)
+		if (s->fns)
 			flag += 2;
-		if (s->wng)
+		if (s->wns)
 			flag += 4;
 		flags[i] = flag;
 	}
diff --git a/buffer_tree.c b/buffer_tree.c
index 23599b83..1be2037f 100644
--- a/buffer_tree.c
+++ b/buffer_tree.c
@@ -6,6 +6,7 @@
 #include "string.h"
 #include "buffer_tree.h"
 #include "error.h"
+#include "sched.h"
 
 
 struct btr_buffer {
@@ -29,6 +30,8 @@ struct btr_node {
 	struct list_head node;
 	/* The children nodes of this btr node are linked together in a list. */
 	struct list_head children;
+	/* Time of first data transfer. */
+	struct timeval start;
 	/**
 	 * The input queue is a list of references to btr buffers. Each item on
 	 * the list represents an input buffer which has not been completely
@@ -58,6 +61,8 @@ struct btr_node *btr_new_node(const char *name, struct btr_node *parent,
 	btrn->parent = parent;
 	btrn->execute = handler;
 	btrn->context = context;
+	btrn->start.tv_sec = 0;
+	btrn->start.tv_usec = 0;
 	if (parent)
 		list_add_tail(&btrn->node, &parent->children);
 	INIT_LIST_HEAD(&btrn->children);
@@ -106,12 +111,16 @@ static void add_btrb_to_children(struct btr_buffer *btrb,
 {
 	struct btr_node *ch;
 
+	if (btrn->start.tv_sec == 0)
+		btrn->start = *now;
 	FOR_EACH_CHILD(ch, btrn) {
 		struct btr_buffer_reference *br = para_malloc(sizeof(*br));
 		br->btrb = btrb;
 		br->consumed = consumed;
 		list_add_tail(&br->node, &ch->input_queue);
 		btrb->refcount++;
+		if (ch->start.tv_sec == 0)
+			ch->start = *now;
 	}
 }
 
@@ -353,8 +362,8 @@ static int merge_input(struct btr_node *btrn)
 	}
 	/* make a new btrb that combines the two buffers and a br to it. */
 	sz = szs[0] + szs[1];
-	//PARA_CRIT_LOG("merging input buffers: (%zu, %zu) -> %zu\n",
-	//	szs[0], szs[1], sz);
+	PARA_DEBUG_LOG("merging input buffers: (%zu, %zu) -> %zu\n",
+		szs[0], szs[1], sz);
 	buf = para_malloc(sz);
 	/* TODO: Avoid this memcopy by introducing btr buffer pool. */
 	memcpy(buf, bufs[0], szs[0]);
@@ -379,6 +388,7 @@ void btr_merge(struct btr_node *btrn, size_t dest_size)
 		size_t len = btr_next_buffer(btrn, &buf);
 		if (len >= dest_size)
 			return;
+		PARA_DEBUG_LOG("input size = %zu < %zu = dest\n", len, dest_size);
 		if (merge_input(btrn) < 2)
 			return;
 	}
@@ -434,3 +444,8 @@ int btr_node_status(struct btr_node *btrn, size_t min_iqs,
 	}
 	return 1;
 }
+
+void btr_get_node_start(struct btr_node *btrn, struct timeval *tv)
+{
+	*tv = btrn->start;
+}
diff --git a/buffer_tree.h b/buffer_tree.h
index 90399686..9de99268 100644
--- a/buffer_tree.h
+++ b/buffer_tree.h
@@ -33,3 +33,4 @@ int btr_pushdown_one(struct btr_node *btrn);
 bool btr_inplace_ok(struct btr_node *btrn);
 int btr_node_status(struct btr_node *btrn, size_t min_iqs,
 		enum btr_node_type type);
+void btr_get_node_start(struct btr_node *btrn, struct timeval *tv);
diff --git a/compress_filter.c b/compress_filter.c
index a0b53bde..39055618 100644
--- a/compress_filter.c
+++ b/compress_filter.c
@@ -188,6 +188,11 @@ static void open_compress(struct filter_node *fn)
 	pcd->max_gain = 1 << (pcd->conf->inertia_arg + pcd->conf->aggressiveness_arg);
 }
 
+static void compress_free_config(void *conf)
+{
+	compress_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the compress filter.
  *
@@ -204,6 +209,7 @@ void compress_filter_init(struct filter *f)
 	f->pre_select = generic_filter_pre_select;
 	f->post_select = compress_post_select;
 	f->parse_config = compress_parse_config;
+	f->free_config = compress_free_config;
 	f->help = (struct ggo_help) {
 		.short_help = compress_filter_args_info_help,
 		.detailed_help = compress_filter_args_info_detailed_help
diff --git a/dccp_recv.c b/dccp_recv.c
index 58c969a2..4e1860e3 100644
--- a/dccp_recv.c
+++ b/dccp_recv.c
@@ -136,6 +136,11 @@ static void dccp_recv_post_select(struct sched *s, struct task *t)
 		t->error = -E_RECV_EOF;
 }
 
+static void dccp_recv_free_config(void *conf)
+{
+	dccp_recv_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the dccp receiver.
  *
@@ -154,6 +159,7 @@ void dccp_recv_init(struct receiver *r)
 	r->pre_select = dccp_recv_pre_select;
 	r->post_select = dccp_recv_post_select;
 	r->parse_config = dccp_recv_parse_config;
+	r->free_config = dccp_recv_free_config;
 	r->help = (struct ggo_help) {
 		.short_help = dccp_recv_args_info_help,
 		.detailed_help = dccp_recv_args_info_detailed_help
diff --git a/file_write.c b/file_write.c
index 4c043864..7a24b3ac 100644
--- a/file_write.c
+++ b/file_write.c
@@ -182,6 +182,11 @@ __malloc static void *file_write_parse_config(const char *options)
 	return NULL;
 }
 
+static void file_write_free_config(void *conf)
+{
+	file_cmdline_parser_free(conf);
+}
+
 /** the init function of the file writer */
 void file_write_init(struct writer *w)
 {
@@ -194,6 +199,7 @@ void file_write_init(struct writer *w)
 	w->post_select = file_write_post_select;
 	w->post_select_btr = file_write_post_select_btr;
 	w->parse_config = file_write_parse_config;
+	w->free_config = file_write_free_config;
 	w->close = file_write_close;
 	w->shutdown = NULL; /* nothing to do */
 	w->help = (struct ggo_help) {
diff --git a/filter.c b/filter.c
index e7fc2cd7..599aaad9 100644
--- a/filter.c
+++ b/filter.c
@@ -154,7 +154,7 @@ static int parse_config(int argc, char *argv[])
 	return 1;
 }
 
-/* TODO: support more than one filter, actually parse options */
+/* TODO: support buffer tree, not just a chain. */
 static int __noreturn main_btr(void)
 {
 	static struct sched s;
diff --git a/filter.h b/filter.h
index ae0a11e3..4a17c10a 100644
--- a/filter.h
+++ b/filter.h
@@ -191,6 +191,7 @@ struct filter {
 	 * argv. On failure, a negative paraslash error code must be returned.
 	 */
 	int (*parse_config)(int argc, char **argv, void **config);
+	void (*free_config)(void *conf);
 
 	/** The help texts for this filter. */
 	struct ggo_help help;
diff --git a/http_recv.c b/http_recv.c
index b3b62103..ed362bfa 100644
--- a/http_recv.c
+++ b/http_recv.c
@@ -103,7 +103,7 @@ static void http_recv_pre_select(struct sched *s, struct task *t)
 		para_fd_set(phd->fd, &s->rfds, &s->max_fileno);
 }
 
-#define HTTP_RECV_READ_BUF_SIZE 4096
+#define HTTP_RECV_READ_BUF_SIZE 16384
 
 static void http_recv_post_select(struct sched *s, struct task *t)
 {
@@ -186,7 +186,6 @@ static void http_recv_close(struct receiver_node *rn)
 	close(phd->fd);
 	free(rn->buf);
 	free(rn->private_data);
-	http_recv_cmdline_parser_free(rn->conf);
 }
 
 static void *http_recv_parse_config(int argc, char **argv)
@@ -221,12 +220,17 @@ static int http_recv_open(struct receiver_node *rn)
 	return 1;
 }
 
+static void http_recv_free_config(void *conf)
+{
+	http_recv_cmdline_parser_free(conf);
+}
+
 /**
- * the init function of the http receiver
+ * The init function of the http receiver.
  *
- * \param r pointer to the receiver struct to initialize
+ * \param r Pointer to the receiver struct to initialize.
  *
- * Just initialize all function pointers of \a r.
+ * This initializes all function pointers of \a r.
  */
 void http_recv_init(struct receiver *r)
 {
@@ -239,6 +243,7 @@ void http_recv_init(struct receiver *r)
 	r->post_select = http_recv_post_select;
 	r->shutdown = http_shutdown;
 	r->parse_config = http_recv_parse_config;
+	r->free_config = http_recv_free_config;
 	r->help = (struct ggo_help) {
 		.short_help = http_recv_args_info_help,
 		.detailed_help = http_recv_args_info_detailed_help
diff --git a/mp3dec_filter.c b/mp3dec_filter.c
index 397a2420..2472b2cf 100644
--- a/mp3dec_filter.c
+++ b/mp3dec_filter.c
@@ -179,7 +179,6 @@ static void mp3dec_close(struct filter_node *fn)
 
 	free(fn->buf);
 	fn->buf = NULL;
-	mp3dec_cmdline_parser_free(fn->conf);
 	free(pmd);
 	fn->private_data = NULL;
 }
@@ -318,6 +317,11 @@ static int mp3dec_execute(struct btr_node *btrn, const char *cmd, char **result)
 	}
 	return -ERRNO_TO_PARA_ERROR(ENOTSUP);
 }
+
+static void mp3dec_free_config(void *conf)
+{
+	mp3dec_cmdline_parser_free(conf);
+}
 /**
  * The init function of the mp3dec filter.
  *
@@ -334,6 +338,7 @@ void mp3dec_filter_init(struct filter *f)
 	f->convert = mp3dec;
 	f->close = mp3dec_close;
 	f->parse_config = mp3dec_parse_config;
+	f->free_config = mp3dec_free_config;
 	f->pre_select = generic_filter_pre_select;
 	f->post_select = mp3dec_post_select;
 	f->execute = mp3dec_execute;
diff --git a/oggdec_filter.c b/oggdec_filter.c
index 2c9d25f2..511e6405 100644
--- a/oggdec_filter.c
+++ b/oggdec_filter.c
@@ -384,6 +384,11 @@ err:
 	return ret;
 }
 
+static void oggdec_free_config(void *conf)
+{
+	oggdec_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the ogg vorbis decoder.
  *
@@ -400,6 +405,7 @@ void oggdec_filter_init(struct filter *f)
 	f->pre_select = generic_filter_pre_select;
 	f->post_select = ogg_post_select;
 	f->parse_config = oggdec_parse_config;
+	f->free_config = oggdec_free_config;
 	f->execute = oggdec_execute;
 	f->help = (struct ggo_help) {
 		.short_help = oggdec_filter_args_info_help,
diff --git a/oss_write.c b/oss_write.c
index 20b05009..4e5f9b56 100644
--- a/oss_write.c
+++ b/oss_write.c
@@ -282,6 +282,11 @@ err_out:
 	return NULL;
 }
 
+static void oss_free_config(void *conf)
+{
+	oss_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the oss writer.
  *
@@ -301,6 +306,7 @@ void oss_write_init(struct writer *w)
 	w->post_select = oss_post_select;
 	w->post_select_btr = oss_post_select_btr;
 	w->parse_config = oss_parse_config;
+	w->free_config = oss_free_config;
 	w->shutdown = NULL;
 	w->help = (struct ggo_help) {
 		.short_help = oss_write_args_info_help,
diff --git a/prebuffer_filter.c b/prebuffer_filter.c
index adfe9de9..207d3324 100644
--- a/prebuffer_filter.c
+++ b/prebuffer_filter.c
@@ -162,6 +162,11 @@ static void prebuffer_open(struct filter_node *fn)
 	fn->buf = para_malloc(fn->bufsize);
 }
 
+static void prebuffer_free_config(void *conf)
+{
+	prebuffer_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the prebuffer filter.
  *
@@ -176,6 +181,7 @@ void prebuffer_filter_init(struct filter *f)
 	f->close = prebuffer_close;
 	f->convert = prebuffer_convert;
 	f->parse_config = prebuffer_parse_config;
+	f->free_config = prebuffer_free_config;
 	f->pre_select = prebuffer_pre_select;
 	f->post_select = prebuffer_post_select;
 	f->help = (struct ggo_help) {
diff --git a/recv.h b/recv.h
index eb6175c2..3f13f108 100644
--- a/recv.h
+++ b/recv.h
@@ -56,6 +56,7 @@ struct receiver {
 	 * \a argc and \a argv.
 	 */
 	void *(*parse_config)(int argc, char **argv);
+	void (*free_config)(void *conf);
 	/**
 	 * Open one instance of the receiver.
 	 *
diff --git a/udp_recv.c b/udp_recv.c
index bb166c51..f68a7100 100644
--- a/udp_recv.c
+++ b/udp_recv.c
@@ -18,7 +18,6 @@
 #include "ggo.h"
 #include "recv.h"
 #include "udp_recv.cmdline.h"
-#include "audiod.h"
 #include "string.h"
 #include "net.h"
 #include "fd.h"
@@ -273,6 +272,11 @@ err:
 	return ret;
 }
 
+static void udp_recv_free_config(void *conf)
+{
+	udp_recv_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the udp receiver.
  *
@@ -291,6 +295,7 @@ void udp_recv_init(struct receiver *r)
 	r->pre_select = udp_recv_pre_select;
 	r->post_select = udp_recv_post_select;
 	r->parse_config = udp_recv_parse_config;
+	r->free_config = udp_recv_free_config;
 	r->help = (struct ggo_help) {
 		.short_help = udp_recv_args_info_help,
 		.detailed_help = udp_recv_args_info_detailed_help
diff --git a/wmadec_filter.c b/wmadec_filter.c
index 00bdd6eb..49f69b54 100644
--- a/wmadec_filter.c
+++ b/wmadec_filter.c
@@ -1265,7 +1265,7 @@ next_buffer:
 	for (;;) {
 		char *out;
 		int out_size = WMA_OUTPUT_BUFFER_SIZE;
-		if (converted + WMA_FRAME_SKIP + pwd->ahi.block_align > len)
+		if (converted + fn->min_iqs > len)
 			break;
 		out = para_malloc(WMA_OUTPUT_BUFFER_SIZE);
 		ret = wma_decode_superframe(pwd, out,
diff --git a/write.c b/write.c
index 4d06e804..c64ecb63 100644
--- a/write.c
+++ b/write.c
@@ -282,7 +282,7 @@ static int main_btr(struct sched *s)
 {
 	int i, ret;
 	struct check_wav_task_btr _cwt, *cwt = &_cwt;
-	struct writer_node **wns;
+	struct writer_node *wns;
 
 	loglevel = get_loglevel_by_name(conf.loglevel_arg);
 	sit.btrn = btr_new_node("stdin", NULL /* stdin has no parent */, NULL, NULL);
@@ -300,17 +300,17 @@ static int main_btr(struct sched *s)
 	ret = -E_WRITE_SYNTAX;
 	if (!conf.writer_given) {
 		i = 0;
-		wns = para_malloc(sizeof(*wns));
-		wns[0] = setup_writer_node(NULL, cwt->btrn);
-		if (!wns[0])
+		wns = para_calloc(sizeof(*wns));
+		ret = setup_writer_node(NULL, cwt->btrn, wns);
+		if (ret < 0)
 			goto out;
 		i = 1;
 	} else {
-		wns = para_malloc(conf.writer_given * sizeof(*wns));
+		wns = para_calloc(conf.writer_given * sizeof(*wns));
 		for (i = 0; i < conf.writer_given; i++) {
-			wns[i] = setup_writer_node(conf.writer_arg[i],
-				cwt->btrn);
-			if (!wns[i])
+			ret = setup_writer_node(conf.writer_arg[i],
+				cwt->btrn, wns + i);
+			if (ret < 0)
 				goto out;
 		}
 	}
@@ -320,7 +320,7 @@ static int main_btr(struct sched *s)
 	ret = schedule(s);
 out:
 	for (i--; i >= 0; i--) {
-		struct writer_node *wn = wns[i];
+		struct writer_node *wn = wns + i;
 		struct writer *w = writers + wn->writer_num;
 
 		w->close(wn);
diff --git a/write.h b/write.h
index 49e56c38..8e244517 100644
--- a/write.h
+++ b/write.h
@@ -46,6 +46,7 @@ struct writer {
 	 * more than once with different values of \a options.
 	 */
 	void *(*parse_config)(const char *options);
+	void (*free_config)(void *conf);
 	/**
 	 * Open one instance of this writer.
 	 *
diff --git a/write_common.c b/write_common.c
index 9af8d682..8c33b39e 100644
--- a/write_common.c
+++ b/write_common.c
@@ -237,6 +237,21 @@ struct writer_node_group *setup_default_wng(void)
 	return wng;
 }
 
+void register_writer_node(struct writer_node *wn, struct btr_node *parent)
+{
+	struct writer *w = writers + wn->writer_num;
+	char *name = make_message("%s writer", writer_names[wn->writer_num]);
+	int ret;
+
+	wn->btrn = btr_new_node(name, parent, w->execute, wn);
+	strcpy(wn->task.status, name);
+	free(name);
+	ret = w->open(wn);
+	wn->task.post_select = w->post_select_btr;
+	wn->task.pre_select = w->pre_select_btr;
+	register_task(&wn->task);
+}
+
 /**
  * Setup a writer node with the default writer.
  *
@@ -253,32 +268,19 @@ struct writer_node_group *setup_default_wng(void)
  *
  * \return A pointer to the allocated writer node group.
  */
-struct writer_node *setup_writer_node(const char *arg, struct btr_node *parent)
+int setup_writer_node(const char *arg, struct btr_node *parent,
+		struct writer_node *wn)
 {
-	struct writer_node *wn = para_calloc(sizeof(*wn));
-	struct writer *w;
-	char *name;
-
 	if (arg)
 		wn->conf = check_writer_arg(arg, &wn->writer_num);
 	else {
 		wn->writer_num = DEFAULT_WRITER;
 		wn->conf = writers[DEFAULT_WRITER].parse_config("");
 	}
-	if (!wn->conf) {
-		free(wn);
-		return NULL;
-	}
-	w = writers + wn->writer_num;
-	name = make_message("%s writer", writer_names[wn->writer_num]);
-	wn->btrn = btr_new_node(name, parent, w->execute, wn);
-	strcpy(wn->task.status, name);
-	free(name);
-	w->open(wn);
-	wn->task.post_select = w->post_select_btr;
-	wn->task.pre_select = w->pre_select_btr;
-	register_task(&wn->task);
-	return wn;
+	if (!wn->conf)
+		return -E_WRITE_COMMON_SYNTAX;
+	register_writer_node(wn, parent);
+	return 1;
 }
 
 
diff --git a/write_common.h b/write_common.h
index ef5a7c03..8c33a518 100644
--- a/write_common.h
+++ b/write_common.h
@@ -13,6 +13,8 @@ void writer_init(void);
 void *check_writer_arg(const char *wa, int *writer_num);
 struct writer_node_group *setup_default_wng(void);
 void print_writer_helps(int detailed);
-struct writer_node *setup_writer_node(const char *arg, struct btr_node *parent);
+void register_writer_node(struct writer_node *wn, struct btr_node *parent);
+int setup_writer_node(const char *arg, struct btr_node *parent,
+		struct writer_node *wn);
 int get_btr_samplerate(struct btr_node *btrn, int32_t *result);
 int get_btr_channels(struct btr_node *btrn, int32_t *result);