X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=stdin.c;h=abe69cafd4270ddf0e82ed9d2b30d3cf3603f43a;hp=f66d1bba325f557e09588cf6b256abce8a99bb53;hb=2b1f064a6357d027b4c06eee6cf191940726859e;hpb=e61e21f2aae608ff694c924cdeb2056ac1136b77 diff --git a/stdin.c b/stdin.c index f66d1bba..abe69caf 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,39 @@ 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) { + if (ret < 0) + ret = -ERRNO_TO_PARA_ERROR(errno); + else + ret = -E_STDIN_EOF; + 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 +141,14 @@ void stdin_set_defaults(struct stdin_task *sit) { int ret; - sit->bufsize = 32 * 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)