X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=audiod.c;h=d635fbfc334316b9dd78ab2b67e266dd757c828c;hp=dccc2e15ba92bbbeb68f5acdb683270f9e25b482;hb=2a8029478dfc65f0c935d864faf4aea9e3deb27d;hpb=14f817d233309fe9ca2580fa8e4b78fd823b9ab9

diff --git a/audiod.c b/audiod.c
index dccc2e15..d635fbfc 100644
--- a/audiod.c
+++ b/audiod.c
@@ -18,7 +18,6 @@
 
 /** \file audiod.c the paraslash's audio daemon */
 
-#include <sys/time.h> /* gettimeofday */
 #include "para.h"
 
 #include "audiod.cmdline.h"
@@ -36,6 +35,8 @@
 #include "daemon.h"
 #include "string.h"
 #include "fd.h"
+#include "write.h"
+#include "write_common.h"
 
 /** define the array of error lists needed by para_audiod */
 INIT_AUDIOD_ERRLISTS;
@@ -53,48 +54,43 @@ enum {AUDIOD_OFF, AUDIOD_ON, AUDIOD_STANDBY};
 
 /** defines how to handle one supported audio format */
 struct audio_format_info {
-/** pointer to the receiver for this audio format */
+	/** pointer to the receiver for this audio format */
 	struct receiver *receiver;
-/** the receiver configuration */
+	/** the receiver configuration */
 	void *receiver_conf;
-/** the number of filters that should be activated for this audio format */
+	/** the number of filters that should be activated for this audio format */
 	unsigned int num_filters;
-/** pointer to the array of filters to be activated */
+	/** pointer to the array of filters to be activated */
 	struct filter **filters;
-/** pointer to the array of filter configurations */
+	/** pointer to the array of filter configurations */
 	void **filter_conf;
-/** output of the last filter is written to stdin of this command */
-	char *write_cmd;
-/** do not start receiver/filters/writer before this time */
+	/** the number of filters that should be activated for this audio format */
+	unsigned int num_writers;
+	/** pointer to the array of writers to be activated */
+	struct writer **writers;
+	/** pointer to the array of writer configurations */
+	void **writer_conf;
+	/** do not start receiver/filters/writer before this time */
 	struct timeval restart_barrier;
 };
 
 /**
  * describes one instance of a receiver-filter-writer chain
  *
- * \sa receier_node, receiver, filter, filter_node, filter_chain_info
+ * \sa receier_node, receiver, filter, filter_node, filter_chain, writer,
+ * writer_node, writer_node_group.
   */
 struct slot_info {
 	/** number of the audio format in this slot */
 	int format;
-	/** the file descriptor of the writer */
-	int write_fd;
-	/** the process id of the writer */
-	pid_t wpid;
-	/** time of the last successful read from the receiver */
-	struct timeval rtime;
-	/** time the last write to the write fd happend */
-	struct timeval wtime;
 	/** writer start time */
 	struct timeval wstime;
-	/** did we include \a write_fd in the fdset */
-	int  wcheck;
-	/** set to one if we have sent the TERM signal to \a wpid */
-	int wkilled;
 	/** the receiver info associated with this slot */
 	struct receiver_node *receiver_node;
 	/** the active filter chain */
 	struct filter_chain *fc;
+	/** the active writer node group */
+	struct writer_node_group *wng;
 };
 static struct slot_info slot[MAX_STREAM_SLOTS];
 
@@ -110,29 +106,39 @@ static char *af_status, /* the audio format announced in server status */
 static char *stat_item_values[NUM_STAT_ITEMS];
 static FILE *logfile;
 static const struct timeval restart_delay = {0, 300 * 1000};
-
 static struct audio_format_info afi[NUM_AUDIO_FORMATS];
+static struct timeval *now;
 
 static struct signal_task signal_task_struct, *sig_task = &signal_task_struct;
 
+/**
+ * the task for handling audiod commands
+ *
+ * \sa struct task, struct sched
+ */
 struct command_task {
+	/** the local listening socket */
 	int fd;
+	/** the associated task structure */
 	struct task task;
 };
-static struct command_task command_task_struct, *cmd_task = &command_task_struct;
 
+/**
+ * the task for audiod's child (para_client stat)
+ *
+ * \sa struct task, struct sched
+ */
 struct status_task {
+	/** the output of the stat command is read from this fd */
 	int fd;
-	struct task task;
+	/** stat data is stored here */
 	char buf[STRINGSIZE];
+	/** number of bytes loaded in \a buf */
 	unsigned loaded;
-};
-static struct status_task status_task_struct, *stat_task = &status_task_struct;
-
-struct audiod_task {
+	/** the associated task structure */
 	struct task task;
 };
-static struct audiod_task audiod_task_struct, *at = &audiod_task_struct;
+static struct status_task status_task_struct, *stat_task = &status_task_struct;
 
 struct signal_task {
 	int fd;
@@ -140,13 +146,17 @@ struct signal_task {
 	struct task task;
 };
 
-
 /** defines one command of para_audiod */
 struct audiod_command {
 	/** the name of the command */
 	const char *name;
 	/** pointer to the function that handles the command */
 	int (*handler)(int, int, char**);
+	/**
+	 * if the command prefers to handle the full line (rather than the usual
+	 * argv[] array), it stores a pointer to the corresponding line handling
+	 * function here. In this case, the above \a handler pointer must be NULL.
+	 */
 	int (*line_handler)(int, char*);
 	/** one-line description of the command */
 	const char *description;
@@ -314,7 +324,7 @@ static int client_write(int fd, const char *buf)
 
 static char *get_time_string(struct timeval *newest_stime)
 {
-	struct timeval now, diff, adj_stream_start, tmp;
+	struct timeval diff, adj_stream_start, tmp;
 	int total = 0, use_server_time = 1;
 
 	if (!playing) {
@@ -338,8 +348,7 @@ static char *get_time_string(struct timeval *newest_stime)
 			use_server_time = 0;
 		}
 	}
-	gettimeofday(&now, NULL);
-	tv_diff(&now, &tmp, &diff);
+	tv_diff(now, &tmp, &diff);
 	total = diff.tv_sec + offset_seconds;
 	if (total > length_seconds)
 		total = length_seconds;
@@ -374,7 +383,7 @@ static struct timeval *wstime(void)
 	struct timeval *max = NULL;
 	FOR_EACH_SLOT(i) {
 		struct slot_info *s = &slot[i];
-		if (s->wpid <= 0)
+		if (!s->wng)
 			continue;
 		if (max && tv_diff(&s->wstime, max, NULL) <= 0)
 			continue;
@@ -392,7 +401,7 @@ __malloc static char *decoder_flags(void)
 		char flag = '0';
 		if (s->receiver_node)
 			flag += 1;
-		if (s->wpid > 0)
+		if (s->wng)
 			flag += 2;
 		decoder_flags[i] = flag;
 	}
@@ -465,29 +474,6 @@ static void clear_slot(int slot_num)
 	s->format = -1;
 }
 
-static void kill_stream_writer(int slot_num)
-{
-	struct slot_info *s = &slot[slot_num];
-
-	if (s->format < 0 || s->wkilled || s->wpid <= 0)
-		return;
-	PARA_DEBUG_LOG("kill -TERM %d (%s stream writer in slot %d)\n",
-		s->wpid, audio_formats[s->format], slot_num);
-	kill(s->wpid, SIGTERM);
-	s->wkilled = 1;
-}
-
-static void set_restart_barrier(int format, struct timeval *now)
-{
-	struct timeval tmp;
-
-	if (now)
-		tmp = *now;
-	else
-		gettimeofday(&tmp, NULL);
-	tv_add(&tmp, &restart_delay, &afi[format].restart_barrier);
-}
-
 static void close_receiver(int slot_num)
 {
 	struct slot_info *s = &slot[slot_num];
@@ -496,55 +482,26 @@ static void close_receiver(int slot_num)
 	if (s->format < 0 || !s->receiver_node)
 		return;
 	a = &afi[s->format];
-	PARA_NOTICE_LOG("closing %s recevier in slot %d (eof = %d)\n",
+	PARA_NOTICE_LOG("closing %s receiver in slot %d (eof = %d)\n",
 		audio_formats[s->format] , slot_num, s->receiver_node->eof);
 	if (!s->receiver_node->eof)
 		unregister_task(&s->receiver_node->task);
 	a->receiver->close(s->receiver_node);
 	free(s->receiver_node);
 	s->receiver_node = NULL;
-	set_restart_barrier(s->format, NULL);
+	/* set restart barrier */
+	tv_add(now, &restart_delay, &afi[s->format].restart_barrier);
 }
 
 static void kill_all_decoders(void)
 {
 	int i;
 
-	FOR_EACH_SLOT(i)
-		if (slot[i].format >= 0) {
-			PARA_INFO_LOG("stopping decoder in slot %d\n", i);
-			kill_stream_writer(i);
-		}
-}
-
-static void check_sigchld(void)
-{
-	pid_t pid;
-	int i;
-	struct timeval now;
-	gettimeofday(&now, NULL);
-
-reap_next_child:
-	pid = para_reap_child();
-	if (pid <= 0)
-		return;
 	FOR_EACH_SLOT(i) {
 		struct slot_info *s = &slot[i];
-		long lifetime;
-		if (s->format < 0)
-			continue;
-		if (pid == s->wpid) {
-			s->wpid = -1;
-			lifetime = now.tv_sec - s->wstime.tv_sec;
-			PARA_INFO_LOG("%s stream writer in slot %d died "
-				"after %li secs\n",
-				audio_formats[s->format], i, lifetime);
-			set_restart_barrier(s->format, &now);
-			goto reap_next_child;
-		}
+		if (s->receiver_node)
+			s->receiver_node->eof = 1;
 	}
-	PARA_CRIT_LOG("para_client died (pid %d)\n", pid);
-	goto reap_next_child;
 }
 
 static int get_empty_slot(void)
@@ -558,11 +515,7 @@ static int get_empty_slot(void)
 			clear_slot(i);
 			return i;
 		}
-		if (s->write_fd > 0 || s->wpid > 0)
-			continue;
-		if (s->receiver_node)
-			continue;
-		if (s->fc)
+		if (s->wng || s->receiver_node || s->fc)
 			continue;
 		clear_slot(i);
 		return i;
@@ -579,7 +532,7 @@ static int decoder_running(int format)
 		s = &slot[i];
 		if (s->format == format && s->receiver_node)
 			ret |= 1;
-		if (s->format == format && s->wpid > 0)
+		if (s->format == format && s->wng)
 			ret |= 2;
 	}
 	return ret;
@@ -605,7 +558,8 @@ static void close_stat_pipe(void)
 	offset_seconds = 0;
 	audiod_status_dump();
 	playing = 0;
-	stat_item_values[SI_STATUS_BAR] = make_message("%s:no connection to para_server\n",
+	stat_item_values[SI_STATUS_BAR] = make_message(
+		"%s:no connection to para_server\n",
 		status_item_list[SI_STATUS_BAR]);
 	stat_client_write(stat_item_values[SI_STATUS_BAR], SI_STATUS_BAR);
 }
@@ -613,7 +567,6 @@ static void close_stat_pipe(void)
 static void __noreturn clean_exit(int status, const char *msg)
 {
 	PARA_EMERG_LOG("%s\n", msg);
-	kill_all_decoders();
 	if (socket_name)
 		unlink(socket_name);
 	if (stat_task->fd >= 0)
@@ -621,38 +574,21 @@ static void __noreturn clean_exit(int status, const char *msg)
 	exit(status);
 }
 
-__malloc static char *glob_cmd(char *cmd)
-{
-	char *ret, *replacement;
-	struct timeval tmp, delay, rss; /* real stream start */
-
-	delay.tv_sec = conf.stream_delay_arg / 1000;
-	delay.tv_usec = (conf.stream_delay_arg % 1000) * 1000;
-//	PARA_INFO_LOG("delay: %lu:%lu\n", delay.tv_sec, delay.tv_usec);
-	if (sa_time_diff_sign < 0)
-		tv_add(&server_stream_start, &sa_time_diff, &rss);
-	else
-		tv_diff(&server_stream_start, &sa_time_diff, &rss);
-	tv_add(&rss, &delay, &tmp);
-	replacement = make_message("%lu:%lu",
-		(long unsigned)tmp.tv_sec,
-		(long unsigned)tmp.tv_usec);
-	ret = s_a_r(cmd, "STREAM_START", replacement);
-	free(replacement);
-	if (!ret)
-		goto out;
-	PARA_INFO_LOG("cmd: %s, repl: %s\n", cmd, ret);
-out:
-	return ret;
-}
-
-/** get the number of filters for the given audio format */
+/**
+ * get the number of filters
+ *
+ * \param audio_format_num the number identifying the audio format
+ *
+ * \return the number of filters for the given audio format
+ *
+ * \sa struct filter;
+ */
 int num_filters(int audio_format_num)
 {
 	return afi[audio_format_num].num_filters;
 }
 
-void filter_event_handler(struct task *t)
+static void filter_event_handler(struct task *t)
 {
 	PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
 	unregister_task(t);
@@ -664,17 +600,16 @@ static void open_filters(int slot_num)
 	struct audio_format_info *a = &afi[s->format];
 	int nf = a->num_filters;
 	int i;
-	static int output_eof; /* FIXME */
 
-	s->fc = para_calloc(sizeof(struct filter_chain));
-	INIT_LIST_HEAD(&s->fc->filters);
+	s->fc = NULL;
 	if (!nf)
 		return;
+	PARA_INFO_LOG("opening %s filters\n", audio_formats[s->format]);
+	s->fc = para_calloc(sizeof(struct filter_chain));
+	INIT_LIST_HEAD(&s->fc->filters);
 	s->fc->inbuf = s->receiver_node->buf;
 	s->fc->in_loaded = &s->receiver_node->loaded;
 	s->fc->input_eof = &s->receiver_node->eof;
-	s->fc->output_eof = &output_eof;
-	output_eof = 0;
 
 	s->fc->task.pre_select = filter_pre_select;
 	s->fc->task.event_handler = filter_event_handler;
@@ -697,8 +632,6 @@ static void open_filters(int slot_num)
 		s->fc->out_loaded = &fn->loaded;
 	}
 	register_task(&s->fc->task);
-//	PARA_DEBUG_LOG("output loaded for filter chain %p: %p\n", s->fc,
-//		s->fc->out_loaded);
 }
 
 static struct filter_node *find_filter_node(int slot_num, int format, int filternum)
@@ -726,40 +659,52 @@ static struct filter_node *find_filter_node(int slot_num, int format, int filter
 	return NULL;
 }
 
-static void start_stream_writer(int slot_num)
+static void wng_event_handler(struct task *t)
+{
+	PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret));
+	unregister_task(t);
+}
+
+static void open_writers(int slot_num)
 {
-	int ret, fds[3] = {1, -1, -1};
+	int ret, i;
 	struct slot_info *s = &slot[slot_num];
 	struct audio_format_info *a = &afi[s->format];
-	char *glob = NULL;
-
-	if (a->write_cmd)
-		glob = glob_cmd(a->write_cmd);
-	if (!glob)
-		glob = para_strdup("para_write -w alsa");
-	PARA_INFO_LOG("starting stream writer: %s\n", glob);
-	open_filters(slot_num);
-	ret = para_exec_cmdline_pid(&s->wpid, glob, fds);
-	free(glob);
-	if (ret < 0) {
-		PARA_ERROR_LOG("exec failed (%d)\n", ret);
-		return;
+
+	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->buf = s->fc->outbuf;
+		s->wng->loaded = s->fc->out_loaded;
+		s->wng->input_eof = &s->fc->eof;
+		s->fc->output_eof = &s->wng->eof;
+	} else {
+		s->wng->buf = s->receiver_node->buf;
+		s->wng->loaded = &s->receiver_node->loaded;
+		s->wng->input_eof = &s->receiver_node->eof;
+	}
+	s->wng->task.event_handler = wng_event_handler;
+	for (i = 0; i < a->num_writers; i++) {
+		s->wng->writer_nodes[i].conf = a->writer_conf[i];
+		s->wng->writer_nodes[i].writer = a->writers[i];
+		sprintf(s->wng->writer_nodes[i].task.status, "writer_node");
 	}
-	s->write_fd = fds[0];
-	add_close_on_fork_list(s->write_fd);
-	/* we write to this fd in do_select, so we need non-blocking */
-	mark_fd_nonblock(s->write_fd);
-	gettimeofday(&s->wstime, NULL);
+	ret = wng_open(s->wng);
+	s->wstime = *now;
 	current_decoder = slot_num;
 	activate_inactive_grab_clients(slot_num, s->format, &s->fc->filters);
 }
 
-void rn_event_handler(struct task *t)
+static void rn_event_handler(struct task *t)
 {
 //	struct receiver_node *rn = t->private_data;
 	PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
 	unregister_task(t);
 }
+
 static void open_receiver(int format)
 {
 	struct audio_format_info *a = &afi[format];
@@ -772,8 +717,6 @@ static void open_receiver(int format)
 		clean_exit(EXIT_FAILURE, PARA_STRERROR(-slot_num));
 	s = &slot[slot_num];
 	s->format = format;
-	gettimeofday(&s->rtime, NULL);
-	s->wtime = s->rtime;
 	s->receiver_node = para_calloc(sizeof(struct receiver_node));
 	rn = s->receiver_node;
 	rn->receiver = a->receiver;
@@ -789,7 +732,6 @@ static void open_receiver(int format)
 	PARA_NOTICE_LOG("started %s: %s receiver in slot %d\n",
 		audio_formats[s->format], a->receiver->name, slot_num);
 	rn->task.private_data = s->receiver_node;
-	PARA_NOTICE_LOG("rn = %p\n", rn->task.private_data);
 	rn->task.pre_select = a->receiver->pre_select;
 	rn->task.post_select = a->receiver->post_select;
 	rn->task.event_handler = rn_event_handler;
@@ -800,14 +742,12 @@ static void open_receiver(int format)
 
 static int is_frozen(int format)
 {
-	struct timeval now;
 	struct audio_format_info *a = &afi[format];
 
-	gettimeofday(&now, NULL);
-	return (tv_diff(&now, &a->restart_barrier, NULL) > 0)? 0 : 1;
+	return (tv_diff(now, &a->restart_barrier, NULL) > 0)? 0 : 1;
 }
 
-static void start_current_receiver(void)
+static void open_current_receiver(void)
 {
 	int i;
 
@@ -816,21 +756,20 @@ static void start_current_receiver(void)
 	i = get_audio_format_num(af_status);
 	if (i < 0)
 		return;
-	if ((decoder_running(i) & 1) || is_frozen(i))
+	if (decoder_running(i) || is_frozen(i))
 		return;
 	open_receiver(i);
 }
 
 static void compute_time_diff(const struct timeval *status_time)
 {
-	struct timeval now, tmp, diff;
+	struct timeval tmp, diff;
 	static int count;
 	int sign;
 	const struct timeval max_deviation = {0, 500 * 1000};
 	const int time_smooth = 5;
 
-	gettimeofday(&now, NULL);
-	sign = tv_diff(status_time, &now, &diff);
+	sign = tv_diff(status_time, now, &diff);
 //		PARA_NOTICE_LOG("%s: sign = %i, sa_time_diff_sign = %i\n", __func__,
 //			sign, sa_time_diff_sign);
 	if (!count) {
@@ -913,7 +852,12 @@ static void handle_signal(int sig)
 {
 	switch (sig) {
 	case SIGCHLD:
-		return check_sigchld();
+		for (;;) {
+			pid_t pid = para_reap_child();
+			if (pid <= 0)
+				return;
+			PARA_CRIT_LOG("para_client died (pid %d)\n", pid);
+		}
 	case SIGINT:
 	case SIGTERM:
 	case SIGHUP:
@@ -923,184 +867,64 @@ static void handle_signal(int sig)
 	}
 }
 
-static void check_timeouts(void)
-{
-	struct timeval now;
-	int slot_num, timeout = conf.stream_timeout_arg;
-
-	gettimeofday(&now, NULL);
-	FOR_EACH_SLOT(slot_num) {
-		struct slot_info *s = &slot[slot_num];
-		if (s->format < 0)
-			continue;
-		/* check read time */
-		if (s->receiver_node &&
-			now.tv_sec > s->rtime.tv_sec + timeout) {
-			PARA_INFO_LOG("%s input buffer (slot %d) not ready\n",
-				audio_formats[s->format], slot_num);
-			s->receiver_node->eof = 1;
-		}
-		/* check write time */
-		if (s->wpid > 0 && !s->wkilled &&
-			now.tv_sec > s->wtime.tv_sec + timeout) {
-			PARA_INFO_LOG("%s output buffer (slot %d) not ready\n",
-				audio_formats[s->format], slot_num);
-			kill_stream_writer(slot_num);
-		}
-	}
-}
-
-static size_t get_loaded_bytes(int slot_num)
-{
-	size_t loaded = 0;
-	struct slot_info *s = &slot[slot_num];
-	struct receiver_node *rn = s->receiver_node;
-
-	if (s->format < 0)
-		goto out;
-
-	if (afi[s->format].num_filters) {
-		if (s->fc)
-			loaded = *s->fc->out_loaded;
-	} else {
-		if (rn)
-			loaded = rn->loaded;
-	}
-out:
-	return loaded;
-}
-
-static void close_writer(int slot_num)
-{
-	struct slot_info *s = &slot[slot_num];
-	if (s->write_fd > 0) {
-		PARA_INFO_LOG("slot %d: closing write fd %d\n", slot_num,
-			s->write_fd);
-		close(s->write_fd);
-		del_close_on_fork_list(s->write_fd);
-		s->write_fd = -1;
-	}
-	if (s->fc)
-		*s->fc->output_eof = 1; /* FIXME */
-}
-
-
-static void close_decoder_if_idle(int slot_num)
+static void try_to_close_slot(int slot_num)
 {
 	struct slot_info *s = &slot[slot_num];
-	struct receiver_node *rn = s->receiver_node;
 
 	if (s->format < 0)
 		return;
-	if (!s->fc)
+	if (s->receiver_node && !s->receiver_node->eof)
 		return;
-	if (!rn->eof && !s->fc->eof && s->wpid > 0)
+	if (s->fc && !s->fc->eof)
 		return;
-	if (!s->fc->eof && s->wpid > 0 && get_loaded_bytes(slot_num))
+	if (s->wng && !s->wng->eof)
 		return;
-	close_writer(slot_num);
-	if (s->wpid > 0)
-		return; /* wait until writer dies before closing filters */
-	PARA_INFO_LOG("closing all filters in slot %d (filter_chain %p)\n",
-		slot_num, s->fc);
+	PARA_INFO_LOG("closing slot %d \n", slot_num);
+	wng_close(s->wng);
+	wng_destroy(s->wng);
 	close_filters(s->fc);
 	free(s->fc);
 	close_receiver(slot_num);
 	clear_slot(slot_num);
 }
 
-static void audiod_pre_select(struct sched *s, struct task *t)
+static void audiod_pre_select(struct sched *s, __a_unused struct task *t)
 {
 	int i;
 
+	now = &s->now;
 	if (audiod_status != AUDIOD_ON)
 		kill_all_decoders();
 	else if (playing)
-		start_current_receiver();
-	check_timeouts();
+		open_current_receiver();
 	FOR_EACH_SLOT(i) {
-		struct audio_format_info *a;
 		struct receiver_node *rn;
 
-		close_decoder_if_idle(i);
-		slot[i].wcheck = 0;
+		try_to_close_slot(i);
 		if (slot[i].format < 0)
 			continue;
-		a = &afi[slot[i].format];
 		rn = slot[i].receiver_node;
-		if (rn && rn->loaded && !slot[i].wpid) {
-			PARA_INFO_LOG("no writer in slot %d\n", i);
-			start_stream_writer(i);
+		if (rn && rn->loaded && !slot[i].wng) {
+			open_filters(i);
+			open_writers(i);
 		}
-		if (slot[i].write_fd <= 0)
-			continue;
-		if (!get_loaded_bytes(i))
-			continue;
-		para_fd_set(slot[i].write_fd, &s->wfds, &s->max_fileno);
-		slot[i].wcheck = 1;
-	}
-}
-
-static int write_audio_data(int slot_num)
-{
-	struct slot_info *s = &slot[slot_num];
-	struct audio_format_info *a = &afi[s->format];
-	struct receiver_node *rn = s->receiver_node;
-	int ret;
-	char **buf;
-	size_t *len;
-
-	if (a->num_filters) {
-		buf = &s->fc->outbuf;
-		len = s->fc->out_loaded;
-	} else {
-		buf = &rn->buf;
-		len = &rn->loaded;
 	}
-	PARA_DEBUG_LOG("writing %p (%zd bytes)\n", *buf, *len);
-	ret = write(s->write_fd, *buf, *len);
-	PARA_DEBUG_LOG("wrote %d/%zd\n", ret, *len);
-	if (ret < 0) {
-		PARA_WARNING_LOG("write error in slot %d (fd %d): %s\n",
-			slot_num, s->write_fd, strerror(errno));
-		*len = 0;
-		close_writer(slot_num);
-//		s->fc->error = E_WRITE_AUDIO_DATA;
-	} else if (ret != *len) {
-		PARA_DEBUG_LOG("partial %s write (%i/%zd) for slot %d\n",
-			audio_formats[s->format], ret, *len, slot_num);
-		*len -= ret;
-		memmove(*buf, *buf + ret, *len);
-	} else
-		*len = 0;
-	if (ret > 0)
-		gettimeofday(&s->wtime, NULL);
-	return ret;
 }
 
-static void audiod_post_select(struct sched *s, struct task *t)
+static void audiod_post_select(struct sched *s, __a_unused struct task *t)
 {
-	int ret, i;
-
-	FOR_EACH_SLOT(i) {
-		struct receiver_node *rn = slot[i].receiver_node;
-
-		if (rn && rn->loaded)
-			slot[i].rtime = s->now;
-		if (slot[i].write_fd <= 0 || !slot[i].wcheck
-			|| !FD_ISSET(slot[i].write_fd, &s->wfds))
-			continue;
-		ret = write_audio_data(i);
-	}
+	/* only save away the current time for other users */
+	now = &s->now;
 }
 
-static void init_audiod_task(struct audiod_task *at)
+static void init_audiod_task(struct task *t)
 {
-	at->task.pre_select = audiod_pre_select;
-	at->task.post_select = audiod_post_select;
-	at->task.private_data = at;
-	at->task.flags = 0;
-	sprintf(at->task.status, "audiod task");
+	t->pre_select = audiod_pre_select;
+	t->post_select = audiod_post_select;
+	t->event_handler = NULL;
+	t->private_data = t;
+	t->flags = 0;
+	sprintf(t->status, "audiod task");
 }
 
 static int parse_stream_command(const char *txt, char **cmd)
@@ -1135,53 +959,51 @@ static int add_filter(int format, char *cmdline)
 	return filter_num;
 }
 
-static int setup_default_filters(void)
+static int init_writers(void)
 {
-	int i, ret = 1;
+	int i, ret, nw;
+	char *cmd;
+	struct audio_format_info *a;
 
+	init_supported_writers();
+	nw = PARA_MAX(1, conf.writer_given);
+	PARA_INFO_LOG("maximal number of writers: %d\n", nw);
 	FOR_EACH_AUDIO_FORMAT(i) {
-		struct audio_format_info *a = &afi[i];
-		char *tmp;
-		int j;
-		if (a->num_filters)
-			continue;
-		/* add "dec" to audio format name */
-		tmp = make_message("%sdec", audio_formats[i]);
-		for (j = 0; filters[j].name; j++)
-			if (!strcmp(tmp, filters[j].name))
-				break;
-		free(tmp);
-		ret = -E_UNSUPPORTED_FILTER;
-		if (!filters[j].name)
-			goto out;
-		tmp = para_strdup(filters[j].name);
-		ret = add_filter(i, tmp);
-		free(tmp);
+		a = &afi[i];
+		a->writer_conf = para_malloc(nw * sizeof(void *));
+		a->writers = para_malloc(nw * sizeof(struct writer *));
+		a->num_writers = 0;
+	}
+	for (i = 0; i < conf.writer_given; i++) {
+		void *wconf;
+		int writer_num;
+		ret = parse_stream_command(conf.writer_arg[i], &cmd);
 		if (ret < 0)
 			goto out;
-		PARA_INFO_LOG("%s -> default filter: %s\n", audio_formats[i],
-			filters[j].name);
-		ret = add_filter(i, "wav");
-		if (ret < 0)
+		a = &afi[ret];
+		nw = a->num_writers;
+		wconf = check_writer_arg(cmd, &writer_num);
+		if (!wconf) {
+			ret = writer_num;
 			goto out;
-		PARA_INFO_LOG("%s -> default filter: wav\n", audio_formats[i]);
+		}
+		a->writers[nw] = &writers[writer_num];
+		a->writer_conf[nw] = wconf;
+		PARA_INFO_LOG("%s writer #%d: %s\n", audio_formats[ret],
+			nw, writer_names[writer_num]);
+		a->num_writers++;
 	}
+	ret = 1;
 out:
 	return ret;
 }
 
-static int init_stream_io(void)
+static int init_receivers(void)
 {
-	int i, ret, receiver_num, nf;
-	char *cmd;
+	int i, ret, receiver_num;
+	char *cmd = NULL;
+	struct audio_format_info *a;
 
-	for (i = 0; i < conf.stream_write_cmd_given; i++) {
-		ret = parse_stream_command(conf.stream_write_cmd_arg[i], &cmd);
-		if (ret < 0)
-			goto out;
-		afi[ret].write_cmd = para_strdup(cmd);
-		PARA_INFO_LOG("%s write command: %s\n", audio_formats[ret], afi[ret].write_cmd);
-	}
 	for (i = 0; receivers[i].name; i++) {
 		PARA_INFO_LOG("initializing %s receiver\n", receivers[i].name);
 		receivers[i].init(&receivers[i]);
@@ -1210,7 +1032,7 @@ static int init_stream_io(void)
 	 */
 	cmd = para_strdup(receivers[0].name);
 	FOR_EACH_AUDIO_FORMAT(i) {
-		struct audio_format_info *a = &afi[i];
+		a = &afi[i];
 		if (a->receiver_conf)
 			continue;
 		a->receiver_conf = check_receiver_arg(cmd, &receiver_num);
@@ -1218,17 +1040,57 @@ static int init_stream_io(void)
 			return -E_RECV_SYNTAX;
 		a->receiver = &receivers[receiver_num];
 	}
+	ret = 1;
+out:
 	free(cmd);
-	/* filters */
+	return ret;
+}
+
+static int init_default_filters(void)
+{
+	int i, ret = 1;
+
+	FOR_EACH_AUDIO_FORMAT(i) {
+		struct audio_format_info *a = &afi[i];
+		char *tmp;
+		int j;
+
+		if (a->num_filters)
+			continue; /* no default -- nothing to to */
+		/* add "dec" to audio format name */
+		tmp = make_message("%sdec", audio_formats[i]);
+		for (j = 0; filters[j].name; j++)
+			if (!strcmp(tmp, filters[j].name))
+				break;
+		free(tmp);
+		ret = -E_UNSUPPORTED_FILTER;
+		if (!filters[j].name)
+			goto out;
+		tmp = para_strdup(filters[j].name);
+		ret = add_filter(i, tmp);
+		free(tmp);
+		if (ret < 0)
+			goto out;
+		PARA_INFO_LOG("%s -> default filter: %s\n", audio_formats[i],
+			filters[j].name);
+	}
+out:
+	return ret;
+}
+
+static int init_filters(void)
+{
+	int i, ret, nf;
+
 	filter_init(filters);
-	nf = PARA_MAX(2,  conf.filter_given) + 1;
-	PARA_INFO_LOG("allocating space for %d filters\n", nf);
+	nf = PARA_MAX(1,  conf.filter_given);
+	PARA_INFO_LOG("maximal number of filters: %d\n", nf);
 	FOR_EACH_AUDIO_FORMAT(i) {
-		afi[i].filter_conf = para_malloc(nf * sizeof(char *));
+		afi[i].filter_conf = para_malloc(nf * sizeof(void *));
 		afi[i].filters = para_malloc(nf * sizeof(struct filter *));
 	}
 	if (!conf.no_default_filters_given)
-		return setup_default_filters();
+		return init_default_filters();
 	for (i = 0; i < conf.filter_given; i++) {
 		char *arg = conf.filter_arg[i];
 		char *filter_name = strchr(arg, ':');
@@ -1244,11 +1106,27 @@ static int init_stream_io(void)
 		if (ret < 0)
 			goto out;
 	}
-	ret = 1;
+	ret = init_default_filters(); /* use default values for the rest */
 out:
 	return ret;
 }
 
+static int init_stream_io(void)
+{
+	int ret;
+
+	ret = init_writers();
+	if (ret < 0)
+		return ret;
+	ret = init_receivers();
+	if (ret < 0)
+		return ret;
+	ret = init_filters();
+	if (ret < 0)
+		return ret;
+	return 1;
+}
+
 static int dump_commands(int fd)
 {
 	char *buf = para_strdup(""), *tmp = NULL;
@@ -1376,29 +1254,6 @@ out:
 	return ret;
 }
 
-#if 0
-static char *list_filters(void)
-{
-	int i, j;
-	char *tmp, *msg = make_message("format\tnum\tcmd\n");
-
-	FOR_EACH_AUDIO_FORMAT(i) {
-		for (j = 0; j < afi[i].num_filters; j++) {
-			tmp = make_message("%s\t%i\t%s\n",
-				afi[i].name, j, afi[i].filter_cmds[j]);
-			msg = para_strcat(msg, tmp);
-			free(tmp);
-		}
-		tmp = make_message("%s\t%i\t%s\n", afi[i].name,
-			j, afi[i].write_cmd);
-		msg = para_strcat(msg, tmp);
-		free(tmp);
-	}
-	return msg;
-}
-#endif
-
-
 static int com_grab(int fd, char *cmdline)
 {
 	struct grab_client *gc;
@@ -1492,13 +1347,13 @@ static int check_perms(uid_t uid)
 	return -E_UCRED_PERM;
 }
 
-static int handle_connect(void)
+static int handle_connect(int accept_fd)
 {
 	int i, argc, ret, clifd = -1;
 	char *cmd = NULL, *p, *buf = para_calloc(MAXLINE), **argv = NULL;
 	struct sockaddr_un unix_addr;
 
-	ret = para_accept(cmd_task->fd, &unix_addr, sizeof(struct sockaddr_un));
+	ret = para_accept(accept_fd, &unix_addr, sizeof(struct sockaddr_un));
 	if (ret < 0)
 		goto out;
 	clifd = ret;
@@ -1549,9 +1404,10 @@ out:
 	return ret;
 }
 
-static void audiod_get_socket(void)
+static int audiod_get_socket(void)
 {
 	struct sockaddr_un unix_addr;
+	int fd;
 
 	if (conf.socket_given)
 		socket_name = para_strdup(conf.socket_arg);
@@ -1561,20 +1417,21 @@ static void audiod_get_socket(void)
 			hn);
 		free(hn);
 	}
-	PARA_NOTICE_LOG("connecting to local socket %s\n", socket_name);
+	PARA_NOTICE_LOG("local socket: %s\n", socket_name);
 	if (conf.force_given)
 		unlink(socket_name);
-	cmd_task->fd = create_pf_socket(socket_name, &unix_addr,
+	fd = create_pf_socket(socket_name, &unix_addr,
 			S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IWOTH);
-	if (cmd_task->fd < 0) {
+	if (fd < 0) {
 		PARA_EMERG_LOG("%s", "can not connect to socket\n");
 		exit(EXIT_FAILURE); /* do not unlink socket */
 	}
-	if (listen(cmd_task->fd , 5) < 0) {
+	if (listen(fd , 5) < 0) {
 		PARA_EMERG_LOG("%s", "can not listen on socket\n");
 		exit(EXIT_FAILURE); /* do not unlink socket */
 	}
-	add_close_on_fork_list(cmd_task->fd);
+	add_close_on_fork_list(fd);
+	return fd;
 }
 
 static int open_stat_pipe(void)
@@ -1640,17 +1497,19 @@ static void command_post_select(struct sched *s, struct task *t)
 	t->ret = 1; /* always successful */
 	if (!FD_ISSET(ct->fd, &s->rfds))
 		return;
-	ret = handle_connect();
+	ret = handle_connect(ct->fd);
 	if (ret < 0)
 		PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
 }
 
-void init_command_task(struct command_task *ct)
+static void init_command_task(struct command_task *ct)
 {
 	ct->task.pre_select = command_pre_select;
 	ct->task.post_select = command_post_select;
+	ct->task.event_handler = NULL;
 	ct->task.private_data = ct;
 	ct->task.flags = 0;
+	ct->fd = audiod_get_socket(); /* doesn't return on errors */
 	sprintf(ct->task.status, "command task");
 }
 
@@ -1682,7 +1541,7 @@ static void status_post_select(struct sched *s, struct task *t)
 	if (ret <= 0) {
 		close_stat_pipe();
 		/* avoid busy loop if server is down */
-		while (sleep(1) > 0)
+		while (sleep(1) > 0) /* FIXME */
 			; /* try again*/
 	} else {
 		st->buf[ret + st->loaded] = '\0';
@@ -1703,84 +1562,6 @@ static void init_status_task(struct status_task *st)
 	sprintf(st->task.status, "status task");
 }
 
-
-
-#if 0
-/* TODO: move everything before the select call to pre_select() */
-static void __noreturn audiod_mainloop(void)
-{
-	fd_set rfds, wfds;
-	int ret, max_fileno, sbo = 0;
-	char status_buf[STRINGSIZE] = "";
-	struct timeval tv;
-
-
-
-repeat:
-	FD_ZERO(&wfds);
-	FD_ZERO(&rfds);
-	max_fileno = -1;
-	/* always check signal pipe and the local socket */
-	para_fd_set(signal_pipe, &rfds, &max_fileno);
-	para_fd_set(audiod_socket, &rfds, &max_fileno);
-
-	if (audiod_status != AUDIOD_ON)
-		kill_all_decoders();
-	else if (playing)
-		start_current_receiver();
-
-	set_stream_fds(&wfds, &max_fileno);
-	/* status pipe */
-	if (stat_pipe >= 0 && audiod_status == AUDIOD_OFF)
-		close_stat_pipe();
-	if (stat_pipe < 0 && audiod_status != AUDIOD_OFF) {
-		stat_pipe = open_stat_pipe();
-		sbo = 0;
-		status_buf[0] = '\0';
-	}
-	if (stat_pipe >= 0 && audiod_status != AUDIOD_OFF)
-		para_fd_set(stat_pipe, &rfds, &max_fileno);
-	/* local socket */
-	tv.tv_sec = 0;
-	tv.tv_usec = 200 * 1000;
-	audiod_pre_select(&rfds, &wfds, &tv, &max_fileno);
-	ret = para_select(max_fileno + 1, &rfds, &wfds, &tv);
-	if (ret < 0)
-		goto repeat;
-	if (audiod_status != AUDIOD_OFF)
-		audiod_status_dump();
-	audiod_post_select(ret, &rfds, &wfds);
-	/* read status pipe */
-	if (stat_pipe >=0 && FD_ISSET(stat_pipe, &rfds)) {
-		ret = read(stat_pipe, status_buf + sbo, STRINGSIZE - 1 - sbo);
-		if (ret <= 0) {
-			close_stat_pipe();
-			/* avoid busy loop if server is down */
-			while (sleep(1) > 0)
-				; /* try again*/
-		} else {
-			status_buf[ret + sbo] = '\0';
-			sbo = for_each_line(status_buf, ret + sbo,
-				&check_stat_line);
-		}
-	}
-	slot_io(&wfds);
-	if (FD_ISSET(audiod_socket, &rfds)) {
-		ret = handle_connect();
-		if (ret < 0) {
-			PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
-		}
-	}
-	/* signals */
-	if (FD_ISSET(signal_pipe, &rfds)) {
-		int sig_nr = para_next_signal();
-		if (sig_nr > 0)
-			handle_signal(sig_nr);
-	}
-	goto repeat;
-}
-#endif
-
 static void set_initial_status(void)
 {
 	audiod_status = AUDIOD_ON;
@@ -1803,6 +1584,8 @@ int main(int argc, char *argv[])
 	char *cf;
 	int ret, i;
 	struct sched s;
+	struct command_task command_task_struct, *cmd_task = &command_task_struct;
+	struct task audiod_task_struct, *audiod_task = &audiod_task_struct;
 
 	init_sched();
 
@@ -1831,21 +1614,20 @@ int main(int argc, char *argv[])
 		clear_slot(i);
 	init_grabbing();
 	setup_signal_handling();
-	if (conf.daemon_given)
-		daemon_init();
-	audiod_get_socket(); /* doesn't return on errors */
-
 	signal_setup_default(sig_task);
 	sig_task->task.event_handler = signal_event_handler;
 
 	init_status_task(stat_task);
 	init_command_task(cmd_task);
-	init_audiod_task(at);
+	init_audiod_task(audiod_task);
+
+	if (conf.daemon_given)
+		daemon_init();
 
 	register_task(&sig_task->task);
 	register_task(&cmd_task->task);
 	register_task(&stat_task->task);
-	register_task(&at->task);
+	register_task(audiod_task);
 	s.default_timeout.tv_sec = 0;
 	s.default_timeout.tv_usec = 99 * 1000;
 	ret = sched(&s);