From d65a66f9d813294917822554311281c5e1a7126b Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 29 Dec 2020 19:01:24 +0300 Subject: Libunit: processing single port message. This partially reverts the optimisation introduced in 1d84b9e4b459 to avoid an unpredictable block in nxt_unit_process_port_msg(). Under high load, this function may never return control to its caller, and the external event loop (in Node.js and Python asyncio) won't be able to process other scheduled events. To reproduce the issue, two request processing types are needed: 'fast' and 'furious'. The 'fast' one simply returns a small response, while the 'furious' schedules asynchronous calls to external resources. Thus, if Unit is subjected to a large amount of 'fast' requests, the 'furious' request processing freezes until the high load ends. The issue was found by Wu Jian Ping (@wujjpp) during Node.js stream implementation discussion and relates to PR #502 on GitHub. --- src/python/nxt_python_asgi.c | 41 ++++++++++++++++++++++++++++------------- 1 file changed, 28 insertions(+), 13 deletions(-) (limited to 'src/python/nxt_python_asgi.c') diff --git a/src/python/nxt_python_asgi.c b/src/python/nxt_python_asgi.c index 98aeedf4..a6f94507 100644 --- a/src/python/nxt_python_asgi.c +++ b/src/python/nxt_python_asgi.c @@ -1131,11 +1131,12 @@ nxt_py_asgi_shm_ack_handler(nxt_unit_ctx_t *ctx) static PyObject * nxt_py_asgi_port_read(PyObject *self, PyObject *args) { - int rc; - PyObject *arg; - Py_ssize_t n; - nxt_unit_ctx_t *ctx; - nxt_unit_port_t *port; + int rc; + PyObject *arg0, *arg1, *res; + Py_ssize_t n; + nxt_unit_ctx_t *ctx; + nxt_unit_port_t *port; + nxt_py_asgi_ctx_data_t *ctx_data; n = PyTuple_GET_SIZE(args); @@ -1147,31 +1148,45 @@ nxt_py_asgi_port_read(PyObject *self, PyObject *args) return PyErr_Format(PyExc_TypeError, "invalid number of arguments"); } - arg = PyTuple_GET_ITEM(args, 0); - if (nxt_slow_path(arg == NULL || PyLong_Check(arg) == 0)) { + arg0 = PyTuple_GET_ITEM(args, 0); + if (nxt_slow_path(arg0 == NULL || PyLong_Check(arg0) == 0)) { return PyErr_Format(PyExc_TypeError, "the first argument is not a long"); } - ctx = PyLong_AsVoidPtr(arg); + ctx = PyLong_AsVoidPtr(arg0); - arg = PyTuple_GET_ITEM(args, 1); - if (nxt_slow_path(arg == NULL || PyLong_Check(arg) == 0)) { + arg1 = PyTuple_GET_ITEM(args, 1); + if (nxt_slow_path(arg1 == NULL || PyLong_Check(arg1) == 0)) { return PyErr_Format(PyExc_TypeError, "the second argument is not a long"); } - port = PyLong_AsVoidPtr(arg); - - nxt_unit_debug(ctx, "asgi_port_read %p %p", ctx, port); + port = PyLong_AsVoidPtr(arg1); rc = nxt_unit_process_port_msg(ctx, port); + nxt_unit_debug(ctx, "asgi_port_read(%p,%p): %d", ctx, port, rc); + if (nxt_slow_path(rc == NXT_UNIT_ERROR)) { return PyErr_Format(PyExc_RuntimeError, "error processing port %d message", port->id.id); } + if (rc == NXT_UNIT_OK) { + ctx_data = ctx->data; + + res = PyObject_CallFunctionObjArgs(ctx_data->loop_call_soon, + nxt_py_port_read, + arg0, arg1, NULL); + if (nxt_slow_path(res == NULL)) { + nxt_unit_alert(ctx, "Python failed to call 'loop.call_soon'"); + nxt_python_print_exception(); + } + + Py_XDECREF(res); + } + Py_RETURN_NONE; } -- cgit