From 567545213d95e608b54ce92bfc33fac4327a9f93 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 20 Jul 2021 10:37:50 +0300 Subject: Python: fixing ASGI receive() issues. The receive() call never blocks for a GET request and always returns the same empty body message. The Starlette framework creates a separate task when receive() is called in a loop until an 'http.disconnect' message is received. The 'http.disconnect' message was previously issued after the response header had been sent. However, the correct behavior is to respond with 'http.disconnect' after sending the response is complete. This closes #564 issue on GitHub. --- src/python/nxt_python_asgi_http.c | 87 ++++++++++++++++++++++++--------------- 1 file changed, 54 insertions(+), 33 deletions(-) (limited to 'src/python') diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c index d88c4b00..3074c09f 100644 --- a/src/python/nxt_python_asgi_http.c +++ b/src/python/nxt_python_asgi_http.c @@ -23,10 +23,11 @@ typedef struct { PyObject *send_future; uint64_t content_length; uint64_t bytes_sent; - int complete; - int closed; PyObject *send_body; Py_ssize_t send_body_off; + uint8_t complete; + uint8_t closed; + uint8_t empty_body_received; } nxt_py_asgi_http_t; @@ -37,6 +38,7 @@ static PyObject *nxt_py_asgi_http_response_start(nxt_py_asgi_http_t *http, PyObject *dict); static PyObject *nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict); +static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http); static PyObject *nxt_py_asgi_http_done(PyObject *self, PyObject *future); @@ -94,10 +96,11 @@ nxt_py_asgi_http_create(nxt_unit_request_info_t *req) http->send_future = NULL; http->content_length = -1; http->bytes_sent = 0; - http->complete = 0; - http->closed = 0; http->send_body = NULL; http->send_body_off = 0; + http->complete = 0; + http->closed = 0; + http->empty_body_received = 0; } return (PyObject *) http; @@ -117,7 +120,7 @@ nxt_py_asgi_http_receive(PyObject *self, PyObject *none) nxt_unit_req_debug(req, "asgi_http_receive"); - if (nxt_slow_path(http->closed || nxt_unit_response_is_sent(req))) { + if (nxt_slow_path(http->closed || http->complete )) { msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str); } else { @@ -171,6 +174,14 @@ nxt_py_asgi_http_read_msg(nxt_py_asgi_http_t *http) size = nxt_py_asgi_http_body_buf_size; } + if (size == 0) { + if (http->empty_body_received) { + Py_RETURN_NONE; + } + + http->empty_body_received = 1; + } + if (size > 0) { body = PyBytes_FromStringAndSize(NULL, size); if (nxt_slow_path(body == NULL)) { @@ -442,6 +453,8 @@ nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict) if (more_body == NULL || more_body == Py_False) { http->complete = 1; + + nxt_py_asgi_http_emit_disconnect(http); } Py_INCREF(http); @@ -449,6 +462,41 @@ nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict) } +static void +nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http) +{ + PyObject *msg, *future, *res; + + if (http->receive_future == NULL) { + return; + } + + msg = nxt_py_asgi_new_msg(http->req, nxt_py_http_disconnect_str); + if (nxt_slow_path(msg == NULL)) { + return; + } + + if (msg == Py_None) { + Py_DECREF(msg); + return; + } + + future = http->receive_future; + http->receive_future = NULL; + + res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL); + if (nxt_slow_path(res == NULL)) { + nxt_unit_req_alert(http->req, "'set_result' call failed"); + nxt_python_print_exception(); + } + + Py_XDECREF(res); + Py_DECREF(future); + + Py_DECREF(msg); +} + + void nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req) { @@ -573,7 +621,6 @@ fail: void nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req) { - PyObject *msg, *future, *res; nxt_py_asgi_http_t *http; http = req->data; @@ -582,33 +629,7 @@ nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req) http->closed = 1; - if (http->receive_future == NULL) { - return; - } - - msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str); - if (nxt_slow_path(msg == NULL)) { - return; - } - - if (msg == Py_None) { - Py_DECREF(msg); - return; - } - - future = http->receive_future; - http->receive_future = NULL; - - res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL); - if (nxt_slow_path(res == NULL)) { - nxt_unit_req_alert(req, "'set_result' call failed"); - nxt_python_print_exception(); - } - - Py_XDECREF(res); - Py_DECREF(future); - - Py_DECREF(msg); + nxt_py_asgi_http_emit_disconnect(http); } -- cgit From dfbdc1c11a201e46d61f4bc61cfbe5741fc4fd70 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 20 Jul 2021 10:37:53 +0300 Subject: Python: fixing exceptions in Future.set_result for ASGI implementation. An ASGI application can cancel the Future object returned by the receive() call. In this case, Unit's ASGI implementation should not call set_result() because the Future is already handled. In particular, the Starlette framework was noted to cancel the received Future. This patch adds a done() check for the Future before attempting a set_result(). This is related to #564 issue on GitHub. --- src/python/nxt_python_asgi_http.c | 55 +++++++++++++++++++++++---------------- 1 file changed, 32 insertions(+), 23 deletions(-) (limited to 'src/python') diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c index 3074c09f..c4a77d53 100644 --- a/src/python/nxt_python_asgi_http.c +++ b/src/python/nxt_python_asgi_http.c @@ -39,6 +39,8 @@ static PyObject *nxt_py_asgi_http_response_start(nxt_py_asgi_http_t *http, static PyObject *nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict); static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http); +static void nxt_py_asgi_http_set_result(nxt_py_asgi_http_t *http, + PyObject *future, PyObject *msg); static PyObject *nxt_py_asgi_http_done(PyObject *self, PyObject *future); @@ -465,7 +467,7 @@ nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http, PyObject *dict) static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http) { - PyObject *msg, *future, *res; + PyObject *msg, *future; if (http->receive_future == NULL) { return; @@ -484,23 +486,45 @@ nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http) future = http->receive_future; http->receive_future = NULL; - res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL); + nxt_py_asgi_http_set_result(http, future, msg); + + Py_DECREF(msg); +} + + +static void +nxt_py_asgi_http_set_result(nxt_py_asgi_http_t *http, PyObject *future, + PyObject *msg) +{ + PyObject *res; + + res = PyObject_CallMethodObjArgs(future, nxt_py_done_str, NULL); if (nxt_slow_path(res == NULL)) { - nxt_unit_req_alert(http->req, "'set_result' call failed"); + nxt_unit_req_alert(http->req, "'done' call failed"); nxt_python_print_exception(); } + if (nxt_fast_path(res == Py_False)) { + res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, + NULL); + if (nxt_slow_path(res == NULL)) { + nxt_unit_req_alert(http->req, "'set_result' call failed"); + nxt_python_print_exception(); + } + + } else { + res = NULL; + } + Py_XDECREF(res); Py_DECREF(future); - - Py_DECREF(msg); } void nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req) { - PyObject *msg, *future, *res; + PyObject *msg, *future; nxt_py_asgi_http_t *http; http = req->data; @@ -524,14 +548,7 @@ nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req) future = http->receive_future; http->receive_future = NULL; - res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL); - if (nxt_slow_path(res == NULL)) { - nxt_unit_req_alert(req, "'set_result' call failed"); - nxt_python_print_exception(); - } - - Py_XDECREF(res); - Py_DECREF(future); + nxt_py_asgi_http_set_result(http, future, msg); Py_DECREF(msg); } @@ -575,15 +592,7 @@ nxt_py_asgi_http_drain(nxt_queue_link_t *lnk) future = http->send_future; http->send_future = NULL; - res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, Py_None, - NULL); - if (nxt_slow_path(res == NULL)) { - nxt_unit_req_alert(http->req, "'set_result' call failed"); - nxt_python_print_exception(); - } - - Py_XDECREF(res); - Py_DECREF(future); + nxt_py_asgi_http_set_result(http, future, Py_None); return NXT_UNIT_OK; -- cgit From f27fbd9b4d2bdaddf1e7001d0d0bc5586ba04cd4 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 20 Jul 2021 10:37:54 +0300 Subject: Python: using default event_loop for main thread for ASGI. Unit's ASGI implementation creates a new event loop to run an application for each thread since 542b5b8c0647. This may cause unexpected exceptions or strange bugs if asyncio synchronisation primitives are initialised before the application starts (e.g. globally). Although the approach with a new event loop for the main thread is consistent and helps to prepare the application to run in multiple threads, it can be a source of pain for people who just want to run single-threaded ASGI applications in Unit. This is related to #560 issue on GitHub. --- src/python/nxt_python.c | 4 ++-- src/python/nxt_python.h | 2 +- src/python/nxt_python_asgi.c | 28 +++++++++++++++++----------- src/python/nxt_python_wsgi.c | 4 ++-- 4 files changed, 22 insertions(+), 16 deletions(-) (limited to 'src/python') diff --git a/src/python/nxt_python.c b/src/python/nxt_python.c index 588a147a..bdce68b2 100644 --- a/src/python/nxt_python.c +++ b/src/python/nxt_python.c @@ -264,7 +264,7 @@ nxt_python_start(nxt_task_t *task, nxt_process_data_t *data) goto fail; } - rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data); + rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data, 1); if (nxt_slow_path(rc != NXT_UNIT_OK)) { goto fail; } @@ -504,7 +504,7 @@ nxt_python_init_threads(nxt_python_app_conf_t *c) for (i = 0; i < c->threads - 1; i++) { ti = &nxt_py_threads[i]; - res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data); + res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data, 0); if (nxt_slow_path(res != NXT_UNIT_OK)) { return NXT_UNIT_ERROR; } diff --git a/src/python/nxt_python.h b/src/python/nxt_python.h index a5c1d9a6..e4eac9dc 100644 --- a/src/python/nxt_python.h +++ b/src/python/nxt_python.h @@ -60,7 +60,7 @@ typedef struct { typedef struct { - int (*ctx_data_alloc)(void **pdata); + int (*ctx_data_alloc)(void **pdata, int main); void (*ctx_data_free)(void *data); int (*startup)(void *data); int (*run)(nxt_unit_ctx_t *ctx); diff --git a/src/python/nxt_python_asgi.c b/src/python/nxt_python_asgi.c index 1d220678..26003805 100644 --- a/src/python/nxt_python_asgi.c +++ b/src/python/nxt_python_asgi.c @@ -17,7 +17,7 @@ static PyObject *nxt_python_asgi_get_func(PyObject *obj); -static int nxt_python_asgi_ctx_data_alloc(void **pdata); +static int nxt_python_asgi_ctx_data_alloc(void **pdata, int main); static void nxt_python_asgi_ctx_data_free(void *data); static int nxt_python_asgi_startup(void *data); static int nxt_python_asgi_run(nxt_unit_ctx_t *ctx); @@ -194,10 +194,11 @@ nxt_python_asgi_init(nxt_unit_init_t *init, nxt_python_proto_t *proto) static int -nxt_python_asgi_ctx_data_alloc(void **pdata) +nxt_python_asgi_ctx_data_alloc(void **pdata, int main) { uint32_t i; - PyObject *asyncio, *loop, *new_event_loop, *obj; + PyObject *asyncio, *loop, *event_loop, *obj; + const char *event_loop_func; nxt_py_asgi_ctx_data_t *ctx_data; ctx_data = nxt_unit_malloc(NULL, sizeof(nxt_py_asgi_ctx_data_t)); @@ -232,23 +233,28 @@ nxt_python_asgi_ctx_data_alloc(void **pdata) goto fail; } - new_event_loop = PyDict_GetItemString(PyModule_GetDict(asyncio), - "new_event_loop"); - if (nxt_slow_path(new_event_loop == NULL)) { + event_loop_func = main ? "get_event_loop" : "new_event_loop"; + + event_loop = PyDict_GetItemString(PyModule_GetDict(asyncio), + event_loop_func); + if (nxt_slow_path(event_loop == NULL)) { nxt_unit_alert(NULL, - "Python failed to get 'new_event_loop' from module 'asyncio'"); + "Python failed to get '%s' from module 'asyncio'", + event_loop_func); goto fail; } - if (nxt_slow_path(PyCallable_Check(new_event_loop) == 0)) { + if (nxt_slow_path(PyCallable_Check(event_loop) == 0)) { nxt_unit_alert(NULL, - "'asyncio.new_event_loop' is not a callable object"); + "'asyncio.%s' is not a callable object", + event_loop_func); goto fail; } - loop = PyObject_CallObject(new_event_loop, NULL); + loop = PyObject_CallObject(event_loop, NULL); if (nxt_slow_path(loop == NULL)) { - nxt_unit_alert(NULL, "Python failed to call 'asyncio.new_event_loop'"); + nxt_unit_alert(NULL, "Python failed to call 'asyncio.%s'", + event_loop_func); goto fail; } diff --git a/src/python/nxt_python_wsgi.c b/src/python/nxt_python_wsgi.c index b80d10fa..87dcfaa2 100644 --- a/src/python/nxt_python_wsgi.c +++ b/src/python/nxt_python_wsgi.c @@ -51,7 +51,7 @@ typedef struct { } nxt_python_ctx_t; -static int nxt_python_wsgi_ctx_data_alloc(void **pdata); +static int nxt_python_wsgi_ctx_data_alloc(void **pdata, int main); static void nxt_python_wsgi_ctx_data_free(void *data); static int nxt_python_wsgi_run(nxt_unit_ctx_t *ctx); static void nxt_python_wsgi_done(void); @@ -210,7 +210,7 @@ fail: static int -nxt_python_wsgi_ctx_data_alloc(void **pdata) +nxt_python_wsgi_ctx_data_alloc(void **pdata, int main) { nxt_python_ctx_t *pctx; -- cgit From 3580842d34f8543f7bb41551f7a0dec8723289a8 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Mon, 9 Aug 2021 10:15:00 +0300 Subject: Python: fixing misprint in error message. --- src/python/nxt_python.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src/python') diff --git a/src/python/nxt_python.c b/src/python/nxt_python.c index bdce68b2..abb04194 100644 --- a/src/python/nxt_python.c +++ b/src/python/nxt_python.c @@ -364,13 +364,13 @@ nxt_python_set_target(nxt_task_t *task, nxt_python_target_t *target, obj = PyDict_GetItemString(PyModule_GetDict(module), callable); if (nxt_slow_path(obj == NULL)) { nxt_alert(task, "Python failed to get \"%s\" from module \"%s\"", - callable, module); + callable, module_name); goto fail; } if (nxt_slow_path(PyCallable_Check(obj) == 0)) { nxt_alert(task, "\"%s\" in module \"%s\" is not a callable object", - callable, module); + callable, module_name); goto fail; } -- cgit