X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=stdin.c;h=8ef325b0fa9fea40ed18e40e96205a10851f6434;hp=8eb86dee69ba7322524e960bba6169916e9bef14;hb=d8213149049056e0cabbcbe575a10dfb1a83583e;hpb=badcf6125a8e0b2845348aca46d288f141e4f509 diff --git a/stdin.c b/stdin.c index 8eb86dee..8ef325b0 100644 --- a/stdin.c +++ b/stdin.c @@ -8,6 +8,8 @@ #include /* readdir() */ #include +#include +#include #include "para.h" #include "list.h" @@ -15,6 +17,8 @@ #include "fd.h" #include "error.h" #include "stdin.h" +#include "buffer_tree.h" +#include "string.h" /** * The pre_select function of the stdin task. @@ -42,6 +46,21 @@ static void stdin_pre_select(struct sched *s, struct task *t) para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno); } +static void stdin_pre_select_btr(struct sched *s, struct task *t) +{ + struct stdin_task *sit = container_of(t, struct stdin_task, task); + int ret; + + t->error = 0; + ret = btr_node_status(sit->btrn, 0, BTR_NT_ROOT); + if (ret > 0) + para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno); + else if (ret < 0) { + s->timeout.tv_sec = 0; + s->timeout.tv_usec = 1; + } +} + /** * The post select function of the stdin task. * @@ -76,6 +95,38 @@ static void stdin_post_select(struct sched *s, struct task *t) t->error = -E_STDIN_EOF; } +#define STDIN_INPUT_BUFFER_SIZE (1024 * 32) +static void stdin_post_select_btr(struct sched *s, struct task *t) +{ + struct stdin_task *sit = container_of(t, struct stdin_task, task); + ssize_t ret; + char *buf = NULL; + + t->error = 0; + ret = btr_node_status(sit->btrn, 0, BTR_NT_ROOT); + if (ret < 0) + goto err; + if (ret == 0) + return; + if (!FD_ISSET(STDIN_FILENO, &s->rfds)) + return; + buf = para_malloc(STDIN_INPUT_BUFFER_SIZE); + ret = read(STDIN_FILENO, buf, STDIN_INPUT_BUFFER_SIZE); + //PARA_CRIT_LOG("read ret: %d\n", ret); + if (ret < 0) + t->error = -ERRNO_TO_PARA_ERROR(errno); + if (ret == 0) + t->error = -E_STDIN_EOF; + if (t->error < 0) + goto err; + btr_add_output(buf, ret, sit->btrn); + return; +err: + free(buf); + btr_remove_node(sit->btrn); + t->error = ret; +} + /** * Initialize a stdin task structure with default values. * @@ -89,9 +140,14 @@ void stdin_set_defaults(struct stdin_task *sit) { int ret; - sit->bufsize = 16 * 1024, - sit->task.pre_select = stdin_pre_select; - sit->task.post_select = stdin_post_select; + sit->bufsize = 32 * 1024; + if (sit->btrn) { + sit->task.pre_select = stdin_pre_select_btr; + sit->task.post_select = stdin_post_select_btr; + } else { + sit->task.pre_select = stdin_pre_select; + sit->task.post_select = stdin_post_select; + } sprintf(sit->task.status, "stdin reader"); ret = mark_fd_nonblocking(STDIN_FILENO); if (ret >= 0)