X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=write_common.c;h=acf1d292a4a33ed8ed7ef2649382c9a2c3f4a3e8;hp=a7896fcb4ebd039e7401671c232ad2198bad8917;hb=8187d1acf18bfcf63db7b3a35ecb2ea3358fa2ad;hpb=780f86d5f849308b5100b087c6b223a6deef1dd7;ds=inline diff --git a/write_common.c b/write_common.c index a7896fcb..acf1d292 100644 --- a/write_common.c +++ b/write_common.c @@ -16,83 +16,117 @@ * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111, USA. */ +/** \file write_common.c common functions of para_audiod and para_write */ + #include "para.h" #include "string.h" +#include "list.h" +#include "sched.h" #include "write.h" #include "error.h" const char *writer_names[] ={WRITER_NAMES}; struct writer writers[NUM_SUPPORTED_WRITERS] = {WRITER_ARRAY}; -int wng_write(struct writer_node_group *g, char *buf, size_t *loaded) +static void wng_pre_select(__a_unused struct sched *s, struct task *t) +{ + struct writer_node_group *g = t->private_data; + int i; + + FOR_EACH_WRITER_NODE(i, g) { + struct writer_node *wn = &g->writer_nodes[i]; + t->ret = wn->writer->pre_select(s, wn); + if (t->ret < 0) { + g->eof = 1; + return; + } + } +} + +static void wng_post_select(struct sched *s, struct task *t) { - int ret, i, need_more_writes = 1; + struct writer_node_group *g = t->private_data; + int i; size_t min_written = 0; - while (need_more_writes) { - need_more_writes = 0; - FOR_EACH_WRITER_NODE(i, g) { - size_t w = g->written[i]; - unsigned char *p = buf + w; - int bytes_to_write; - struct writer_node *wn = &g->writer_nodes[i]; - if (!i) - min_written = w; - else - min_written = PARA_MIN(min_written, w); - if (w == *loaded) - continue; - if (!g->eof && (*loaded < wn->chunk_bytes + w)) - continue; - bytes_to_write = PARA_MIN(wn->chunk_bytes, - *loaded - w); - ret = wn->writer->write(p, bytes_to_write, wn); - if (ret < 0) - goto out; - if (ret != bytes_to_write) - PARA_WARNING_LOG("short write: %d/%d\n", ret, - bytes_to_write); - g->written[i] += ret; - need_more_writes = 1; + FOR_EACH_WRITER_NODE(i, g) { + struct writer_node *wn = &g->writer_nodes[i]; + t->ret = wn->writer->post_select(s, wn); + if (t->ret < 0) { + g->eof = 1; + return; } + if (!i) + min_written = wn->written; + else + min_written = PARA_MIN(min_written, wn->written); + } +// PARA_INFO_LOG("loaded: %zd, min_written: %zd bytes\n", *g->loaded, min_written); + if (min_written) { + *g->loaded -= min_written; + FOR_EACH_WRITER_NODE(i, g) + g->writer_nodes[i].written -= min_written; + } + if (!*g->loaded && *g->input_eof) { + g->eof = 1; + t->ret = -E_WNG_EOF; + return; + } + t->ret = 1; + if (*g->loaded && min_written) { +// PARA_INFO_LOG("moving %zd bytes\n", *g->loaded); + memmove(g->buf, g->buf + min_written, *g->loaded); } - *loaded -= min_written; - ret = 0; - if (g->eof) - goto out; - if (*loaded) - memmove(buf, buf + min_written, *loaded); - FOR_EACH_WRITER_NODE(i, g) - g->written[i] -= min_written; - ret = 1; -out: - return ret; } int wng_open(struct writer_node_group *g) { int i, ret = 1; + PARA_NOTICE_LOG("opening wng %p with %d writer(s)\n", g, g->num_writers); + register_task(&g->task); FOR_EACH_WRITER_NODE(i, g) { struct writer_node *wn = &g->writer_nodes[i]; + wn->wng = g; ret = wn->writer->open(wn); if (ret < 0) - goto out; + goto err_out; wn->chunk_bytes = ret; g->max_chunk_bytes = PARA_MAX(g->max_chunk_bytes, ret); } -out: + sprintf(g->task.status, "%s", "writer node group"); + g->eof = 0; + return 1; +err_out: + unregister_task(&g->task); + while (i > 0) { + struct writer_node *wn = &g->writer_nodes[--i]; + wn->writer->close(wn); + } + g->num_writers = 0; + g->eof = 1; return ret; } +void wng_unregister(struct writer_node_group *g) +{ + unregister_task(&g->task); + g->eof = 1; +} + void wng_close(struct writer_node_group *g) { int i; + if (!g) + return; + PARA_NOTICE_LOG("closing wng with %d writer(s)\n", g->num_writers); FOR_EACH_WRITER_NODE(i, g) { struct writer_node *wn = &g->writer_nodes[i]; wn->writer->close(wn); } + free(g->writer_nodes); + free(g); } struct writer_node_group *wng_new(unsigned num_writers) @@ -101,19 +135,12 @@ struct writer_node_group *wng_new(unsigned num_writers) g->num_writers = num_writers; g->writer_nodes = para_calloc(num_writers * sizeof(struct writer_node)); - g->written = para_calloc(num_writers * sizeof(size_t)); + g->task.private_data = g; + g->task.post_select = wng_post_select; + g->task.pre_select = wng_pre_select; return g; } -void wng_destroy(struct writer_node_group *g) -{ - if (!g) - return; - free(g->written); - free(g->writer_nodes); - free(g); -} - void init_supported_writers(void) { int i; @@ -122,22 +149,30 @@ void init_supported_writers(void) writers[i].init(&writers[i]); } -int check_writer_arg(const char *arg) +void *check_writer_arg(char *wa, int *writer_num) { - int i, ret = -E_WRITE_COMMON_SYNTAX; - char *a = para_strdup(arg), *p = strchr(a, ':'); - if (p) - *p = '\0'; - p++; + int i; + + *writer_num = -E_WRITE_COMMON_SYNTAX; + PARA_INFO_LOG("checking %s\n", wa); FOR_EACH_WRITER(i) { - if (strcmp(writer_names[i], a)) + const char *name = writer_names[i]; + size_t len = strlen(name); + char c; + if (strlen(wa) < len) + continue; + if (strncmp(name, wa, len)) continue; - ret = i; - goto out; + c = wa[len]; + if (c && c != ' ') + continue; + if (c && !writers[i].parse_config) + return NULL; + *writer_num = i; + return writers[i].parse_config(c? wa + len + 1 : ""); } -out: - free(a); - return ret; + PARA_ERROR_LOG("%s", "writer not found\n"); + return NULL; } struct writer_node_group *setup_default_wng(void) @@ -150,7 +185,8 @@ struct writer_node_group *setup_default_wng(void) else default_writer = 1; wng->writer_nodes[0].writer = &writers[default_writer]; - PARA_INFO_LOG("using default writer: %s\n", - writer_names[default_writer]); + PARA_INFO_LOG("using default writer: %s %p\n", + writer_names[default_writer], writers[default_writer].parse_config); + wng->writer_nodes[0].conf = writers[default_writer].parse_config(""); return wng; }