X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=http_recv.c;h=cf93076d0c9bcea5705d624b980f6a8eda5a1f41;hp=4283db4ba847dcd9a75025424bd48af36f1c799f;hb=1e012cf40238883621692051a22fb9c7cad5e944;hpb=7a0852896a75d852dbd2ac848ec5c0be3b5dea71;ds=inline diff --git a/http_recv.c b/http_recv.c index 4283db4b..cf93076d 100644 --- a/http_recv.c +++ b/http_recv.c @@ -1,11 +1,12 @@ /* - * Copyright (C) 2005-2007 Andre Noll + * Copyright (C) 2005-2009 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ /** \file http_recv.c paraslash's http receiver */ +#include #include #include @@ -14,11 +15,13 @@ #include "http.h" #include "list.h" #include "sched.h" +#include "ggo.h" #include "recv.h" #include "http_recv.cmdline.h" #include "net.h" #include "string.h" #include "fd.h" +#include "buffer_tree.h" /** the output buffer size of the http receiver */ #define BUFSIZE (32 * 1024) @@ -84,70 +87,99 @@ static char *make_request_msg(void) static void http_recv_pre_select(struct sched *s, struct task *t) { - struct receiver_node *rn = t->private_data; + struct receiver_node *rn = container_of(t, struct receiver_node, task); struct private_http_recv_data *phd = rn->private_data; - t->ret = 1; + t->error = 0; if (phd->status == HTTP_CONNECTED) para_fd_set(phd->fd, &s->wfds, &s->max_fileno); else para_fd_set(phd->fd, &s->rfds, &s->max_fileno); } +#define HTTP_RECV_MAX_PENDING (1024 * 1024) +#define HTTP_RECV_READ_BUF_SIZE 4096 static void http_recv_post_select(struct sched *s, struct task *t) { - struct receiver_node *rn = t->private_data; + struct receiver_node *rn = container_of(t, struct receiver_node, task); struct private_http_recv_data *phd = rn->private_data; + struct http_recv_args_info *conf = rn->conf; - t->ret = -E_HTTP_RECV_EOF; - if (rn->output_eof && *rn->output_eof) - goto out; - t->ret = 1; - if (!s->select_ret) - goto out; - if (phd->status == HTTP_CONNECTED) { + if (rn->output_error && *rn->output_error < 0) { + t->error = *rn->output_error; + goto err; + } + if (phd->status == HTTP_CONNECTED) { char *rq; if (!FD_ISSET(phd->fd, &s->wfds)) - goto out; + return; rq = make_request_msg(); - PARA_INFO_LOG("%s", "sending http request\n"); - t->ret = send_va_buffer(phd->fd, "%s", rq); + PARA_INFO_LOG("sending http request\n"); + t->error = send_va_buffer(phd->fd, "%s", rq); free(rq); - if (t->ret > 0) - phd->status = HTTP_SENT_GET_REQUEST; - goto out; + if (t->error < 0) + goto err; + phd->status = HTTP_SENT_GET_REQUEST; + return; } if (!FD_ISSET(phd->fd, &s->rfds)) - goto out; + return; if (phd->status == HTTP_SENT_GET_REQUEST) { - t->ret = recv_pattern(phd->fd, HTTP_OK_MSG, MAXLINE); - if (t->ret < 0) - goto out; - PARA_INFO_LOG("%s", "received ok msg, streaming\n"); - t->ret = 1; + t->error = recv_pattern(phd->fd, HTTP_OK_MSG, strlen(HTTP_OK_MSG)); + if (t->error < 0) + goto err; + PARA_INFO_LOG("received ok msg, streaming\n"); phd->status = HTTP_STREAMING; - goto out; + return; + } + if (conf->buffer_tree_given) { + char *buf; + + if (btr_no_children(rn->btrn)) { + t->error = -E_HTTP_RECV_NO_CHILD; + goto err; + } + if (btr_bytes_pending(rn->btrn) > HTTP_RECV_MAX_PENDING) { + t->error = -E_HTTP_RECV_OVERRUN; + goto err; + } + buf = para_malloc(HTTP_RECV_READ_BUF_SIZE); + t->error = recv_bin_buffer(phd->fd, buf, HTTP_RECV_READ_BUF_SIZE); + if (t->error == 0) + t->error = -E_RECV_EOF; + if (t->error < 0) { + free(buf); + goto err; + } + btr_add_output(buf, t->error, rn->btrn); + return; } - t->ret = -E_HTTP_RECV_OVERRUN; + t->error = -E_HTTP_RECV_OVERRUN; if (rn->loaded >= BUFSIZE) - goto out; - t->ret = recv_bin_buffer(phd->fd, rn->buf + rn->loaded, + goto err; + t->error = recv_bin_buffer(phd->fd, rn->buf + rn->loaded, BUFSIZE - rn->loaded); - if (t->ret <= 0) { - if (!t->ret) - t->ret = -E_HTTP_RECV_EOF; - goto out; + if (t->error == 0) + t->error = -E_RECV_EOF; + if (t->error < 0) + goto err; + rn->loaded += t->error; + return; +err: + if (conf->buffer_tree_given) { + btr_del_node(rn->btrn); + rn->btrn = NULL; } - rn->loaded += t->ret; -out: - if (t->ret < 0) - rn->eof = 1; } static void http_recv_close(struct receiver_node *rn) { struct private_http_recv_data *phd = rn->private_data; + struct http_recv_args_info *conf = rn->conf; + + if (conf->buffer_tree_given) + btr_del_node(rn->btrn); close(phd->fd); free(rn->buf); free(rn->private_data); @@ -167,17 +199,23 @@ static int http_recv_open(struct receiver_node *rn) { struct private_http_recv_data *phd; struct http_recv_args_info *conf = rn->conf; - int ret = makesock(AF_UNSPEC, IPPROTO_TCP, 0, conf->host_arg, conf->port_arg); + int fd, ret = makesock(AF_UNSPEC, IPPROTO_TCP, 0, conf->host_arg, + conf->port_arg); if (ret < 0) return ret; - + fd = ret; + ret = mark_fd_nonblocking(fd); + if (ret < 0) { + close(fd); + return ret; + } rn->buf = para_calloc(BUFSIZE); rn->private_data = phd = para_calloc(sizeof(struct private_http_recv_data)); - - phd->fd = ret; - mark_fd_nonblocking(phd->fd); + phd->fd = fd; phd->status = HTTP_CONNECTED; + if (conf->buffer_tree_given) + rn->btrn = btr_new_node("receiver", NULL, NULL, NULL); return 1; } @@ -190,10 +228,18 @@ static int http_recv_open(struct receiver_node *rn) */ void http_recv_init(struct receiver *r) { + struct http_recv_args_info dummy; + + http_recv_cmdline_parser_init(&dummy); r->open = http_recv_open; r->close = http_recv_close; r->pre_select = http_recv_pre_select; r->post_select = http_recv_post_select; r->shutdown = http_shutdown; r->parse_config = http_recv_parse_config; + r->help = (struct ggo_help) { + .short_help = http_recv_args_info_help, + .detailed_help = http_recv_args_info_detailed_help + }; + http_recv_cmdline_parser_free(&dummy); }