X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=audiod.c;h=fcce8ae9dd1d87e7480412524b77cb5f5b97cbdf;hp=ae20d70137a6bb097109d731dd6fce63afbd7c19;hb=899a19c03fc05f6fd98687e3eaa86841a4633a10;hpb=586cea4b850e53eedf7426379f6af6a7e76dee6e

diff --git a/audiod.c b/audiod.c
index ae20d701..fcce8ae9 100644
--- a/audiod.c
+++ b/audiod.c
@@ -24,6 +24,7 @@
 #include "audiod.cmdline.h"
 #include "list.h"
 #include "close_on_fork.h"
+#include "sched.h"
 #include "recv.h"
 #include "filter.h"
 #include "grab_client.cmdline.h"
@@ -74,61 +75,36 @@ struct audio_format_info {
  * \sa receier_node, receiver, filter, filter_node, filter_chain_info
   */
 struct slot_info {
-/** number of the audio format in this slot */
+	/** number of the audio format in this slot */
 	int format;
-/** the file descriptor of the writer */
+	/** the file descriptor of the writer */
 	int write_fd;
-/** the process id of the writer */
+	/** the process id of the writer */
 	pid_t wpid;
-/** time of the last successful read from the receiver */
+	/** time of the last successful read from the receiver */
 	struct timeval rtime;
-/** time the last write to the write fd happend */
+	/** time the last write to the write fd happend */
 	struct timeval wtime;
-/** writer start time */
+	/** writer start time */
 	struct timeval wstime;
-/** did we include \a write_fd in the fdset */
+	/** did we include \a write_fd in the fdset */
 	int  wcheck;
-/** set to one if we have sent the TERM signal to \a wpid */
+	/** set to one if we have sent the TERM signal to \a wpid */
 	int wkilled;
-/** the receiver info associated with this slot */
+	/** the receiver info associated with this slot */
 	struct receiver_node *receiver_node;
-/** the active filter chain */
-	struct filter_chain_info *fci;
+	/** the active filter chain */
+	struct filter_chain *fc;
 };
-
 static struct slot_info slot[MAX_STREAM_SLOTS];
 
-/** 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**);
-/** one-line description of the command */
-const char *description;
-/** summary of the command line options */
-const char *synopsis;
-/** the long help text */
-const char *help;
-};
-
 extern const char *status_item_list[NUM_STAT_ITEMS];
 
-static int com_grab(int, int, char **);
-static int com_cycle(int, int, char **);
-static int com_help(int, int, char **);
-static int com_off(int, int, char **);
-static int com_on(int, int, char **);
-static int com_sb(int, int, char **);
-static int com_stat(int, int, char **);
-static int com_term(int, int, char **);
-static int stat_pipe = -1, signal_pipe;
-
 static struct gengetopt_args_info conf;
 static struct timeval server_stream_start, sa_time_diff;
 static int playing, current_decoder = -1,
 	audiod_status = AUDIOD_ON, offset_seconds, length_seconds,
-	sa_time_diff_sign = 1, audiod_socket = -1;
+	sa_time_diff_sign = 1;
 static char *af_status, /* the audio format announced in server status */
 	*socket_name, *hostname;
 static char *stat_item_values[NUM_STAT_ITEMS];
@@ -137,6 +113,56 @@ static const struct timeval restart_delay = {0, 300 * 1000};
 
 static struct audio_format_info afi[NUM_AUDIO_FORMATS];
 
+static struct signal_task signal_task_struct, *sig_task = &signal_task_struct;
+
+struct command_task {
+	int fd;
+	struct task task;
+};
+static struct command_task command_task_struct, *cmd_task = &command_task_struct;
+
+struct status_task {
+	int fd;
+	struct task task;
+	char buf[STRINGSIZE];
+	unsigned loaded;
+};
+static struct status_task status_task_struct, *stat_task = &status_task_struct;
+
+struct audiod_task {
+	struct task task;
+};
+static struct audiod_task audiod_task_struct, *at = &audiod_task_struct;
+
+struct signal_task {
+	int fd;
+	int signum;
+	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**);
+	int (*line_handler)(int, char*);
+	/** one-line description of the command */
+	const char *description;
+	/** summary of the command line options */
+	const char *synopsis;
+	/** the long help text */
+	const char *help;
+};
+static int com_grab(int, char *);
+static int com_cycle(int, int, char **);
+static int com_help(int, int, char **);
+static int com_off(int, int, char **);
+static int com_on(int, int, char **);
+static int com_sb(int, int, char **);
+static int com_stat(int, int, char **);
+static int com_term(int, int, char **);
 static struct audiod_command cmds[] = {
 {
 .name = "cycle",
@@ -150,7 +176,7 @@ static struct audiod_command cmds[] = {
 },
 {
 .name = "grab",
-.handler = com_grab,
+.line_handler = com_grab,
 .description = "grab the audio stream",
 .synopsis = "-- grab [grab_options]",
 .help =
@@ -233,7 +259,7 @@ static struct audiod_command cmds[] = {
 };
 
 /** iterate over all slots */
-#define FOR_EACH_SLOT(slot) for (slot = 0; slot < MAX_STREAM_SLOTS; slot++)
+#define FOR_EACH_SLOT(_slot) for (_slot = 0; _slot < MAX_STREAM_SLOTS; _slot++)
 /** iterate over all supported audio formats */
 #define FOR_EACH_AUDIO_FORMAT(af) for (af = 0; af < NUM_AUDIO_FORMATS; af++)
 /** iterate over the array of all audiod commands */
@@ -368,8 +394,6 @@ __malloc static char *decoder_flags(void)
 			flag += 1;
 		if (s->wpid > 0)
 			flag += 2;
-		if (flag != '0')
-			flag += s->format * 4;
 		decoder_flags[i] = flag;
 	}
 	decoder_flags[MAX_STREAM_SLOTS] = '\0';
@@ -391,8 +415,8 @@ static char *configfile_exists(void)
 
 static void setup_signal_handling(void)
 {
-	signal_pipe = para_signal_init();
-	PARA_INFO_LOG("signal pipe: fd %d\n", signal_pipe);
+	sig_task->fd = para_signal_init();
+	PARA_INFO_LOG("signal pipe: fd %d\n", sig_task->fd);
 	para_install_sighandler(SIGINT);
 	para_install_sighandler(SIGTERM);
 	para_install_sighandler(SIGCHLD);
@@ -451,7 +475,6 @@ static void kill_stream_writer(int slot_num)
 		s->wpid, audio_formats[s->format], slot_num);
 	kill(s->wpid, SIGTERM);
 	s->wkilled = 1;
-	s->fci->error = 1;
 }
 
 static void set_restart_barrier(int format, struct timeval *now)
@@ -473,8 +496,10 @@ 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\n",
-		audio_formats[s->format] , slot_num);
+	PARA_NOTICE_LOG("closing %s recevier 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;
@@ -537,7 +562,7 @@ static int get_empty_slot(void)
 			continue;
 		if (s->receiver_node)
 			continue;
-		if (s->fci)
+		if (s->fc)
 			continue;
 		clear_slot(i);
 		return i;
@@ -564,12 +589,12 @@ static void close_stat_pipe(void)
 {
 	int i;
 
-	if (stat_pipe < 0)
+	if (stat_task->fd < 0)
 		return;
 	PARA_NOTICE_LOG("%s", "closing status pipe\n");
-	close(stat_pipe);
-	del_close_on_fork_list(stat_pipe);
-	stat_pipe = -1;
+	close(stat_task->fd);
+	del_close_on_fork_list(stat_task->fd);
+	stat_task->fd = -1;
 	kill_all_decoders();
 	for (i = 0; i < NUM_STAT_ITEMS; i++) {
 		free(stat_item_values[i]);
@@ -591,7 +616,7 @@ static void __noreturn clean_exit(int status, const char *msg)
 	kill_all_decoders();
 	if (socket_name)
 		unlink(socket_name);
-	if (stat_pipe >= 0)
+	if (stat_task->fd >= 0)
 		close_stat_pipe();
 	exit(status);
 }
@@ -627,38 +652,53 @@ int num_filters(int audio_format_num)
 	return afi[audio_format_num].num_filters;
 }
 
+void filter_event_handler(struct task *t)
+{
+	PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
+	unregister_task(t);
+}
+
 static void open_filters(int slot_num)
 {
 	struct slot_info *s = &slot[slot_num];
 	struct audio_format_info *a = &afi[s->format];
 	int nf = a->num_filters;
 	int i;
+	static int output_eof; /* FIXME */
 
-	s->fci = para_calloc(sizeof(struct filter_chain_info));
-	INIT_LIST_HEAD(&s->fci->filters);
+	s->fc = para_calloc(sizeof(struct filter_chain));
+	INIT_LIST_HEAD(&s->fc->filters);
 	if (!nf)
 		return;
-	s->fci->inbuf = s->receiver_node->buf;
-	s->fci->in_loaded = &s->receiver_node->loaded;
-	s->fci->outbuf = s->receiver_node->buf;
-	s->fci->out_loaded = &s->receiver_node->loaded;
-	s->fci->eof = &s->receiver_node->eof;
+	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;
+	s->fc->task.private_data = s->fc;
+	s->fc->task.flags = 0;
+	s->fc->eof = 0;
+	sprintf(s->fc->task.status, "filter chain");
 	for (i = 0; i < nf; i++) {
 		struct filter_node *fn = para_calloc(sizeof(struct filter_node));
 		fn->conf = a->filter_conf[i];
-		fn->fci = s->fci;
+		fn->fc = s->fc;
 		fn->filter = a->filters[i];
 		INIT_LIST_HEAD(&fn->callbacks);
-		list_add_tail(&fn->node, &s->fci->filters);
+		list_add_tail(&fn->node, &s->fc->filters);
 		fn->filter->open(fn);
 		PARA_NOTICE_LOG("%s filter %d/%d (%s) started in slot %d\n",
 			audio_formats[s->format], i + 1,  nf,
 			fn->filter->name, slot_num);
-		s->fci->outbuf = fn->buf;
-		s->fci->out_loaded = &fn->loaded;
+		s->fc->outbuf = fn->buf;
+		s->fc->out_loaded = &fn->loaded;
 	}
-	PARA_DEBUG_LOG("output buffer for filter chain %p: %p\n", s->fci,
-		s->fci->outbuf);
+	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)
@@ -668,7 +708,7 @@ static struct filter_node *find_filter_node(int slot_num, int format, int filter
 
 	FOR_EACH_SLOT(i) {
 		struct slot_info *s = &slot[i];
-		if (s->format < 0 || !s->fci)
+		if (s->format < 0 || !s->fc)
 			continue;
 		if (slot_num >= 0 && slot_num != i)
 			continue;
@@ -678,7 +718,7 @@ static struct filter_node *find_filter_node(int slot_num, int format, int filter
 			continue;
 		/* success */
 		j = 1;
-		list_for_each_entry(fn, &s->fci->filters, node)
+		list_for_each_entry(fn, &s->fc->filters, node)
 			if (filternum <= 0 || j++ == filternum)
 				break;
 		return fn;
@@ -696,7 +736,7 @@ static void start_stream_writer(int slot_num)
 	if (a->write_cmd)
 		glob = glob_cmd(a->write_cmd);
 	if (!glob)
-		glob = para_strdup("para_play");
+		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);
@@ -711,14 +751,21 @@ static void start_stream_writer(int slot_num)
 	mark_fd_nonblock(s->write_fd);
 	gettimeofday(&s->wstime, NULL);
 	current_decoder = slot_num;
-	activate_inactive_grab_clients(slot_num, s->format, &s->fci->filters);
+	activate_inactive_grab_clients(slot_num, s->format, &s->fc->filters);
 }
 
+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];
 	struct slot_info *s;
 	int ret, slot_num;
+	struct receiver_node *rn;
 
 	slot_num = get_empty_slot();
 	if (slot_num < 0)
@@ -728,7 +775,9 @@ static void open_receiver(int format)
 	gettimeofday(&s->rtime, NULL);
 	s->wtime = s->rtime;
 	s->receiver_node = para_calloc(sizeof(struct receiver_node));
-	s->receiver_node->conf = a->receiver_conf;
+	rn = s->receiver_node;
+	rn->receiver = a->receiver;
+	rn->conf = a->receiver_conf;
 	ret = a->receiver->open(s->receiver_node);
 	if (ret < 0) {
 		PARA_ERROR_LOG("failed to open receiver (%s)\n",
@@ -739,6 +788,14 @@ 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;
+	rn->task.flags = 0;
+	sprintf(rn->task.status, "receiver node");
+	register_task(&rn->task);
 }
 
 static int is_frozen(int format)
@@ -806,10 +863,10 @@ static void check_stat_line(char *line)
 {
 	int itemnum;
 	size_t ilen = 0;
-	struct timeval tv;
+	long unsigned sec, usec;
 	char *tmp;
 
-	PARA_INFO_LOG("line: %s\n", line);
+//	PARA_INFO_LOG("line: %s\n", line);
 	if (!line)
 		return;
 	itemnum = stat_line_valid(line);
@@ -838,14 +895,16 @@ static void check_stat_line(char *line)
 		length_seconds = atoi(line + ilen + 1);
 		break;
 	case SI_STREAM_START:
-		if (sscanf(line + ilen + 1, "%lu.%lu",
-				&tv.tv_sec, &tv.tv_usec) == 2)
-			server_stream_start = tv;
+		if (sscanf(line + ilen + 1, "%lu.%lu", &sec, &usec) == 2) {
+			server_stream_start.tv_sec = sec;
+			server_stream_start.tv_usec = usec;
+		}
 		break;
 	case SI_CURRENT_TIME:
-		if (sscanf(line + ilen + 1, "%lu.%lu", &tv.tv_sec,
-				&tv.tv_usec) == 2)
+		if (sscanf(line + ilen + 1, "%lu.%lu", &sec, &usec) == 2) {
+			struct timeval tv = {sec, usec};
 			compute_time_diff(&tv);
+		}
 		break;
 	}
 }
@@ -879,18 +938,14 @@ static void check_timeouts(void)
 			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);
-			if (s->fci)
-				s->fci->error = 42;
-			else
-				close_receiver(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);
-			if (s->fci)
-				s->fci->error = 42;
+			kill_stream_writer(slot_num);
 		}
 	}
 }
@@ -905,8 +960,8 @@ static size_t get_loaded_bytes(int slot_num)
 		goto out;
 
 	if (afi[s->format].num_filters) {
-		if (s->fci)
-			loaded = *s->fci->out_loaded;
+		if (s->fc)
+			loaded = *s->fc->out_loaded;
 	} else {
 		if (rn)
 			loaded = rn->loaded;
@@ -915,6 +970,20 @@ 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)
 {
@@ -923,59 +992,52 @@ static void close_decoder_if_idle(int slot_num)
 
 	if (s->format < 0)
 		return;
-	if (!s->fci)
+	if (!s->fc)
 		return;
-	if (!rn->eof && !s->fci->error && s->wpid > 0)
+	if (!rn->eof && !s->fc->eof && s->wpid > 0)
 		return;
-	if (!s->fci->error && s->wpid > 0) { /* eof */
-		if (filter_io(s->fci) > 0)
-			return;
-		if (get_loaded_bytes(slot_num))
-			return;
-	}
-	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->eof && s->wpid > 0 && get_loaded_bytes(slot_num))
+		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->fci);
-	close_filters(s->fci);
-	free(s->fci);
+		slot_num, s->fc);
+	close_filters(s->fc);
+	free(s->fc);
 	close_receiver(slot_num);
 	clear_slot(slot_num);
 }
 
-static void set_stream_fds(fd_set *wfds, int *max_fileno)
+static void audiod_pre_select(struct sched *s, struct task *t)
 {
 	int i;
 
+	if (audiod_status != AUDIOD_ON)
+		kill_all_decoders();
+	else if (playing)
+		start_current_receiver();
 	check_timeouts();
 	FOR_EACH_SLOT(i) {
-		struct slot_info *s = &slot[i];
 		struct audio_format_info *a;
 		struct receiver_node *rn;
 
 		close_decoder_if_idle(i);
-		s->wcheck = 0;
-		if (s->format < 0)
+		slot[i].wcheck = 0;
+		if (slot[i].format < 0)
 			continue;
-		a = &afi[s->format];
-		rn = s->receiver_node;
-		if (rn && rn->loaded && !s->wpid) {
+		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 (s->write_fd <= 0)
+		if (slot[i].write_fd <= 0)
 			continue;
 		if (!get_loaded_bytes(i))
 			continue;
-		para_fd_set(s->write_fd, wfds, max_fileno);
-		s->wcheck = 1;
+		para_fd_set(slot[i].write_fd, &s->wfds, &s->max_fileno);
+		slot[i].wcheck = 1;
 	}
 }
 
@@ -984,60 +1046,63 @@ 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 rv;
+	int ret;
 	char **buf;
 	size_t *len;
 
 	if (a->num_filters) {
-		buf = &s->fci->outbuf;
-		len = s->fci->out_loaded;
+		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);
-	rv = write(s->write_fd, *buf, *len);
-	PARA_DEBUG_LOG("wrote %d/%zd\n", rv, *len);
-	if (rv < 0) {
+	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;
-		s->fci->error = E_WRITE_AUDIO_DATA;
-	} else if (rv != *len) {
+		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], rv, *len, slot_num);
-		*len -= rv;
-		memmove(*buf, *buf + rv, *len);
+			audio_formats[s->format], ret, *len, slot_num);
+		*len -= ret;
+		memmove(*buf, *buf + ret, *len);
 	} else
 		*len = 0;
-	if (rv > 0)
+	if (ret > 0)
 		gettimeofday(&s->wtime, NULL);
-	return rv;
+	return ret;
 }
 
-static void slot_io(fd_set *wfds)
+static void audiod_post_select(struct sched *s, struct task *t)
 {
 	int ret, i;
 
 	FOR_EACH_SLOT(i) {
-		struct slot_info *s = &slot[i];
-		struct receiver_node *rn = s->receiver_node;
+		struct receiver_node *rn = slot[i].receiver_node;
 
 		if (rn && rn->loaded)
-			gettimeofday(&s->rtime, NULL);
-		if (s->format >= 0 && s->write_fd > 0 && s->fci) {
-			ret = filter_io(s->fci);
-			if (ret < 0)
-				s->fci->error = -ret;
-//			PARA_DEBUG_LOG("slot %d, filter io %d bytes, check write: %d, loaded: %d/%d, eof: %d\n",
-//				 i, ret, s->wcheck, rn->loaded, *s->fci->out_loaded, rn->eof);
-		}
-		if (s->write_fd <= 0 || !s->wcheck || !FD_ISSET(s->write_fd, wfds))
+			slot[i].rtime = s->now;
+		if (slot[i].write_fd <= 0 || !slot[i].wcheck
+			|| !FD_ISSET(slot[i].write_fd, &s->wfds))
 			continue;
-		write_audio_data(i);
+		ret = write_audio_data(i);
 	}
 }
 
+static void init_audiod_task(struct audiod_task *at)
+{
+	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");
+}
+
 static int parse_stream_command(const char *txt, char **cmd)
 {
 	char *p = strchr(txt, ':');
@@ -1094,8 +1159,9 @@ static int setup_default_filters(void)
 		free(tmp);
 		if (ret < 0)
 			goto out;
-		PARA_INFO_LOG("%s -> default filter: %s\n", audio_formats[i], filters[j].name);
-		ret = add_filter(i, para_strdup("wav"));
+		PARA_INFO_LOG("%s -> default filter: %s\n", audio_formats[i],
+			filters[j].name);
+		ret = add_filter(i, "wav");
 		if (ret < 0)
 			goto out;
 		PARA_INFO_LOG("%s -> default filter: wav\n", audio_formats[i]);
@@ -1332,16 +1398,15 @@ static char *list_filters(void)
 }
 #endif
 
-static int com_grab(int fd, int argc, char **argv)
+
+static int com_grab(int fd, char *cmdline)
 {
 	struct grab_client *gc;
 	struct filter_node *fn;
 	int i, err;
 	char *msg;
 
-	PARA_INFO_LOG("argc: %d, argv[0]: %s, optind: %d\n", argc, argv[0], optind);
-	gc = grab_client_new(fd, argc, argv, &err);
-	PARA_INFO_LOG("argc: %d, argv[0]: %s, optind: %d\n", argc, argv[0], optind);
+	gc = grab_client_new(fd, cmdline, &err);
 	if (!gc)
 		goto err_out;
 	fn = find_filter_node(gc->conf->slot_arg, gc->audio_format_num, gc->conf->filter_num_arg);
@@ -1430,30 +1495,49 @@ static int check_perms(uid_t uid)
 static int handle_connect(void)
 {
 	int i, argc, ret, clifd = -1;
-	char *buf = para_malloc(MAXLINE), **argv = NULL;
+	char *cmd = NULL, *p, *buf = para_calloc(MAXLINE), **argv = NULL;
 	struct sockaddr_un unix_addr;
 
-	ret = para_accept(audiod_socket, &unix_addr, sizeof(struct sockaddr_un));
+	ret = para_accept(cmd_task->fd, &unix_addr, sizeof(struct sockaddr_un));
 	if (ret < 0)
 		goto out;
 	clifd = ret;
 	ret = recv_cred_buffer(clifd, buf, MAXLINE - 1);
 	if (ret < 0)
 		goto out;
-	PARA_INFO_LOG("connection from user %i\n",  ret);
+	PARA_INFO_LOG("connection from user %i, buf: %s\n",  ret, buf);
 	ret = check_perms(ret);
 	if (ret < 0)
 		goto out;
-	argc = split_args(buf, &argv, "\n");
-	PARA_INFO_LOG("argv[0]: %s, argc= %d\n", argv[0], argc);
+	ret = -E_INVALID_AUDIOD_CMD;
+	cmd = para_strdup(buf);
+	p = strchr(cmd, '\n');
+	if (!p)
+		p = "";
+	else {
+		*p = '\0';
+		p++;
+	}
 	for (i = 0; cmds[i].name; i++) {
-		if (strcmp(cmds[i].name, argv[0]))
+		int j;
+		if (strcmp(cmds[i].name, cmd))
 			continue;
-		ret = cmds[i].handler(clifd, argc, argv);
+		if (cmds[i].handler) {
+			argc = split_args(buf, &argv, "\n");
+			PARA_INFO_LOG("argv[0]: %s, argc= %d\n", argv[0], argc);
+			ret = cmds[i].handler(clifd, argc, argv);
+			goto out;
+		}
+		for (j = 0; p[j]; j++)
+			if (p[j] == '\n')
+				p[j] = ' ';
+		PARA_INFO_LOG("cmd: %s, options: %s\n", cmd, p);
+		ret = cmds[i].line_handler(clifd, p);
 		goto out;
 	}
 	ret = -E_INVALID_AUDIOD_CMD; /* cmd not found */
 out:
+	free(cmd);
 	free(buf);
 	free(argv);
 	if (clifd > 0 && ret < 0 && ret != -E_CLIENT_WRITE) {
@@ -1480,17 +1564,17 @@ static void audiod_get_socket(void)
 	PARA_NOTICE_LOG("connecting to local socket %s\n", socket_name);
 	if (conf.force_given)
 		unlink(socket_name);
-	audiod_socket = create_pf_socket(socket_name, &unix_addr,
+	cmd_task->fd = create_pf_socket(socket_name, &unix_addr,
 			S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IWOTH);
-	if (audiod_socket < 0) {
+	if (cmd_task->fd < 0) {
 		PARA_EMERG_LOG("%s", "can not connect to socket\n");
 		exit(EXIT_FAILURE); /* do not unlink socket */
 	}
-	if (listen(audiod_socket, 5) < 0) {
+	if (listen(cmd_task->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(audiod_socket);
+	add_close_on_fork_list(cmd_task->fd);
 }
 
 static int open_stat_pipe(void)
@@ -1507,48 +1591,121 @@ static int open_stat_pipe(void)
 	return ret;
 }
 
-static void audiod_pre_select(fd_set *rfds, fd_set *wfds, struct timeval *tv,
-		int *max_fileno)
+void signal_event_handler(struct task *t)
 {
-	int i, ret;
+	struct signal_task *st = t->private_data;
+	handle_signal(st->signum);
+}
 
-	FOR_EACH_SLOT(i) {
-		struct slot_info *s = &slot[i];
-		struct audio_format_info *a;
-		struct receiver_node *rn = s->receiver_node;
-		if (s->format < 0 || !rn)
-			continue;
-		a = &afi[s->format];
-		ret = a->receiver->pre_select(rn, rfds, wfds, tv);
-//		PARA_NOTICE_LOG("%s preselect: %d\n", a->receiver->name, ret);
-		*max_fileno = PARA_MAX(*max_fileno, ret);
+void signal_pre_select(struct sched *s, struct task *t)
+{
+	struct signal_task *st = t->private_data;
+	t->ret = 1;
+	para_fd_set(st->fd, &s->rfds, &s->max_fileno);
+}
+
+void signal_post_select(struct sched *s, struct task *t)
+{
+	struct signal_task *st = t->private_data;
+	t->ret = 1;
+	if (!FD_ISSET(st->fd, &s->rfds))
+		return;
+	t->ret = -E_SIGNAL_CAUGHT;
+	st->signum = para_next_signal();
+}
+
+void signal_setup_default(struct signal_task *st)
+{
+	st->task.pre_select = signal_pre_select;
+	st->task.post_select = signal_post_select;
+	st->task.private_data = st;
+	st->task.flags = 0;
+	sprintf(st->task.status, "signal task");
+}
+
+static void command_pre_select(struct sched *s, struct task *t)
+{
+	struct command_task *ct = t->private_data;
+	para_fd_set(ct->fd, &s->rfds, &s->max_fileno);
+
+}
+
+static void command_post_select(struct sched *s, struct task *t)
+{
+	int ret;
+	struct command_task *ct = t->private_data;
+
+	if (audiod_status != AUDIOD_OFF)
+		audiod_status_dump();
+	t->ret = 1; /* always successful */
+	if (!FD_ISSET(ct->fd, &s->rfds))
+		return;
+	ret = handle_connect();
+	if (ret < 0)
+		PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
+}
+
+void init_command_task(struct command_task *ct)
+{
+	ct->task.pre_select = command_pre_select;
+	ct->task.post_select = command_post_select;
+	ct->task.private_data = ct;
+	ct->task.flags = 0;
+	sprintf(ct->task.status, "command task");
+}
+
+static void status_pre_select(struct sched *s, struct task *t)
+{
+	struct status_task *st = t->private_data;
+	t->ret = 1;
+	if (st->fd >= 0 && audiod_status == AUDIOD_OFF)
+		close_stat_pipe();
+	if (st->fd < 0 && audiod_status != AUDIOD_OFF) {
+		st->fd = open_stat_pipe();
+		st->loaded = 0;
+		st->buf[0] = '\0';
 	}
+	if (st->fd >= 0 && audiod_status != AUDIOD_OFF)
+		para_fd_set(st->fd, &s->rfds, &s->max_fileno);
 }
-static void audiod_post_select(int select_ret, fd_set *rfds, fd_set *wfds)
+
+static void status_post_select(struct sched *s, struct task *t)
 {
-	int i, ret;
+	struct status_task *st = t->private_data;
+	int ret;
 
-	FOR_EACH_SLOT(i) {
-		struct slot_info *s = &slot[i];
-		struct audio_format_info *a;
-		struct receiver_node *rn = s->receiver_node;
-		if (s->format < 0 || !rn || rn->eof)
-			continue;
-		a = &afi[s->format];
-		ret = a->receiver->post_select(rn, select_ret, rfds, wfds);
-		if (ret <= 0) {
-			if (ret)
-				PARA_ERROR_LOG("%s post select failed: %s (slot %d)\n",
-				a->receiver->name, PARA_STRERROR(-ret), i);
-			else
-				PARA_INFO_LOG("eof in slot %d\n", i);
-			rn->eof = 1;
-		}
-		if (ret < 0 && s->fci)
-			s->fci->error = ret;
+	t->ret = 1;
+	if (st->fd < 0 || !FD_ISSET(st->fd, &s->rfds))
+		return;
+	ret = read(st->fd, st->buf + st->loaded,
+		STRINGSIZE - 1 - st->loaded);
+	if (ret <= 0) {
+		close_stat_pipe();
+		/* avoid busy loop if server is down */
+		while (sleep(1) > 0)
+			; /* try again*/
+	} else {
+		st->buf[ret + st->loaded] = '\0';
+		st->loaded = for_each_line(st->buf, ret + st->loaded,
+			&check_stat_line);
 	}
 }
 
+static void init_status_task(struct status_task *st)
+{
+	st->task.pre_select = status_pre_select;
+	st->task.post_select = status_post_select;
+	st->task.private_data = st;
+	st->task.flags = 0;
+	st->loaded = 0;
+	st->fd = -1;
+	st->buf[0] = '\0';
+	sprintf(st->task.status, "status task");
+}
+
+
+
+#if 0
 /* TODO: move everything before the select call to pre_select() */
 static void __noreturn audiod_mainloop(void)
 {
@@ -1556,12 +1713,15 @@ static void __noreturn audiod_mainloop(void)
 	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 */
-	FD_SET(signal_pipe, &rfds);
-	max_fileno = signal_pipe;
+	para_fd_set(signal_pipe, &rfds, &max_fileno);
 	para_fd_set(audiod_socket, &rfds, &max_fileno);
 
 	if (audiod_status != AUDIOD_ON)
@@ -1619,6 +1779,7 @@ repeat:
 	}
 	goto repeat;
 }
+#endif
 
 static void set_initial_status(void)
 {
@@ -1637,12 +1798,14 @@ static void set_initial_status(void)
 		PARA_WARNING_LOG("%s", "invalid mode\n");
 }
 
-int __noreturn main(int argc, char *argv[])
+int main(int argc, char *argv[])
 {
 	char *cf;
-	int i;
+	int ret, i;
+	struct sched s;
+
+	init_sched();
 
-	fprintf(stderr, "argc: %d\n", argc);
 	valid_fd_012();
 	hostname = para_hostname();
 	cmdline_parser(argc, argv, &conf);
@@ -1671,5 +1834,22 @@ int __noreturn main(int argc, char *argv[])
 	if (conf.daemon_given)
 		daemon_init();
 	audiod_get_socket(); /* doesn't return on errors */
-	audiod_mainloop();
+
+	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);
+
+	register_task(&sig_task->task);
+	register_task(&cmd_task->task);
+	register_task(&stat_task->task);
+	register_task(&at->task);
+	s.default_timeout.tv_sec = 0;
+	s.default_timeout.tv_usec = 999 * 1000;
+	ret = sched(&s);
+
+	PARA_EMERG_LOG("%s\n", PARA_STRERROR(-ret));
+	return EXIT_FAILURE;
 }