From: Andre Date: Sun, 28 May 2006 18:49:05 +0000 (+0200) Subject: integrate writers into para_audiod X-Git-Tag: v0.2.14~101^2~18 X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=commitdiff_plain;h=4b6458e9fc11eb741200dfebd8dc96a333956e4b;hp=4255eeea6a72f82d4b57a8c6f6b218ecd0d04afc integrate writers into para_audiod Seems to work, but might still be full of bugs. --- diff --git a/audiod.c b/audiod.c index dccc2e15..7f103789 100644 --- a/audiod.c +++ b/audiod.c @@ -36,6 +36,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 +55,47 @@ 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]; @@ -374,7 +375,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 +393,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,18 +466,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; @@ -505,22 +494,31 @@ static void close_receiver(int slot_num) s->receiver_node = NULL; set_restart_barrier(s->format, NULL); } +static void close_writer(int slot_num) +{ + struct slot_info *s = &slot[slot_num]; + if (s->wng) { + PARA_INFO_LOG("slot %d: closing writer node group\n", + slot_num); + wng_close(s->wng); + s->wng = NULL; + } + if (s->fc) + *s->fc->output_eof = 1; /* FIXME */ +} + 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); - } + close_writer(i); } static void check_sigchld(void) { pid_t pid; - int i; struct timeval now; gettimeofday(&now, NULL); @@ -528,21 +526,6 @@ 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; - } - } PARA_CRIT_LOG("para_client died (pid %d)\n", pid); goto reap_next_child; } @@ -558,7 +541,7 @@ static int get_empty_slot(void) clear_slot(i); return i; } - if (s->write_fd > 0 || s->wpid > 0) + if (s->wng) continue; if (s->receiver_node) continue; @@ -579,7 +562,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; @@ -621,6 +604,7 @@ static void __noreturn clean_exit(int status, const char *msg) exit(status); } +#if 0 __malloc static char *glob_cmd(char *cmd) { char *ret, *replacement; @@ -645,6 +629,7 @@ __malloc static char *glob_cmd(char *cmd) out: return ret; } +#endif /** get the number of filters for the given audio format */ int num_filters(int audio_format_num) @@ -664,17 +649,15 @@ 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; + 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; @@ -726,30 +709,53 @@ 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) { - int ret, fds[3] = {1, -1, -1}; + struct writer_node_group *g = t->private_data; + int i; + + PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret)); + unregister_task(t); + FOR_EACH_SLOT(i) { + if (slot[i].wng != g) + continue; + wng_close(g); + wng_destroy(g); + slot[i].wng = NULL; + } +} + +static void start_stream_writer(int slot_num, struct timeval *now) +{ + 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); + PARA_INFO_LOG("opening %s filters\n", audio_formats[s->format]); 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->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); + 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"); + } + ret = wng_open(s->wng); + s->wstime = *now; current_decoder = slot_num; activate_inactive_grab_clients(slot_num, s->format, &s->fc->filters); } @@ -936,71 +942,23 @@ static void check_timeouts(void) /* 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", + PARA_INFO_LOG("%s stream (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) { struct slot_info *s = &slot[slot_num]; - struct receiver_node *rn = s->receiver_node; if (s->format < 0) return; if (!s->fc) return; - if (!rn->eof && !s->fc->eof && s->wpid > 0) + if (s->wng) return; - 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->fc); close_filters(s->fc); @@ -1009,7 +967,7 @@ static void close_decoder_if_idle(int slot_num) clear_slot(slot_num); } -static void audiod_pre_select(struct sched *s, struct task *t) +static void audiod_pre_select(__a_unused struct sched *s, __a_unused struct task *t) { int i; @@ -1019,28 +977,20 @@ static void audiod_pre_select(struct sched *s, struct task *t) start_current_receiver(); check_timeouts(); FOR_EACH_SLOT(i) { - struct audio_format_info *a; struct receiver_node *rn; close_decoder_if_idle(i); - slot[i].wcheck = 0; if (slot[i].format < 0) continue; - a = &afi[slot[i].format]; rn = slot[i].receiver_node; - if (rn && rn->loaded && !slot[i].wpid) { + if (rn && rn->loaded && !slot[i].wng) { PARA_INFO_LOG("no writer in slot %d\n", i); - start_stream_writer(i); + start_stream_writer(i, &s->now); } - 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; } } +#if 0 static int write_audio_data(int slot_num) { struct slot_info *s = &slot[slot_num]; @@ -1077,20 +1027,22 @@ static int write_audio_data(int slot_num) gettimeofday(&s->wtime, NULL); return ret; } +#endif -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; + int 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); +// if (slot[i].write_fd <= 0 || !slot[i].wcheck +// || !FD_ISSET(slot[i].write_fd, &s->wfds)) +// continue; +// ret = write_audio_data(i); } } @@ -1172,15 +1124,37 @@ out: static int init_stream_io(void) { - int i, ret, receiver_num, nf; + int i, ret, receiver_num, nf, nw; char *cmd; + 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); + init_supported_writers(); + nw = PARA_MAX(1, conf.writer_given); + PARA_INFO_LOG("allocating space for %d writers\n", nw); + FOR_EACH_AUDIO_FORMAT(i) { + 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; - afi[ret].write_cmd = para_strdup(cmd); - PARA_INFO_LOG("%s write command: %s\n", audio_formats[ret], afi[ret].write_cmd); + a = &afi[ret]; + nw = a->num_writers; + wconf = check_writer_arg(cmd, &writer_num); + if (!wconf) { + ret = writer_num; + goto out; + } + a->writers[nw] = &writers[ret]; + a->writer_conf[nw] = wconf; + PARA_INFO_LOG("%s writer #%d: %s\n", audio_formats[ret], + nw, writer_names[writer_num]); + a->num_writers++; } for (i = 0; receivers[i].name; i++) { PARA_INFO_LOG("initializing %s receiver\n", receivers[i].name); @@ -1210,7 +1184,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); @@ -1224,7 +1198,7 @@ static int init_stream_io(void) nf = PARA_MAX(2, conf.filter_given) + 1; PARA_INFO_LOG("allocating space for %d filters\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) @@ -1825,6 +1799,7 @@ int main(int argc, char *argv[]) PARA_EMERG_LOG("init stream io error: %s\n", PARA_STRERROR(-i)); exit(EXIT_FAILURE); } +// return 0; server_uptime(UPTIME_SET); set_initial_status(); FOR_EACH_SLOT(i) diff --git a/audiod.ggo b/audiod.ggo index dd385a16..89cad59c 100644 --- a/audiod.ggo +++ b/audiod.ggo @@ -143,21 +143,26 @@ see --no_default_filters." string typestr="filter_spec" optional multiple -option "stream_write_cmd" w +option "writer" w #~~~~~~~~~~~~~~~~~~~~~~~~~~ "Specify stream writer. -May be given multiple times, once for each -supported audio format. Default value is -'para_write -w alsa' for both mp3 and ogg. -You can use the START_TIME() macro for these -commands. Each occurence of START_TIME() -gets replaced at runtime by the stream start -time announced by para_server, plus any -offsets." +May be given multiple times, even multiple +times for the same audio format. Default +value is 'alsa' for all supported audio +formats. You can use the START_TIME() macro +for these commands. Each occurence of +START_TIME() gets replaced at runtime by the +stream start time announced by para_server, +plus any offsets. - string typestr="format:command" +Example: + + -w 'aac:osx' + +" + string typestr="writer_spec" optional multiple diff --git a/stdout.h b/stdout.h index 5e45ace8..4dafafde 100644 --- a/stdout.h +++ b/stdout.h @@ -1,3 +1,4 @@ +/** \file stdout.h common code for uitlities that write to stdout */ struct stdout_task { char *buf; size_t *bufsize; diff --git a/write.c b/write.c index b692ac50..2efe07ce 100644 --- a/write.c +++ b/write.c @@ -161,6 +161,17 @@ out: return NULL; } +static void wng_event_handler(struct task *t) +{ + struct writer_node_group *g = t->private_data; + + PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret)); + unregister_task(t); + wng_close(g); + wng_destroy(g); +} + + static void idt_event_handler(struct task *t) { int ret; @@ -170,6 +181,7 @@ static void idt_event_handler(struct task *t) wng->buf = sit.buf; wng->loaded = &sit.loaded; wng->input_eof = &sit.eof; + wng->task.event_handler = wng_event_handler; ret = wng_open(wng); if (ret < 0) { PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret)); diff --git a/write_common.c b/write_common.c index de0502f5..7fb8d123 100644 --- a/write_common.c +++ b/write_common.c @@ -99,16 +99,6 @@ void wng_close(struct writer_node_group *g) } } -static void wng_event_handler(struct task *t) -{ - struct writer_node_group *g = t->private_data; - - PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret)); - unregister_task(t); - wng_close(g); - wng_destroy(g); -} - struct writer_node_group *wng_new(unsigned num_writers) { struct writer_node_group *g = para_calloc(sizeof(struct writer_node_group)); @@ -118,7 +108,6 @@ struct writer_node_group *wng_new(unsigned num_writers) g->written = para_calloc(num_writers * sizeof(size_t)); g->task.private_data = g; g->task.post_select = wng_post_select; - g->task.event_handler = wng_event_handler; g->task.flags = POST_ADD_TAIL; return g; } @@ -135,6 +124,7 @@ void *check_writer_arg(char *wa, int *writer_num) { int i; + *writer_num = -E_WRITE_COMMON_SYNTAX; PARA_INFO_LOG("checking %s\n", wa); FOR_EACH_WRITER(i) { const char *name = writer_names[i];