From 8fdef7a1f9bfc02d20a3b9bf05957b79c72734a8 Mon Sep 17 00:00:00 2001 From: Unbit Date: Sun, 9 Jun 2013 03:02:16 +0200 Subject: [PATCH] added chunked input api --- core/chunked.c | 165 ++++++++++++++++++++++++++++++++ core/utils.c | 7 +- plugins/python/uwsgi_pymodule.c | 39 ++++++++ uwsgi.h | 11 +++ uwsgiconfig.py | 2 +- 5 files changed, 222 insertions(+), 2 deletions(-) create mode 100644 core/chunked.c diff --git a/core/chunked.c b/core/chunked.c new file mode 100644 index 00000000..9067462a --- /dev/null +++ b/core/chunked.c @@ -0,0 +1,165 @@ +#include + +extern struct uwsgi_server uwsgi; + +/* + + Chunked input implementation + + --chunked-input-limit (default 1MB) + + --chunked-input-timeout (default --socket-timeout) + + chunk = uwsgi.chunked_read([timeout]) + + timeout = -1 (wait forever) + timeout = 0 (default) + +*/ + +static ssize_t uwsgi_chunked_input_recv(struct wsgi_request *wsgi_req, int timeout, int nb) { + + if (timeout == 0) timeout = uwsgi.chunked_input_timeout; + if (timeout == 0) timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]; + + int ret = -1; + + for(;;) { + ssize_t rlen = wsgi_req->socket->proto_read_body(wsgi_req, wsgi_req->chunked_input_buf->buf + wsgi_req->chunked_input_buf->pos, wsgi_req->chunked_input_buf->len - wsgi_req->chunked_input_buf->pos); + if (rlen > 0) return rlen; + if (rlen == 0) return -1; + if (rlen < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) { + if (nb) return -1; + goto wait; + } + uwsgi_error("uwsgi_chunked_input_recv()"); + return -1; + } + +wait: + ret = uwsgi.wait_read_hook(wsgi_req->fd, timeout); + if (ret > 0) { + rlen = wsgi_req->socket->proto_read_body(wsgi_req, wsgi_req->chunked_input_buf->buf + wsgi_req->chunked_input_buf->pos, wsgi_req->chunked_input_buf->len - wsgi_req->chunked_input_buf->pos); + if (rlen > 0) return rlen; + if (rlen <= 0) return -1; + } + if (ret < 0) { + uwsgi_error("uwsgi_chunked_input_recv()"); + } + return -1; + } + + return -1; +} + +static ssize_t uwsgi_chunked_readline(struct wsgi_request *wsgi_req) { + size_t i; + int found = 0; + for(i=0;ichunked_input_buf->pos;i++) { + if (found) { + if (wsgi_req->chunked_input_buf->buf[i] == '\n') { + if (uwsgi_buffer_decapitate(wsgi_req->chunked_input_buf, i+1)) return -1; + // strtoul will stop at \r + return strtoul(wsgi_req->chunked_input_buf->buf, NULL, 16); + } + return -1; + } + if ((wsgi_req->chunked_input_buf->buf[i] >= '0' && wsgi_req->chunked_input_buf->buf[i] <= '9') || + (wsgi_req->chunked_input_buf->buf[i] >= 'a' && wsgi_req->chunked_input_buf->buf[i] <= 'z') || + (wsgi_req->chunked_input_buf->buf[i] >= 'A' && wsgi_req->chunked_input_buf->buf[i] <= 'Z')) continue; + if (wsgi_req->chunked_input_buf->buf[i] == '\r') { found = 1; continue; } + return -1; + } + + return -2; +} + +/* + + 0 -> waiting for \r\n + 1 -> waiting for whole body + +*/ + +char *uwsgi_chunked_read(struct wsgi_request *wsgi_req, size_t *len, int timeout, int nb) { + char *ret; + ssize_t chunk_len = 0; + if (!wsgi_req->chunked_input_buf) { + wsgi_req->chunked_input_buf = uwsgi_buffer_new(uwsgi.page_size); + wsgi_req->chunked_input_buf->limit = uwsgi.chunked_input_limit; + wsgi_req->chunked_input_want = 1; + } + + // the whole chunk stream has been consumed + if (wsgi_req->chunked_input_complete) { + *len = 0; + return wsgi_req->chunked_input_buf->buf; + } + + for(;;) { + if (wsgi_req->chunked_input_want || wsgi_req->chunked_input_buf->pos == 0) { + if (uwsgi_buffer_fix(wsgi_req->chunked_input_buf, uwsgi.page_size)) return NULL; + ssize_t rlen = uwsgi_chunked_input_recv(wsgi_req, timeout, nb); + if (rlen <= 0) return NULL; + // update buffer position + wsgi_req->chunked_input_buf->pos += rlen; + wsgi_req->chunked_input_want = 0; + + if (wsgi_req->chunked_input_need > 0) { + if ((size_t)rlen > wsgi_req->chunked_input_need) { + wsgi_req->chunked_input_need = 0; + } + else { + wsgi_req->chunked_input_need -= rlen; + } + if (wsgi_req->chunked_input_need > 0) wsgi_req->chunked_input_want = 1; + } + } + + if (wsgi_req->chunked_input_want) continue; + + // ok we have a frame, let's parse it + if (wsgi_req->chunked_input_buf->pos > 0) { + switch(wsgi_req->chunked_input_parser_status) { + case 0: + chunk_len = uwsgi_chunked_readline(wsgi_req); + if (chunk_len == -2) { + wsgi_req->chunked_input_want = 1; + break; + } + else if (chunk_len < 0) { + return NULL; + } + else if (chunk_len == 0) { + *len = 0; + wsgi_req->chunked_input_complete = 1; + return wsgi_req->chunked_input_buf->buf; + } + // if here the buffer has been already decapitated + if ((size_t)(chunk_len+2) > wsgi_req->chunked_input_buf->pos) { + wsgi_req->chunked_input_need = (chunk_len+2) - wsgi_req->chunked_input_buf->pos; + wsgi_req->chunked_input_parser_status = 1; + wsgi_req->chunked_input_want = 1; + break; + } + *len = chunk_len; + ret = wsgi_req->chunked_input_buf->buf; + if (uwsgi_buffer_decapitate(wsgi_req->chunked_input_buf, chunk_len+2)) return NULL; + return ret; + case 1: + if ((size_t)(chunk_len+2) > wsgi_req->chunked_input_buf->pos) { + wsgi_req->chunked_input_need = (chunk_len+2) - wsgi_req->chunked_input_buf->pos; + wsgi_req->chunked_input_want = 1; + break; + } + *len = chunk_len; + ret = wsgi_req->chunked_input_buf->buf; + if (uwsgi_buffer_decapitate(wsgi_req->chunked_input_buf, chunk_len+2)) return NULL; + wsgi_req->chunked_input_parser_status = 0; + return ret; + + } + } + } +} diff --git a/core/utils.c b/core/utils.c index 02a9ffe7..402ab31b 100644 --- a/core/utils.c +++ b/core/utils.c @@ -708,6 +708,11 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { free(ptr); } + // free chunked input + if (wsgi_req->chunked_input_buf) { + uwsgi_buffer_destroy(wsgi_req->chunked_input_buf); + } + // free websocket engine if (wsgi_req->websocket_buf) { uwsgi_buffer_destroy(wsgi_req->websocket_buf); @@ -1952,7 +1957,7 @@ int uwsgi_list_has_str(char *list, char *str) { return 0; } -char hex2num(char *str) { +static char hex2num(char *str) { char val = 0; diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index d3e9d422..76231f1b 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -1063,6 +1063,42 @@ PyObject *py_uwsgi_websocket_send(PyObject * self, PyObject * args) { return Py_None; } +PyObject *py_uwsgi_chunked_read(PyObject * self, PyObject * args) { + int timeout = 0; + if (!PyArg_ParseTuple(args, "|i:chunked_read", &timeout)) { + return NULL; + } + size_t len = 0; + struct wsgi_request *wsgi_req = py_current_wsgi_req(); + UWSGI_RELEASE_GIL + char *chunk = uwsgi_chunked_read(wsgi_req, &len, timeout, 0); + UWSGI_GET_GIL + if (!chunk) { + return PyErr_Format(PyExc_IOError, "unable to receive chunked part"); + } + + return PyString_FromStringAndSize(chunk, len); +} + +PyObject *py_uwsgi_chunked_read_nb(PyObject * self, PyObject * args) { + size_t len = 0; + struct wsgi_request *wsgi_req = py_current_wsgi_req(); + UWSGI_RELEASE_GIL + char *chunk = uwsgi_chunked_read(wsgi_req, &len, 0, 1); + UWSGI_GET_GIL + if (!chunk) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) { + Py_INCREF(Py_None); + return Py_None; + } + return PyErr_Format(PyExc_IOError, "unable to receive chunked part"); + } + + return PyString_FromStringAndSize(chunk, len); +} + + + PyObject *py_uwsgi_websocket_recv(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = py_current_wsgi_req(); UWSGI_RELEASE_GIL @@ -2484,6 +2520,9 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"websocket_send", py_uwsgi_websocket_send, METH_VARARGS, ""}, {"websocket_handshake", py_uwsgi_websocket_handshake, METH_VARARGS, ""}, + {"chunked_read", py_uwsgi_chunked_read, METH_VARARGS, ""}, + {"chunked_read_nb", py_uwsgi_chunked_read_nb, METH_VARARGS, ""}, + {NULL, NULL}, }; diff --git a/uwsgi.h b/uwsgi.h index 7ce4ec64..f1db35ed 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1428,6 +1428,12 @@ struct wsgi_request { time_t websocket_last_pong; int websocket_closed; + struct uwsgi_buffer *chunked_input_buf; + uint8_t chunked_input_parser_status; + uint8_t chunked_input_want; + size_t chunked_input_need; + uint8_t chunked_input_complete; + uint64_t stream_id; // avoid routing loops @@ -2343,6 +2349,9 @@ struct uwsgi_server { int websockets_pong_tolerance; uint64_t websockets_max_size; + int chunked_input_timeout; + uint64_t chunked_input_limit; + int (*wait_write_hook) (int, int); int (*wait_read_hook) (int, int); @@ -3866,6 +3875,8 @@ int uwsgi_websocket_send(struct wsgi_request *, char *, size_t); struct uwsgi_buffer *uwsgi_websocket_recv(struct wsgi_request *); struct uwsgi_buffer *uwsgi_websocket_recv_nb(struct wsgi_request *); +char *uwsgi_chunked_read(struct wsgi_request *, size_t *, int, int); + uint16_t uwsgi_be16(char *); uint32_t uwsgi_be32(char *); uint64_t uwsgi_be64(char *); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index f5708faf..4f9066dd 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -496,7 +496,7 @@ class uConf(object): 'core/notify', 'core/mule', 'core/subscription', 'core/stats', 'core/sendfile', 'core/async', 'core/master_checks', 'core/offload', 'core/io', 'core/static', 'core/websockets', 'core/spooler', 'core/snmp', 'core/exceptions', 'core/config', 'core/setup_utils', 'core/clock', 'core/init', 'core/buffer', 'core/reader', 'core/writer', 'core/alarm', 'core/cron', - 'core/plugins', 'core/lock', 'core/cache', 'core/daemons', 'core/errors', 'core/hash', 'core/master_events', + 'core/plugins', 'core/lock', 'core/cache', 'core/daemons', 'core/errors', 'core/hash', 'core/master_events', 'core/chunked', 'core/queue', 'core/event', 'core/signal', 'core/strings', 'core/progress', 'core/timebomb', 'core/ini', 'core/rpc', 'core/gateway', 'core/loop', 'core/cookie', 'core/querystring', 'core/rb_timers', 'core/transformations', 'core/uwsgi'] # add protocols