From 353b837f675fe6cade2daeeeb468fd16759a6c8e Mon Sep 17 00:00:00 2001 From: Roberto De Ioris Date: Sat, 20 Oct 2012 16:54:16 +0200 Subject: [PATCH] preliminary port of the http router to the new api --- core/buffer.c | 28 + plugins/fastrouter/fastrouter.c | 193 +++--- plugins/http/http.c | 1134 +++++++++---------------------- uwsgi.h | 2 + uwsgiconfig.py | 62 +- 5 files changed, 515 insertions(+), 904 deletions(-) diff --git a/core/buffer.c b/core/buffer.c index 010c2277..62dda6ce 100644 --- a/core/buffer.c +++ b/core/buffer.c @@ -14,6 +14,7 @@ struct uwsgi_buffer *uwsgi_buffer_new(size_t len) { } int uwsgi_buffer_fix(struct uwsgi_buffer *ub, size_t len) { + if (ub->limit >0 && len > ub->limit) return -1; if (ub->len < len) { char *new_buf = realloc(ub->buf, len); if (!new_buf) { @@ -26,12 +27,39 @@ int uwsgi_buffer_fix(struct uwsgi_buffer *ub, size_t len) { return 0; } +int uwsgi_buffer_ensure(struct uwsgi_buffer *ub, size_t len) { + size_t remains = ub->len - ub->pos; + if (remains < len) { + size_t new_len = ub->len + (len - remains); + if (ub->limit >0 && new_len > ub->limit) { + new_len = ub->limit; + if (new_len == ub->len) return -1; + } + char *new_buf = realloc(ub->buf, new_len); + if (!new_buf) { + uwsgi_error("uwsgi_buffer_ensure()"); + return -1; + } + ub->buf = new_buf; + ub->len = new_len; + } + return 0; +} + + int uwsgi_buffer_append(struct uwsgi_buffer *ub, char *buf, size_t len) { size_t remains = ub->len - ub->pos; if (len > remains) { size_t chunk_size = UMAX(len, (size_t) uwsgi.page_size); + if (ub->limit >0 && ub->len + chunk_size > ub->limit) { + // retry with anothr minimal size + if (len < (size_t) uwsgi.page_size) { + chunk_size = len; + } + if (ub->len + chunk_size > ub->limit) return -1; + } char *new_buf = realloc(ub->buf, ub->len + chunk_size); if (!new_buf) { uwsgi_error("uwsgi_buffer_append()"); diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c index 1efa0f13..bc065473 100644 --- a/plugins/fastrouter/fastrouter.c +++ b/plugins/fastrouter/fastrouter.c @@ -8,31 +8,31 @@ #include "../corerouter/cr.h" struct uwsgi_fastrouter { - struct uwsgi_corerouter cr; + struct uwsgi_corerouter cr; } ufr; extern struct uwsgi_server uwsgi; struct fastrouter_session { - struct corerouter_session crs; - struct uwsgi_buffer *post_buf; - size_t post_buf_max; - size_t post_buf_len; - off_t post_buf_pos; + struct corerouter_session crs; + struct uwsgi_buffer *post_buf; + size_t post_buf_max; + size_t post_buf_len; + off_t post_buf_pos; }; struct uwsgi_option fastrouter_options[] = { {"fastrouter", required_argument, 0, "run the fastrouter on the specified port", uwsgi_opt_corerouter, &ufr, 0}, {"fastrouter-processes", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.cr.processes, 0}, {"fastrouter-workers", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.cr.processes, 0}, - {"fastrouter-zerg", required_argument, 0, "attach the fastrouter to a zerg server", uwsgi_opt_corerouter_zerg, &ufr, 0 }, + {"fastrouter-zerg", required_argument, 0, "attach the fastrouter to a zerg server", uwsgi_opt_corerouter_zerg, &ufr, 0}, {"fastrouter-use-cache", no_argument, 0, "use uWSGI cache as hostname->server mapper for the fastrouter", uwsgi_opt_true, &ufr.cr.use_cache, 0}, {"fastrouter-use-pattern", required_argument, 0, "use a pattern for fastrouter hostname->server mapping", uwsgi_opt_corerouter_use_pattern, &ufr, 0}, {"fastrouter-use-base", required_argument, 0, "use a base dir for fastrouter hostname->server mapping", uwsgi_opt_corerouter_use_base, &ufr, 0}, {"fastrouter-fallback", required_argument, 0, "fallback to the specified node in case of error", uwsgi_opt_add_string_list, &ufr.cr.fallback, 0}, - + {"fastrouter-use-cluster", no_argument, 0, "load balance to nodes subscribed to the cluster", uwsgi_opt_true, &ufr.cr.use_cluster, 0}, {"fastrouter-use-code-string", required_argument, 0, "use code string as hostname->server mapper for the fastrouter", uwsgi_opt_corerouter_cs, &ufr, 0}, @@ -52,7 +52,7 @@ struct uwsgi_option fastrouter_options[] = { {"fastrouter-stats", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, {"fastrouter-stats-server", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, {"fastrouter-ss", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, - {"fastrouter-harakiri", required_argument, 0, "enable fastrouter harakiri", uwsgi_opt_set_int, &ufr.cr.harakiri, 0 }, + {"fastrouter-harakiri", required_argument, 0, "enable fastrouter harakiri", uwsgi_opt_set_int, &ufr.cr.harakiri, 0}, {0, 0, 0, 0, 0, 0, 0}, }; @@ -61,69 +61,70 @@ ssize_t fr_instance_read_response(struct corerouter_session *); ssize_t fr_read_body(struct corerouter_session *); void fr_get_hostname(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) { - - // here i use directly corerouter_session - struct corerouter_session *cs = (struct corerouter_session *) data; - - //uwsgi_log("%.*s = %.*s\n", keylen, key, vallen, val); - if (!uwsgi_strncmp("SERVER_NAME", 11, key, keylen) && !cs->hostname_len) { - cs->hostname = val; - cs->hostname_len = vallen; - return; - } - if (!uwsgi_strncmp("HTTP_HOST", 9, key, keylen) && !cs->has_key) { - cs->hostname = val; - cs->hostname_len = vallen; - return; - } - - if (!uwsgi_strncmp("UWSGI_FASTROUTER_KEY", 20, key, keylen)) { - cs->has_key = 1; - cs->hostname = val; - cs->hostname_len = vallen; - return; - } + // here i use directly corerouter_session + struct corerouter_session *cs = (struct corerouter_session *) data; - if (!uwsgi_strncmp("CONTENT_LENGTH", 14, key, keylen)) { - cs->post_cl = uwsgi_str_num(val, vallen); - return; - } + //uwsgi_log("%.*s = %.*s\n", keylen, key, vallen, val); + if (!uwsgi_strncmp("SERVER_NAME", 11, key, keylen) && !cs->hostname_len) { + cs->hostname = val; + cs->hostname_len = vallen; + return; + } + + if (!uwsgi_strncmp("HTTP_HOST", 9, key, keylen) && !cs->has_key) { + cs->hostname = val; + cs->hostname_len = vallen; + return; + } + + if (!uwsgi_strncmp("UWSGI_FASTROUTER_KEY", 20, key, keylen)) { + cs->has_key = 1; + cs->hostname = val; + cs->hostname_len = vallen; + return; + } + + if (!uwsgi_strncmp("CONTENT_LENGTH", 14, key, keylen)) { + cs->post_cl = uwsgi_str_num(val, vallen); + return; + } } -ssize_t fr_write_body(struct corerouter_session *cs) { +ssize_t fr_write_body(struct corerouter_session * cs) { struct fastrouter_session *fs = (struct fastrouter_session *) cs; - ssize_t len = write(cs->instance_fd, fs->post_buf->buf + fs->post_buf_pos, fs->post_buf_len - fs->post_buf_pos); - if (len < 0) { - cr_try_again; - uwsgi_error("fr_write_body()"); - return -1; - } + ssize_t len = write(cs->instance_fd, fs->post_buf->buf + fs->post_buf_pos, fs->post_buf_len - fs->post_buf_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("fr_write_body()"); + return -1; + } - fs->post_buf_pos += len; + fs->post_buf_pos += len; // the body chunk has been sent, start again reading from client and instance - if (fs->post_buf_pos == fs->post_buf_len) { - uwsgi_cr_hook_instance_write(cs, NULL); + if (fs->post_buf_pos == fs->post_buf_len) { + uwsgi_cr_hook_instance_write(cs, NULL); uwsgi_cr_hook_instance_read(cs, fr_instance_read_response); - uwsgi_cr_hook_read(cs, fr_read_body); - } + uwsgi_cr_hook_read(cs, fr_read_body); + } - return len; + return len; } -ssize_t fr_read_body(struct corerouter_session *cs) { +ssize_t fr_read_body(struct corerouter_session * cs) { struct fastrouter_session *fs = (struct fastrouter_session *) cs; ssize_t len = read(cs->fd, fs->post_buf->buf, fs->post_buf_max); if (len < 0) { - cr_try_again; - uwsgi_error("fr_read_body()"); - return -1; - } + cr_try_again; + uwsgi_error("fr_read_body()"); + return -1; + } // connection closed - if (len == 0) return 0; + if (len == 0) + return 0; fs->post_buf_len = len; fs->post_buf_pos = 0; @@ -132,36 +133,36 @@ ssize_t fr_read_body(struct corerouter_session *cs) { uwsgi_cr_hook_read(cs, NULL); uwsgi_cr_hook_instance_read(cs, NULL); uwsgi_cr_hook_instance_write(cs, fr_write_body); - + return len; } -ssize_t fr_write_response(struct corerouter_session *cs) { +ssize_t fr_write_response(struct corerouter_session * cs) { ssize_t len = write(cs->fd, cs->buffer->buf + cs->buffer_pos, cs->buffer_len - cs->buffer_pos); - if (len < 0) { - cr_try_again; - uwsgi_error("fr_write_response()"); - return -1; - } + if (len < 0) { + cr_try_again; + uwsgi_error("fr_write_response()"); + return -1; + } - cs->buffer_pos += len; + cs->buffer_pos += len; // ok this response chunk is sent, let's wait for another one - if (cs->buffer_pos == cs->buffer_len) { + if (cs->buffer_pos == cs->buffer_len) { uwsgi_cr_hook_write(cs, NULL); - uwsgi_cr_hook_instance_read(cs, fr_instance_read_response); - } + uwsgi_cr_hook_instance_read(cs, fr_instance_read_response); + } - return len; + return len; } -ssize_t fr_instance_read_response(struct corerouter_session *cs) { +ssize_t fr_instance_read_response(struct corerouter_session * cs) { ssize_t len = read(cs->instance_fd, cs->buffer->buf, cs->buffer->len); if (len < 0) { cr_try_again; - uwsgi_error("fr_instance_read_response()"); - return -1; - } + uwsgi_error("fr_instance_read_response()"); + return -1; + } // end of the response if (len == 0) { @@ -176,7 +177,7 @@ ssize_t fr_instance_read_response(struct corerouter_session *cs) { return len; } -ssize_t fr_instance_send_request(struct corerouter_session *cs) { +ssize_t fr_instance_send_request(struct corerouter_session * cs) { ssize_t len = write(cs->instance_fd, cs->buffer->buf + cs->buffer_pos, cs->uh.pktsize - cs->buffer_pos); if (len < 0) { cr_try_again; @@ -202,34 +203,35 @@ ssize_t fr_instance_send_request(struct corerouter_session *cs) { fs->post_buf_max = UMIN(UMAX16, cs->post_cl); } fs->post_buf = uwsgi_buffer_new(fs->post_buf_max); - if (!fs->post_buf) return -1; + if (!fs->post_buf) + return -1; uwsgi_cr_hook_read(cs, fr_read_body); } return len; } -ssize_t fr_instance_send_request_header(struct corerouter_session *cs) { +ssize_t fr_instance_send_request_header(struct corerouter_session * cs) { ssize_t len = write(cs->instance_fd, &cs->uh + cs->buffer_pos, 4 - cs->buffer_pos); - if (len < 0) { - cr_try_again; - uwsgi_error("fr_instance_send_request_header()"); - return -1; - } + if (len < 0) { + cr_try_again; + uwsgi_error("fr_instance_send_request_header()"); + return -1; + } - cs->buffer_pos += len; + cs->buffer_pos += len; - // ok the request is sent, we can start sending client body (if any) and we can start waiting - // for response - if (cs->buffer_pos == 4) { - cs->buffer_pos = 0; - uwsgi_cr_hook_instance_write(cs, fr_instance_send_request); - } + // ok the request is sent, we can start sending client body (if any) and we can start waiting + // for response + if (cs->buffer_pos == 4) { + cs->buffer_pos = 0; + uwsgi_cr_hook_instance_write(cs, fr_instance_send_request); + } - return len; + return len; } -ssize_t fr_instance_connected(struct corerouter_session *cs) { +ssize_t fr_instance_connected(struct corerouter_session * cs) { socklen_t solen = sizeof(int); @@ -253,16 +255,17 @@ ssize_t fr_instance_connected(struct corerouter_session *cs) { return 1; } -ssize_t fr_recv_uwsgi_vars(struct corerouter_session *cs) { +ssize_t fr_recv_uwsgi_vars(struct corerouter_session * cs) { // increase buffer if needed - if (uwsgi_buffer_fix(cs->buffer, cs->uh.pktsize)) return -1; + if (uwsgi_buffer_fix(cs->buffer, cs->uh.pktsize)) + return -1; ssize_t len = read(cs->fd, cs->buffer->buf + cs->buffer_pos, cs->uh.pktsize - cs->buffer_pos); if (len < 0) { cr_try_again; uwsgi_error("fr_recv_uwsgi_vars()"); return -1; } - + cs->buffer_pos += len; // headers received, ready to choose the instance @@ -273,9 +276,11 @@ ssize_t fr_recv_uwsgi_vars(struct corerouter_session *cs) { return -1; } // check the hostname; - if (cs->hostname_len == 0) return -1; + if (cs->hostname_len == 0) + return -1; // find an instance using the key - if (cs->corerouter->mapper(cs->corerouter, cs)) return -1; + if (cs->corerouter->mapper(cs->corerouter, cs)) + return -1; // check instance if (cs->instance_address_len == 0) { // if fallback nodes are configured, trigger them @@ -304,7 +309,7 @@ ssize_t fr_recv_uwsgi_vars(struct corerouter_session *cs) { return len; } -ssize_t fr_recv_uwsgi_header(struct corerouter_session *cs) { +ssize_t fr_recv_uwsgi_header(struct corerouter_session * cs) { ssize_t len = read(cs->fd, cs->buffer->buf + cs->buffer_pos, 4 - cs->buffer_pos); if (len < 0) { cr_try_again; @@ -333,7 +338,7 @@ int fastrouter_init() { ufr.cr.session_size = sizeof(struct fastrouter_session); ufr.cr.alloc_session = fastrouter_alloc_session; - uwsgi_corerouter_init((struct uwsgi_corerouter *) &ufr); + uwsgi_corerouter_init((struct uwsgi_corerouter *) &ufr); return 0; } diff --git a/plugins/http/http.c b/plugins/http/http.c index 1e1612e8..16ff11dc 100644 --- a/plugins/http/http.c +++ b/plugins/http/http.c @@ -1,12 +1,6 @@ /* - uWSGI http - - requires: - - - async - - caching - - pcre (optional) + uWSGI httprouter */ @@ -16,16 +10,6 @@ extern struct uwsgi_server uwsgi; #include "../corerouter/cr.h" -#ifdef __sun__ -#define MAX_HTTP_VEC IOV_MAX*64 -#else -#ifdef IOV_MAX -#define MAX_HTTP_VEC IOV_MAX -#else -#define MAX_HTTP_VEC 128 -#endif -#endif - #ifdef UWSGI_SSL #define UWSGI_HTTP_SSL 1 #define UWSGI_HTTP_FORCE_SSL 2 @@ -41,8 +25,8 @@ struct uwsgi_http { int manage_expect; int raw_body; - int keepalive; + #ifdef UWSGI_SSL int https_export_cert; #endif @@ -50,77 +34,12 @@ struct uwsgi_http { } uhttp; -#ifdef UWSGI_SSL -void uwsgi_opt_https(char *opt, char *value, void *cr) { - struct uwsgi_corerouter *ucr = (struct uwsgi_corerouter *) cr; - char *client_ca = NULL; - - // build socket, certificate and key file - char *sock = uwsgi_str(value); - char *crt = strchr(sock, ','); - if (!crt) { - uwsgi_log("invalid https syntax must be socket,crt,key\n"); - exit(1); - } - *crt = '\0'; crt++; - char *key = strchr(crt, ','); - if (!key) { - uwsgi_log("invalid https syntax must be socket,crt,key\n"); - exit(1); - } - *key = '\0'; key++; - - char *ciphers = strchr(key, ','); - if (ciphers) { - *ciphers = '\0'; ciphers++; - client_ca = strchr(ciphers, ','); - if (client_ca) { - *client_ca = '\0'; client_ca++; - } - } - - struct uwsgi_gateway_socket *ugs = uwsgi_new_gateway_socket(sock, ucr->name); - // ok we have the socket, initialize ssl if required - if (!uwsgi.ssl_initialized) { - uwsgi_ssl_init(); - } - - // initialize ssl context - ugs->ctx = uwsgi_ssl_new_server_context(uwsgi_concat3(ucr->short_name, "-", ugs->name),crt, key, ciphers, client_ca); - // set the ssl mode - ugs->mode = UWSGI_HTTP_SSL; - - ucr->has_sockets++; -} - -void uwsgi_opt_http_to_https(char *opt, char *value, void *cr) { - struct uwsgi_corerouter *ucr = (struct uwsgi_corerouter *) cr; - - char *sock = uwsgi_str(value); - char *port = strchr(sock, ','); - if (port) { - *port = '\0'; - port++; - } - - struct uwsgi_gateway_socket *ugs = uwsgi_new_gateway_socket(sock, ucr->name); - - // set context to the port - ugs->ctx = port; - // force SSL mode - ugs->mode = UWSGI_HTTP_FORCE_SSL; - - ucr->has_sockets++; -} - -#endif - struct uwsgi_option http_options[] = { {"http", required_argument, 0, "add an http router/server on the specified address", uwsgi_opt_corerouter, &uhttp, 0}, #ifdef UWSGI_SSL - {"https", required_argument, 0, "add an https router/server on the specified address with specified certificate and key", uwsgi_opt_https, &uhttp, 0}, + //{"https", required_argument, 0, "add an https router/server on the specified address with specified certificate and key", uwsgi_opt_https, &uhttp, 0}, {"https-export-cert", no_argument, 0, "export uwsgi variable HTTPS_CC containing the raw client certificate", uwsgi_opt_true, &uhttp.https_export_cert, 0}, - {"http-to-https", required_argument, 0, "add an http router/server on the specified address and redirect all of the requests to https", uwsgi_opt_http_to_https, &uhttp, 0}, + //{"http-to-https", required_argument, 0, "add an http router/server on the specified address and redirect all of the requests to https", uwsgi_opt_http_to_https, &uhttp, 0}, #endif {"http-processes", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.cr.processes, 0}, {"http-workers", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.cr.processes, 0}, @@ -137,7 +56,7 @@ struct uwsgi_option http_options[] = { {"http-subscription-server", required_argument, 0, "enable the subscription server", uwsgi_opt_corerouter_ss, &uhttp, 0}, {"http-timeout", required_argument, 0, "set internal http socket timeout", uwsgi_opt_set_int, &uhttp.cr.socket_timeout, 0}, {"http-manage-expect", no_argument, 0, "manage the Expect HTTP request header", uwsgi_opt_true, &uhttp.manage_expect, 0}, - {"http-keepalive", no_argument, 0, "support HTTP keepalive (non-pipelined) requests (requires backend support)", uwsgi_opt_true, &uhttp.keepalive, 0}, + {"http-keepalive", no_argument, 0, "experimental HTTP keepalive support (non-pipelined) requests (requires backend support)", uwsgi_opt_true, &uhttp.keepalive, 0}, {"http-raw-body", no_argument, 0, "blindly send HTTP body to backends (required for WebSockets and Icecast support)", uwsgi_opt_true, &uhttp.raw_body, 0}, @@ -157,28 +76,19 @@ struct uwsgi_option http_options[] = { struct http_session { - struct corerouter_session crs; - - struct uwsgi_header uh; + struct corerouter_session cs; int rnrn; - char *ptr; char *port; int port_len; - struct iovec iov[MAX_HTTP_VEC]; - int iov_len; - char uss[MAX_HTTP_VEC * 2]; - - char buffer[UMAX16]; - size_t buffer_len; - char path_info[UMAX16]; - uint16_t path_info_len; - char *request_uri; uint16_t request_uri_len; + char *path_info; + uint16_t path_info_len; + size_t received_body; #ifdef UWSGI_SSL @@ -189,25 +99,34 @@ struct http_session { char *ssl_cc; #endif + struct uwsgi_buffer *uwsgi_req; + int send_excpect_100; + in_addr_t ip_addr; char ip[INET_ADDRSTRLEN]; + struct uwsgi_buffer *post_buf; + size_t post_buf_max; + size_t post_buf_len; + off_t post_buf_pos; + }; #ifdef UWSGI_SSL int uwsgi_http_ssl_shutdown(struct http_session *, int); #endif -uint16_t http_add_uwsgi_header(struct http_session *h_session, struct iovec *iov, char *strsize1, char *strsize2, char *hh, uint16_t hhlen, int *c) { +int http_add_uwsgi_header(struct http_session *hs, char *hh, uint16_t hhlen) { + + struct corerouter_session *cs = &hs->cs; + struct uwsgi_buffer *buf = hs->uwsgi_req; int i; int status = 0; char *val = hh; uint16_t keylen = 0, vallen = 0; int prefix = 0; - - if (*c >= MAX_HTTP_VEC) - return 0; + char strsize[2]; for (i = 0; i < hhlen; i++) { if (!status) { @@ -230,104 +149,80 @@ uint16_t http_add_uwsgi_header(struct http_session *h_session, struct iovec *iov } if (!keylen) - return 0; - - if ((*c) + 4 >= MAX_HTTP_VEC) - return 0; + return -1; if (!uwsgi_strncmp("HOST", 4, hh, keylen)) { - h_session->crs.hostname = val; - h_session->crs.hostname_len = vallen; + cs->hostname = val; + cs->hostname_len = vallen; } if (!uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) { - h_session->crs.post_cl = uwsgi_str_num(val, vallen); + cs->post_cl = uwsgi_str_num(val, vallen); } if (uwsgi_strncmp("CONTENT_TYPE", 12, hh, keylen) && uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) { keylen += 5; prefix = 1; - if ((*c) + 5 >= MAX_HTTP_VEC) - return 0; } - strsize1[0] = (uint8_t) (keylen & 0xff); - strsize1[1] = (uint8_t) ((keylen >> 8) & 0xff); + strsize[0] = (uint8_t) (keylen & 0xff); + strsize[1] = (uint8_t) ((keylen >> 8) & 0xff); - iov[*c].iov_base = strsize1; - iov[*c].iov_len = 2; - *c += 1; + if (uwsgi_buffer_append(buf, strsize, 2)) return -1; if (prefix) { - iov[*c].iov_base = "HTTP_"; - iov[*c].iov_len = 5; - *c += 1; + if (uwsgi_buffer_append(buf, "HTTP_", 5)) return -1; } - iov[*c].iov_base = hh; - iov[*c].iov_len = keylen - (prefix * 5); - *c += 1; + if (uwsgi_buffer_append(buf, hh, keylen - (prefix * 5))) return -1; - strsize2[0] = (uint8_t) (vallen & 0xff); - strsize2[1] = (uint8_t) ((vallen >> 8) & 0xff); + strsize[0] = (uint8_t) (vallen & 0xff); + strsize[1] = (uint8_t) ((vallen >> 8) & 0xff); - iov[*c].iov_base = strsize2; - iov[*c].iov_len = 2; - *c += 1; + if (uwsgi_buffer_append(buf, strsize, 2)) return -1; + if (uwsgi_buffer_append(buf, val, vallen)) return -1; - iov[*c].iov_base = val; - iov[*c].iov_len = vallen; - *c += 1; - - return 2 + keylen + 2 + vallen; + return 0; } -uint16_t http_add_uwsgi_var(struct iovec * iov, char *strsize1, char *strsize2, char *key, uint16_t keylen, char *val, uint16_t vallen, int *c) { +int http_add_uwsgi_var(struct http_session *hs, char *key, uint16_t keylen, char *val, uint16_t vallen) { - if ((*c) + 4 >= MAX_HTTP_VEC) - return 0; + struct uwsgi_buffer *buf = hs->uwsgi_req; + char strsize[2]; - strsize1[0] = (uint8_t) (keylen & 0xff); - strsize1[1] = (uint8_t) ((keylen >> 8) & 0xff); + strsize[0] = (uint8_t) (keylen & 0xff); + strsize[1] = (uint8_t) ((keylen >> 8) & 0xff); - iov[*c].iov_base = strsize1; - iov[*c].iov_len = 2; - *c += 1; + if (uwsgi_buffer_append(buf, strsize, 2)) return -1; + if (uwsgi_buffer_append(buf, key, keylen)) return -1; - iov[*c].iov_base = key; - iov[*c].iov_len = keylen; - *c += 1; + strsize[0] = (uint8_t) (vallen & 0xff); + strsize[1] = (uint8_t) ((vallen >> 8) & 0xff); - strsize2[0] = (uint8_t) (vallen & 0xff); - strsize2[1] = (uint8_t) ((vallen >> 8) & 0xff); + if (uwsgi_buffer_append(buf, strsize, 2)) return -1; + if (uwsgi_buffer_append(buf, val, vallen)) return -1; - iov[*c].iov_base = strsize2; - iov[*c].iov_len = 2; - *c += 1; - - iov[*c].iov_base = val; - iov[*c].iov_len = vallen; - *c += 1; - - return 2 + keylen + 2 + vallen; + return 0; } -int http_parse(struct http_session *h_session) { +int http_parse(struct http_session *h_session, size_t http_req_len) { - char *ptr = h_session->buffer; - char *watermark = h_session->ptr; + char *ptr = h_session->cs.buffer->buf; + char *watermark = ptr + http_req_len; char *base = ptr; - // leave a slot for uwsgi header - int c = 1; char *query_string = NULL; char *protocol = NULL; size_t protocol_len = 0; + h_session->uwsgi_req = uwsgi_buffer_new(uwsgi.page_size); + if (!h_session->uwsgi_req) return -1; + h_session->uwsgi_req->limit = UMAX16; + // REQUEST_METHOD while (ptr < watermark) { if (*ptr == ' ') { - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "REQUEST_METHOD", 14, base, ptr - base, &c); + if (http_add_uwsgi_var(h_session, "REQUEST_METHOD", 14, base, ptr - base)) return -1; ptr++; break; } @@ -340,23 +235,25 @@ int http_parse(struct http_session *h_session) { if (*ptr == '?' && !query_string) { // PATH_INFO must be url-decoded !!! h_session->path_info_len = ptr - base; + h_session->path_info = uwsgi_malloc(h_session->path_info_len); http_url_decode(base, &h_session->path_info_len, h_session->path_info); - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "PATH_INFO", 9, h_session->path_info, h_session->path_info_len, &c); + if (http_add_uwsgi_var(h_session, "PATH_INFO", 9, h_session->path_info, h_session->path_info_len)) return -1; query_string = ptr + 1; } else if (*ptr == ' ') { h_session->request_uri = base; h_session->request_uri_len = ptr - base; - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "REQUEST_URI", 11, base, ptr - base, &c); + if (http_add_uwsgi_var(h_session, "REQUEST_URI", 11, base, ptr - base)) return -1; if (!query_string) { // PATH_INFO must be url-decoded !!! h_session->path_info_len = ptr - base; + h_session->path_info = uwsgi_malloc(h_session->path_info_len); http_url_decode(base, &h_session->path_info_len, h_session->path_info); - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "PATH_INFO", 9, h_session->path_info, h_session->path_info_len, &c); - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "QUERY_STRING", 12, "", 0, &c); + if (http_add_uwsgi_var(h_session, "PATH_INFO", 9, h_session->path_info, h_session->path_info_len)) return -1; + if (http_add_uwsgi_var(h_session, "QUERY_STRING", 12, "", 0)) return -1; } else { - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "QUERY_STRING", 12, query_string, ptr - query_string, &c); + if (http_add_uwsgi_var(h_session, "QUERY_STRING", 12, query_string, ptr - query_string)) return -1; } ptr++; break; @@ -372,7 +269,7 @@ int http_parse(struct http_session *h_session) { return 0; if (*(ptr + 1) != '\n') return 0; - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SERVER_PROTOCOL", 15, base, ptr - base, &c); + if (http_add_uwsgi_var(h_session, "SERVER_PROTOCOL", 15, base, ptr - base)) return -1; protocol = base; protocol_len = ptr - base; ptr += 2; @@ -382,29 +279,29 @@ int http_parse(struct http_session *h_session) { } // SCRIPT_NAME - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SCRIPT_NAME", 11, "", 0, &c); + if (http_add_uwsgi_var(h_session, "SCRIPT_NAME", 11, "", 0)) return -1; // SERVER_NAME - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SERVER_NAME", 11, uwsgi.hostname, uwsgi.hostname_len, &c); - h_session->crs.hostname = uwsgi.hostname; - h_session->crs.hostname_len = uwsgi.hostname_len; + if (http_add_uwsgi_var(h_session, "SERVER_NAME", 11, uwsgi.hostname, uwsgi.hostname_len)) return -1; + h_session->cs.hostname = uwsgi.hostname; + h_session->cs.hostname_len = uwsgi.hostname_len; // SERVER_PORT - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SERVER_PORT", 11, h_session->port, h_session->port_len, &c); + if (http_add_uwsgi_var(h_session, "SERVER_PORT", 11, h_session->port, h_session->port_len)) return -1; // UWSGI_ROUTER - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "UWSGI_ROUTER", 12, "http", 4, &c); + if (http_add_uwsgi_var(h_session, "UWSGI_ROUTER", 12, "http", 4)) return -1; #ifdef UWSGI_SSL // HTTPS (adapted from nginx) - if (h_session->crs.ugs->mode == UWSGI_HTTP_SSL) { - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "HTTPS", 5, "on", 2, &c); + if (h_session->cs.ugs->mode == UWSGI_HTTP_SSL) { + if (http_add_uwsgi_var(h_session, "HTTPS", 5, "on", 2)) return -1; h_session->ssl_client_cert = SSL_get_peer_certificate(h_session->ssl); if (h_session->ssl_client_cert) { X509_NAME *name = X509_get_subject_name(h_session->ssl_client_cert); if (name) { h_session->ssl_client_dn = X509_NAME_oneline(name, NULL, 0); - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "HTTPS_DN", 8, h_session->ssl_client_dn, strlen(h_session->ssl_client_dn), &c); + if (http_add_uwsgi_var(h_session, "HTTPS_DN", 8, h_session->ssl_client_dn, strlen(h_session->ssl_client_dn))) return -1; } if (uhttp.https_export_cert) { h_session->ssl_bio = BIO_new(BIO_s_mem()); @@ -413,7 +310,7 @@ int http_parse(struct http_session *h_session) { size_t cc_len = BIO_pending(h_session->ssl_bio); h_session->ssl_cc = uwsgi_malloc(cc_len); BIO_read(h_session->ssl_bio, h_session->ssl_cc, cc_len); - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "HTTPS_CC", 8, h_session->ssl_cc, cc_len, &c); + if (http_add_uwsgi_var(h_session, "HTTPS_CC", 8, h_session->ssl_cc, cc_len)) return -1; } } } @@ -423,7 +320,7 @@ int http_parse(struct http_session *h_session) { // REMOTE_ADDR if (inet_ntop(AF_INET, &h_session->ip_addr, h_session->ip, INET_ADDRSTRLEN)) { - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "REMOTE_ADDR", 11, h_session->ip, strlen(h_session->ip), &c); + if (http_add_uwsgi_var(h_session, "REMOTE_ADDR", 11, h_session->ip, strlen(h_session->ip))) return -1; } else { uwsgi_error("inet_ntop()"); @@ -451,11 +348,10 @@ int http_parse(struct http_session *h_session) { // this is an hack with dumb/wrong/useless error checking if (uhttp.manage_expect) { if (!uwsgi_strncmp("Expect: 100-continue", 20, base, ptr - base)) { - if (h_session->crs.send(&uhttp.cr, &h_session->crs, protocol, protocol_len) == (ssize_t) protocol_len) - h_session->crs.send(&uhttp.cr, &h_session->crs, " 100 Continue\r\n\r\n", 17); + h_session->send_excpect_100 = 1; } } - h_session->uh.pktsize += http_add_uwsgi_header(h_session, h_session->iov, h_session->uss + c, h_session->uss + c + 2, base, ptr - base, &c); + if (http_add_uwsgi_header(h_session, base, ptr - base)) return -1; ptr++; base = ptr + 1; } @@ -466,648 +362,276 @@ int http_parse(struct http_session *h_session) { while (hv) { char *equal = strchr(hv->value, '='); if (equal) { - h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, hv->value, equal - hv->value, equal + 1, strlen(equal + 1), &c); + if (http_add_uwsgi_var(h_session, hv->value, equal - hv->value, equal + 1, strlen(equal + 1))) return -1; } hv = hv->next; } - // security check - if (c >= MAX_HTTP_VEC-4) { - uwsgi_log("too much headers in request. skipping it.\n"); - return 0; - } - - return c; + return 0; } -void uwsgi_http_switch_events(struct uwsgi_corerouter *ucr, struct corerouter_session *cs, int interesting_fd) { +ssize_t hr_instance_read_response(struct corerouter_session *); +ssize_t hr_read_body(struct corerouter_session *); - ssize_t len; - int j; +ssize_t hr_write_body(struct corerouter_session * cs) { struct http_session *hs = (struct http_session *) cs; -#ifndef __sun__ - struct msghdr msg; - union { - struct cmsghdr cmsg; - char control[CMSG_SPACE(sizeof(int))]; - } msg_control; - struct cmsghdr *cmsg; -#endif - char bbuf[UMAX16]; + ssize_t len = write(cs->instance_fd, hs->post_buf->buf + hs->post_buf_pos, hs->post_buf_len - hs->post_buf_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_write_body()"); + return -1; + } + + hs->post_buf_pos += len; + + // the body chunk has been sent, start again reading from client and instance + if (hs->post_buf_pos == hs->post_buf_len) { + uwsgi_cr_hook_instance_write(cs, NULL); + uwsgi_cr_hook_instance_read(cs, hr_instance_read_response); + uwsgi_cr_hook_read(cs, hr_read_body); + } + + return len; +} + + +ssize_t hr_read_body(struct corerouter_session * cs) { + struct http_session *hs = (struct http_session *) cs; + ssize_t len = read(cs->fd, hs->post_buf->buf, hs->post_buf_max); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_read_body()"); + return -1; + } + + // connection closed + if (len == 0) + return 0; + + hs->post_buf_len = len; + hs->post_buf_pos = 0; + + // ok we have a body, stop reading from the client and the instance and start writing to the instance + uwsgi_cr_hook_read(cs, NULL); + uwsgi_cr_hook_instance_read(cs, NULL); + uwsgi_cr_hook_instance_write(cs, hr_write_body); + + return len; +} + + +ssize_t hr_write_response(struct corerouter_session * cs) { + ssize_t len = write(cs->fd, cs->buffer->buf + cs->buffer_pos, cs->buffer_len - cs->buffer_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_write_response()"); + return -1; + } + + cs->buffer_pos += len; + + // ok this response chunk is sent, let's wait for another one + if (cs->buffer_pos == cs->buffer_len) { + uwsgi_cr_hook_write(cs, NULL); + uwsgi_cr_hook_instance_read(cs, hr_instance_read_response); + } + + return len; +} + + +ssize_t hr_instance_read_response(struct corerouter_session * cs) { + ssize_t len = read(cs->instance_fd, cs->buffer->buf, cs->buffer->len); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_instance_read_response()"); + return -1; + } + + // end of the response + if (len == 0) { + return 0; + } + + cs->buffer_pos = 0; + cs->buffer_len = len; + // ok stop reading from the instance, and start writing to the client + uwsgi_cr_hook_instance_read(cs, NULL); + uwsgi_cr_hook_write(cs, hr_write_response); + return len; +} + + +ssize_t hr_instance_send_request(struct corerouter_session * cs) { + struct http_session *hs = (struct http_session *) cs; + ssize_t len = write(cs->instance_fd, hs->uwsgi_req->buf + cs->buffer_pos, cs->uh.pktsize - cs->buffer_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_instance_send_request()"); + return -1; + } + + cs->buffer_pos += len; + + // ok the request is sent, we can start sending client body (if any) and we can start waiting + // for response + if (cs->buffer_pos == cs->uh.pktsize) { + cs->buffer_pos = 0; + // stop writing to the instance + uwsgi_cr_hook_instance_write(cs, NULL); + // start reading from the instance + uwsgi_cr_hook_instance_read(cs, hr_instance_read_response); + // re-start reading from the client (for body or connection close) + // allocate a buffer for client body (could be delimited or dynamic) + hs->post_buf_max = UMAX16; + if (cs->post_cl > 0) { + hs->post_buf_max = UMIN(UMAX16, cs->post_cl); + } + hs->post_buf = uwsgi_buffer_new(hs->post_buf_max); + if (!hs->post_buf) + return -1; + uwsgi_cr_hook_read(cs, hr_read_body); + } + + return len; +} + + +ssize_t hr_instance_send_request_header(struct corerouter_session * cs) { + ssize_t len = write(cs->instance_fd, &cs->uh + cs->buffer_pos, 4 - cs->buffer_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("fr_instance_send_request_header()"); + return -1; + } + + cs->buffer_pos += len; + + // ok the request is sent, we can start sending client body (if any) and we can start waiting + // for response + if (cs->buffer_pos == 4) { + cs->buffer_pos = 0; + uwsgi_cr_hook_instance_write(cs, hr_instance_send_request); + } + + return len; +} + + +ssize_t hr_instance_connected(struct corerouter_session * cs) { socklen_t solen = sizeof(int); - switch (cs->status) { - - - case COREROUTER_STATUS_RECV_HDR: - if (interesting_fd == -1) { - goto choose_node; - } - - - len = cs->recv(&uhttp.cr, cs, hs->buffer + cs->pos, UMAX16 - cs->pos); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(ucr->queue, cs->fd); -#endif - if (len <= 0) { - // check for blocking operation on non-blocking socket - if (len < 0 && errno == EINPROGRESS) break; - corerouter_close_session(ucr, cs); - break; - } - - cs->pos += len; - - for (j = 0; j < len; j++) { - if (*hs->ptr == '\r' && (hs->rnrn == 0 || hs->rnrn == 2)) { - hs->rnrn++; - } - else if (*hs->ptr == '\r') { - hs->rnrn = 1; - } - else if (*hs->ptr == '\n' && hs->rnrn == 1) { - hs->rnrn = 2; - } - else if (*hs->ptr == '\n' && hs->rnrn == 3) { - - hs->ptr++; - cs->post_remains = len - (j + 1); - hs->iov_len = http_parse(hs); - - if (hs->iov_len == 0 || cs->hostname_len == 0) { - corerouter_close_session(ucr, cs); - break; - } - -#ifdef UWSGI_SSL - if (cs->ugs->mode == UWSGI_HTTP_FORCE_SSL) { - if (hs->request_uri_len > 0) { - char *https_url; - char *colon = memchr(cs->hostname, ':', cs->hostname_len); - if (colon) { - if (cs->ugs->ctx) { - https_url = uwsgi_concat4n(cs->hostname, colon-cs->hostname, ":", 1, cs->ugs->ctx, strlen(cs->ugs->ctx), hs->request_uri, hs->request_uri_len); - } - else { - https_url = uwsgi_concat2n(cs->hostname, colon-cs->hostname, hs->request_uri, hs->request_uri_len); - } - } - else { - if (cs->ugs->ctx) { - https_url = uwsgi_concat4n(cs->hostname, cs->hostname_len, ":", 1, cs->ugs->ctx, strlen(cs->ugs->ctx), hs->request_uri, hs->request_uri_len); - } - else { - https_url = uwsgi_concat2n(cs->hostname, cs->hostname_len, hs->request_uri, hs->request_uri_len); - } - } - struct iovec iov[4]; - iov[0].iov_base = "HTTP/1.0 301 Moved Permanently\r\n"; - iov[0].iov_len = 32; - iov[1].iov_base = "Location: https://"; - iov[1].iov_len = 18; - iov[2].iov_base = https_url; - iov[2].iov_len = strlen(https_url); - iov[3].iov_base = "\r\n\r\n"; - iov[3].iov_len = 4; - if (writev(cs->fd, iov, 4) <= 0) { - uwsgi_error("writev()"); - } - free(https_url); - } - corerouter_close_session(ucr, cs); - break; - } -#endif - - // the mapper hook - choose_node: - if (ucr->mapper(ucr, cs)) - break; - - - // no address found - if (!cs->instance_address_len) { - // if fallback nodes are configured, trigger them - if (ucr->fallback) { - cs->instance_failed = 1; - } - corerouter_close_session(ucr, cs); - break; - } - - - cs->pass_fd = is_unix(cs->instance_address, cs->instance_address_len); - - cs->instance_fd = uwsgi_connectn(cs->instance_address, cs->instance_address_len, 0, 1); - - if (cs->instance_fd < 0) { - cs->instance_failed = 1; - cs->soopt = errno; - corerouter_close_session(ucr, cs); - break; - } - - - cs->status = COREROUTER_STATUS_CONNECTING; - ucr->cr_table[cs->instance_fd] = cs; - event_queue_add_fd_write(ucr->queue, cs->instance_fd); - break; - - - } - else { - hs->rnrn = 0; - } - - hs->ptr++; - } - - - break; - - - case COREROUTER_STATUS_CONNECTING: - - if (interesting_fd == cs->instance_fd) { - - if (getsockopt(cs->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&cs->soopt), &solen) < 0) { - uwsgi_error("getsockopt()"); - cs->instance_failed = 1; - corerouter_close_session(ucr, cs); - break; - } - - if (cs->soopt) { - cs->instance_failed = 1; - corerouter_close_session(ucr, cs); - break; - } - - -#ifdef __BIG_ENDIAN__ - hs->uh.pktsize = uwsgi_swap16(cs->uh.pktsize); -#endif - hs->uh.modifier1 = cs->modifier1; - - hs->iov[0].iov_base = &hs->uh; - hs->iov[0].iov_len = 4; - - if (cs->post_remains > 0) { - hs->iov[hs->iov_len].iov_base = hs->ptr; - if (cs->post_remains > cs->post_cl) { - cs->post_remains = cs->post_cl; - } - hs->iov[hs->iov_len].iov_len = cs->post_remains; - hs->received_body += cs->post_remains; - hs->iov_len++; - } - - // increment node requests counter - if (cs->un) - cs->un->requests++; - -#ifndef __sun__ - // fd passing: PERFORMANCE EXTREME BOOST !!! - if (cs->pass_fd && !cs->post_remains && !uwsgi.no_fd_passing) { - msg.msg_name = NULL; - msg.msg_namelen = 0; - msg.msg_iov = hs->iov; - msg.msg_iovlen = hs->iov_len; - msg.msg_flags = 0; - msg.msg_control = &msg_control; - msg.msg_controllen = sizeof(msg_control); - - cmsg = CMSG_FIRSTHDR(&msg); - cmsg->cmsg_len = CMSG_LEN(sizeof(int)); - cmsg->cmsg_level = SOL_SOCKET; - cmsg->cmsg_type = SCM_RIGHTS; - - memcpy(CMSG_DATA(cmsg), &cs->fd, sizeof(int)); - - if (sendmsg(cs->instance_fd, &msg, 0) < 0) { - uwsgi_error("sendmsg()"); - } - - corerouter_close_session(ucr, cs); - break; - } - -#endif -#ifdef __sun__ - if (hs->iov_len > IOV_MAX) { - int remains = hs->iov_len; - int iov_len; - while (remains) { - if (remains > IOV_MAX) { - iov_len = IOV_MAX; - } - else { - iov_len = remains; - } - if (writev(cs->instance_fd, hs->iov + (hs->iov_len - remains), iov_len) <= 0) { - uwsgi_error("writev()"); - corerouter_close_session(ucr, cs); - break; - } - remains -= iov_len; - } - } -#else - if (writev(cs->instance_fd, hs->iov, hs->iov_len) <= 0) { - uwsgi_error("writev()"); - corerouter_close_session(ucr, cs); - break; - } -#endif - - event_queue_fd_write_to_read(ucr->queue, cs->instance_fd); - cs->status = COREROUTER_STATUS_RESPONSE; - } - - break; - - case COREROUTER_STATUS_RESPONSE: - - // data from instance - if (interesting_fd == cs->instance_fd) { - // retry later - if (cs->instance_stopped) { - break; - } - // writable ? - if (cs->instance_fd_state) { - len = cs->instance_send(&uhttp.cr, cs, NULL, 0); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_write(ucr->queue, cs->instance_fd); -#endif - if (len <= 0) { - if (len < 0 && errno == EINPROGRESS) break; - corerouter_close_session(ucr, cs); - } - break; - } - - len = cs->instance_recv(&uhttp.cr, cs, hs->buffer, UMAX16); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(ucr->queue, cs->instance_fd); -#endif - if (len <= 0) { - if (len < 0 && errno == EINPROGRESS) break; -/* -Keep-Alive implementation. -As soon as the backend close the connection, enable it. -The server will start waiting for another request (for a maximum of --http-socket seconds timeout) -To have a reliable implementation, we need to reset a bunch of values -*/ - if (len == 0) { - if (uhttp.keepalive) { -#ifdef UWSGI_DEBUG - uwsgi_log("Keep-Alive enabled\n"); -#endif - cs->keepalive = 1; - corerouter_close_session(ucr, cs); - cs->status = COREROUTER_STATUS_RECV_HDR; - hs->ptr = hs->buffer; - hs->rnrn = 0; - cs->pos = 0; - hs->received_body = 0; - cs->post_cl = 0; - cs->instance_fd = -1; - hs->uh.pktsize = 0; - cs->post_remains = 0; - cs->instance_address_len = 0; - cs->hostname_len = 0; - break; - } -#ifdef UWSGI_SSL - if (cs->ugs->mode == UWSGI_HTTP_SSL) { - int ssd_ret = SSL_shutdown(hs->ssl); - // it could fail or success, in both cases close the connection - if (ssd_ret != 0) { - corerouter_close_session(ucr, cs); - break; - } - cs->status = HTTP_SSL_STATUS_SHUTDOWN; - if (uwsgi_http_ssl_shutdown(hs, 1) != 0) { - corerouter_close_session(ucr, cs); - } - break; - } - // something to send in the queue ? - if (cs->write_queue) { - cs->write_queue_close = 1; - break; - } -#endif - } - corerouter_close_session(ucr, cs); - break; - } - - hs->buffer_len = len; - len = cs->send(&uhttp.cr, cs, hs->buffer, len); - if (len <= 0) { - if (len < 0 && errno == EINPROGRESS) break; - // check for blocking operation non non-blocking socket - corerouter_close_session(ucr, cs); - break; - } - - - // update transfer statistics - if (cs->un) - cs->un->transferred += len; - } - - // body from client or client ready to receive - else if (interesting_fd == cs->fd) { - - // writable ? - if (cs->fd_state) { - len = cs->send(&uhttp.cr, cs, hs->buffer,hs->buffer_len); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_write(ucr->queue, cs->fd); -#endif - if (len <= 0) { - if (len < 0 && errno == EINPROGRESS) break; - corerouter_close_session(ucr, cs); - } - break; - } - - len = cs->recv(&uhttp.cr, cs, bbuf, UMAX16); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(ucr->queue, cs->fd); -#endif - if (len <= 0) { - // check for blocking operation on non-blocking socket - if (len < 0 && errno == EINPROGRESS) break; - corerouter_close_session(ucr, cs); - break; - } - - if (cs->post_cl == 0 && uhttp.raw_body) goto raw; - - // avoid pipelined input - if (hs->received_body >= cs->post_cl) { - break; - } - - - if (len + hs->received_body > cs->post_cl) { - len = cs->post_cl - hs->received_body; - } - -raw: - len = cs->instance_send(&uhttp.cr, cs, bbuf, len); - - if (len <= 0) { - if (len < 0 && errno == EINPROGRESS) break; - corerouter_close_session(ucr, cs); - break; - } - - hs->received_body += len; - - } - - break; - -#ifdef UWSGI_SSL - case HTTP_SSL_STATUS_SHUTDOWN: - if (uwsgi_http_ssl_shutdown(hs, 0) != 0) { - corerouter_close_session(ucr, cs); - } - break; -#endif - - - // fallback to destroy !!! - default: - uwsgi_log("unknown event: closing session\n"); - corerouter_close_session(ucr, cs); - break; - } - -} - -void http_setup() { - uhttp.cr.name = uwsgi_str("uWSGI http"); - uhttp.cr.short_name = uwsgi_str("http"); -} - -#ifdef UWSGI_SSL -int uwsgi_http_ssl_shutdown(struct http_session *hs, int state) { - int ret = 0; - if (!state) { - ret = SSL_shutdown(hs->ssl); - } - if (ret == 1) return 1; - int err = SSL_get_error(hs->ssl, ret); - if (err == SSL_ERROR_WANT_READ) { - if (hs->crs.fd_state) { - event_queue_fd_write_to_read(uhttp.cr.queue, hs->crs.fd); - hs->crs.fd_state = 0; - } - return 0; - } - else if (err == SSL_ERROR_WANT_WRITE) { - if (!hs->crs.fd_state) { - event_queue_fd_read_to_write(uhttp.cr.queue, hs->crs.fd); - hs->crs.fd_state = 1; - } - return 0; - } - return -1; -} -ssize_t uwsgi_http_ssl_recv(struct uwsgi_corerouter *cr, struct corerouter_session *cs, char *buf, size_t len) { - struct http_session *hs = (struct http_session *) cs; - int ret = SSL_read(hs->ssl, buf, len); - if (ret > 0) { - if (hs->crs.fd_state) { - event_queue_fd_write_to_read(cr->queue, hs->crs.fd); - hs->crs.fd_state = 0; - } - return ret; - } - if (ret == 0) return 0; - int err = SSL_get_error(hs->ssl, ret); - - if (err == SSL_ERROR_WANT_READ) { - if (hs->crs.fd_state) { - event_queue_fd_write_to_read(cr->queue, hs->crs.fd); - hs->crs.fd_state = 0; - } - errno = EINPROGRESS; - return -1; - } - - else if (err == SSL_ERROR_WANT_WRITE) { - if (!hs->crs.fd_state) { - event_queue_fd_read_to_write(cr->queue, hs->crs.fd); - hs->crs.fd_state = 1; - } - errno = EINPROGRESS; - return -1; - } - - else if (err == SSL_ERROR_SYSCALL) { - uwsgi_error("SSL_read()"); - } - - else if (err == SSL_ERROR_SSL && uwsgi.ssl_verbose) { - ERR_print_errors_fp(stderr); - } - - return -1; - -} - -ssize_t uwsgi_http_ssl_send(struct uwsgi_corerouter *cr, struct corerouter_session *cs, char *buf, size_t len) { - struct http_session *hs = (struct http_session *) cs; - int ret = SSL_write(hs->ssl, buf, len); - if (ret > 0) { - if (cs->instance_stopped) { - event_queue_add_fd_read(cr->queue, cs->instance_fd); - cs->instance_stopped = 0; - } - if (cs->fd_state) { - event_queue_fd_write_to_read(cr->queue, cs->fd); - cs->fd_state = 0; - } - return ret; - } - int err = SSL_get_error(hs->ssl, ret); - if (err == SSL_ERROR_WANT_READ) { - if (cs->instance_fd != -1) { - event_queue_del_fd(cr->queue, cs->instance_fd, event_queue_read()); - cs->instance_stopped = 1; - } - if (cs->fd_state) { - event_queue_fd_write_to_read(cr->queue, cs->fd); - cs->fd_state = 0; - } - errno = EINPROGRESS; - return -1; - } - else if (err == SSL_ERROR_WANT_WRITE) { - if (cs->instance_fd != -1) { - event_queue_del_fd(cr->queue, cs->instance_fd, event_queue_read()); - cs->instance_stopped = 1; - } - if (!cs->fd_state) { - event_queue_fd_read_to_write(cr->queue, cs->fd); - cs->fd_state = 1; - } - errno = EINPROGRESS; - return -1; - } - - else if (err == SSL_ERROR_SYSCALL) { - uwsgi_error("SSL_write()"); - } - - else if (err == SSL_ERROR_SSL && uwsgi.ssl_verbose) { - ERR_print_errors_fp(stderr); - } - - else if (err == SSL_ERROR_ZERO_RETURN) { - return 0; - } - - return -1; -} - -// free ssl memory -void uwsgi_ssl_close(struct uwsgi_corerouter *ucr, struct corerouter_session *cs) { - struct http_session *hs = (struct http_session *) cs; - - if (hs->ssl_client_dn) { - OPENSSL_free(hs->ssl_client_dn); - } - - if (hs->ssl_cc) { - free(hs->ssl_cc); - } - - if (hs->ssl_bio) { - BIO_free(hs->ssl_bio); - } - - if (hs->ssl_client_cert) { - X509_free(hs->ssl_client_cert); - } - - if (!cs->keepalive) - SSL_free(hs->ssl); - -} -#endif - - -ssize_t uwsgi_http_nb_send(struct uwsgi_corerouter *cr, struct corerouter_session *cs, char *buf, size_t len) { - struct http_session *hs = (struct http_session *) cs; - ssize_t ret = write(cs->fd, buf, len); - if (ret == (ssize_t) len) { - if (cs->instance_fd != -1 && cs->instance_stopped) { - event_queue_add_fd_read(cr->queue, cs->instance_fd); - cs->instance_stopped = 0; - } - if (cs->fd_state) { - event_queue_fd_write_to_read(cr->queue, cs->fd); - cs->fd_state = 0; - } - return len; - } - else if (ret == 0) { - return -1; - } - else if (ret < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) { - if (cs->instance_fd != -1 && !cs->instance_stopped) { - event_queue_del_fd(cr->queue, cs->instance_fd, event_queue_read()); - cs->instance_stopped = 1; - } - if (!cs->fd_state) { - event_queue_fd_read_to_write(cr->queue, cs->fd); - cs->fd_state = 1; - } - errno = EINPROGRESS; - return -1; - } - uwsgi_error("write()"); + // first check for errors + if (getsockopt(cs->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&cs->soopt), &solen) < 0) { + uwsgi_error("hr_instance_connected()/getsockopt()"); + cs->instance_failed = 1; return -1; } - // partial write - hs->buffer_len -= ret; - memcpy(hs->buffer, hs->buffer + ret, hs->buffer_len); - // stop reading from the instance - if (cs->instance_fd != -1 && !cs->instance_stopped) { - event_queue_del_fd(cr->queue, cs->instance_fd, event_queue_read()); - cs->instance_stopped = 1; + if (cs->soopt) { + cs->instance_failed = 1; + return -1; } - // wait for write from client - if (!cs->fd_state) { - event_queue_fd_read_to_write(cr->queue, cs->fd); - cs->fd_state = 1; - } + cs->buffer_pos = 0; + cs->uh.modifier1 = cs->modifier1; + cs->uh.modifier2 = 0; + struct http_session *hs = (struct http_session *) cs; + cs->uh.pktsize = hs->uwsgi_req->pos; - errno = EINPROGRESS; - return -1; + // ok instance is connected, wait for write again + uwsgi_cr_hook_instance_write(cs, hr_instance_send_request_header); + // return a value > 0 + return 1; } +ssize_t hr_recv_http(struct corerouter_session * cs) { + // be sure buffer does not grow over 64k + cs->buffer->limit = UMAX16; + // try to always leave 4k available + if (uwsgi_buffer_ensure(cs->buffer, uwsgi.page_size)) return -1; + ssize_t len = read(cs->fd, cs->buffer->buf + cs->buffer_pos, cs->buffer->len - cs->buffer_pos); + if (len < 0) { + cr_try_again; + uwsgi_error("hr_recv_http()"); + return -1; + } + // fix buffer usage (TODO a bit ugly...) + cs->buffer->pos += len; + + // read until \r\n\r\n is found + int j; + char *ptr = cs->buffer->buf + cs->buffer_pos; + struct http_session *hs = (struct http_session *) cs; + cs->buffer_pos += len; + for (j = 0; j < len; j++) { + if (*ptr == '\r' && (hs->rnrn == 0 || hs->rnrn == 2)) { + hs->rnrn++; + } + else if (*ptr == '\r') { + hs->rnrn = 1; + } + else if (*ptr == '\n' && hs->rnrn == 1) { + hs->rnrn = 2; + } + else if (*ptr == '\n' && hs->rnrn == 3) { + cs->post_remains = len - (j + 1); + // parse HTTP request + size_t http_req_len = (cs->buffer_pos-len)+j; + if (http_parse(hs, http_req_len)) return -1; + // check for a valid hostname + if (cs->hostname_len == 0) return -1; + + // get instance name + if (cs->corerouter->mapper(cs->corerouter, cs)) + return -1; + + struct uwsgi_corerouter *ucr = cs->corerouter; + + if (cs->instance_address_len == 0) { + // if fallback nodes are configured, trigger them + if (ucr->fallback) { + cs->instance_failed = 1; + } + return -1; + } + + // stop receiving from the client + uwsgi_cr_hook_read(cs, NULL); + + // start async connect + cs->instance_fd = uwsgi_connectn(cs->instance_address, cs->instance_address_len, 0, 1); + if (cs->instance_fd < 0) { + cs->instance_failed = 1; + cs->soopt = errno; + return -1; + } + // map the instance + cs->corerouter->cr_table[cs->instance_fd] = cs; + // wait for connection + uwsgi_cr_hook_instance_write(cs, hr_instance_connected); + } + else { + hs->rnrn = 0; + } + ptr++; + } + + return len; +} void http_alloc_session(struct uwsgi_corerouter *ucr, struct uwsgi_gateway_socket *ugs, struct corerouter_session *cs, struct sockaddr *sa, socklen_t s_len) { struct http_session *hs = (struct http_session *) cs; - hs->ptr = hs->buffer; + //hs->ptr = hs->buffer; cs->modifier1 = uhttp.modifier1; - cs->send = uwsgi_http_nb_send; + //cs->send = uwsgi_http_nb_send; if (sa && sa->sa_family == AF_INET) { hs->ip_addr = ((struct sockaddr_in *) sa)->sin_addr.s_addr; } + + hs->rnrn = 0; + if (ugs) { hs->port = ugs->port; hs->port_len = ugs->port_len; @@ -1116,18 +640,25 @@ void http_alloc_session(struct uwsgi_corerouter *ucr, struct uwsgi_gateway_socke hs->ssl = SSL_new(ugs->ctx); SSL_set_fd(hs->ssl, cs->fd); SSL_set_accept_state(hs->ssl); - cs->recv = uwsgi_http_ssl_recv; - cs->send = uwsgi_http_ssl_send; - cs->close = uwsgi_ssl_close; + } + else { +#endif + uwsgi_cr_hook_read(cs, hr_recv_http); +#ifdef UWSGI_SSL } #endif } } +void http_setup() { + uhttp.cr.name = uwsgi_str("uWSGI http"); + uhttp.cr.short_name = uwsgi_str("http"); +} + + int http_init() { uhttp.cr.session_size = sizeof(struct http_session); - uhttp.cr.switch_events = uwsgi_http_switch_events; uhttp.cr.alloc_session = http_alloc_session; if (uhttp.cr.has_sockets && !uwsgi.sockets && !uwsgi_courerouter_has_has_backends(&uhttp.cr)) { uwsgi_new_socket(uwsgi_concat2("127.0.0.1:0", "")); @@ -1139,7 +670,6 @@ int http_init() { return 0; } - struct uwsgi_plugin http_plugin = { .name = "http", diff --git a/uwsgi.h b/uwsgi.h index 5cda78b5..4726adde 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -292,6 +292,7 @@ struct uwsgi_buffer { char *buf; off_t pos; size_t len; + size_t limit; }; struct uwsgi_string_list { @@ -3294,6 +3295,7 @@ void uwsgi_set_sockets_protocols(void); struct uwsgi_buffer *uwsgi_buffer_new(size_t); int uwsgi_buffer_append(struct uwsgi_buffer *, char *, size_t); int uwsgi_buffer_fix(struct uwsgi_buffer *, size_t); +int uwsgi_buffer_ensure(struct uwsgi_buffer *, size_t); void uwsgi_buffer_destroy(struct uwsgi_buffer *); void uwsgi_httpize_var(char *, size_t); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 684d52e9..6a2b23f8 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -12,6 +12,8 @@ uwsgi_cpu = os.uname()[4] import sys import subprocess +from threading import Thread +from Queue import Queue from distutils import sysconfig @@ -72,6 +74,32 @@ report['spooler'] = False report['debug'] = False report['plugin_dir'] = False report['ipv6'] = False + +compile_queue = None +thread_compilers = [] + +def thread_compiler(num): + while True: + (objfile, cmdline) = compile_queue.get() + if objfile: + print("[thread %d][%s] %s" % (num, GCC, objfile)) + ret = os.system(cmdline) + if ret != 0: + os._exit(1) + elif cmdline: + print cmdline + else: + return + + +if CPUCOUNT > 1: + compile_queue = Queue(maxsize=CPUCOUNT) + for i in range(0,CPUCOUNT): + t = Thread(target=thread_compiler,args=(i,)) + t.daemon = True + t.start() + thread_compilers.append(t) + def binarize(name): return name.replace('/', '_').replace('.','_').replace('-','_') @@ -134,6 +162,21 @@ def add_o(x): x = x + '.o' return x +def push_print(msg): + if not compile_queue: + print(msg) + else: + compile_queue.put((None, msg)) + +def push_command(objfile, cmdline): + if not compile_queue: + print("[%s] %s" % (GCC, objfile)) + ret = os.system(cmdline) + if ret != 0: + sys.exit(1) + else: + compile_queue.put((objfile, cmdline)) + def compile(cflags, last_cflags_ts, objfile, srcfile): source_stat = os.stat(srcfile) @@ -159,10 +202,7 @@ def compile(cflags, last_cflags_ts, objfile, srcfile): except: pass cmdline = "%s -c %s -o %s %s" % (GCC, cflags, objfile, srcfile) - print("[%s] %s" % (GCC, objfile)) - ret = os.system(cmdline) - if ret != 0: - sys.exit(1) + push_command(objfile, cmdline) def build_uwsgi(uc, print_only=False): @@ -220,7 +260,7 @@ def build_uwsgi(uc, print_only=False): cflags.append('-DUWSGI_CFLAGS=\\"%s\\"' % uwsgi_cflags) cflags.append('-DUWSGI_BUILD_DATE="\\"%s\\""' % time.strftime("%d %B %Y %H:%M:%S")) - print("*** uWSGI compiling server core ***") + push_print("*** uWSGI compiling server core ***") for file in gcc_list: objfile = file if objfile == 'uwsgi': @@ -232,7 +272,7 @@ def build_uwsgi(uc, print_only=False): ep = uc.get('embedded_plugins').split(',') if len(ep) > 0: - print("*** uWSGI compiling embedded plugins ***") + push_print("*** uWSGI compiling embedded plugins ***") for p in ep: if p is None or p == 'None': continue @@ -313,11 +353,11 @@ def build_uwsgi(uc, print_only=False): plugins = uc.get('plugins').split(',') if len(plugins) > 0: - print("*** uWSGI building plugins ***") + push_print("*** uWSGI building plugins ***") for p in plugins: p = p.strip() - print("*** building plugin: %s ***" % p) + push_print("*** building plugin: %s ***" % p) build_plugin("plugins/%s" % p, uc, cflags, ldflags, libs) bin_name = os.environ.get('UWSGI_BIN_NAME', uc.get('bin_name')) @@ -327,6 +367,12 @@ def build_uwsgi(uc, print_only=False): for ef in binary_list: gcc_list.append("build/%s" % ef) + if compile_queue: + for t in thread_compilers: + compile_queue.put((None, None)) + for t in thread_compilers: + t.join() + print("*** uWSGI linking ***") ldline = "%s -o %s %s %s %s" % (GCC, bin_name, ' '.join(uniq_warnings(ldflags)), ' '.join(map(add_o, gcc_list)), ' '.join(uniq_warnings(libs)))