Merged with the default branch.1.25.0-1

5 files changed, 105 insertions, 69 deletions

diff --git a/src/python/nxt_python.c b/src/python/nxt_python.c
index 588a147a..abb04194 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;
     }
@@ -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;
     }
 
@@ -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_asgi_http.c b/src/python/nxt_python_asgi_http.c
index d88c4b00..c4a77d53 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,9 @@ 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 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);
 
 
@@ -94,10 +98,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 +122,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 +176,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 +455,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,10 +464,67 @@ 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;
+
+    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;
+
+    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, "'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);
+}
+
+
 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;
@@ -476,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);
 }
@@ -527,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;
 
@@ -573,7 +630,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 +638,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);
 }
 
 
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;