Fix/python asyncio

src/python/nxt_python.c

@@ -345,7 +345,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, 1);
+    rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data);
     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
         goto fail;
     }
@@ -623,7 +623,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, 0);
+        res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data);
         if (nxt_slow_path(res != NXT_UNIT_OK)) {
             return NXT_UNIT_ERROR;
         }

src/python/nxt_python.h

@@ -37,6 +37,10 @@
 #define NXT_HAVE_ASGI  1
 #endif
 
+#if PY_VERSION_HEX >= NXT_PYTHON_VER(3, 11)
+#define NXT_HAVE_ASYNCIO_RUNNER  1
+#endif
+
 
 typedef struct {
     PyObject    *application;
@@ -62,7 +66,7 @@ typedef struct {
 
 
 typedef struct {
-    int   (*ctx_data_alloc)(void **pdata, int main);
+    int   (*ctx_data_alloc)(void **pdata);
     void  (*ctx_data_free)(void *data);
     int   (*startup)(void *data);
     int   (*run)(nxt_unit_ctx_t *ctx);

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, int main);
+static int nxt_python_asgi_ctx_data_alloc(void **pdata);
 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);
@@ -199,10 +199,133 @@ nxt_python_asgi_init(nxt_unit_init_t *init, nxt_python_proto_t *proto)
 
     return NXT_UNIT_OK;
 }
+#if (NXT_HAVE_ASYNCIO_RUNNER)
+static int
+nxt_python_asgi_ctx_data_alloc(void **pdata)
+{
+    uint32_t                i;
+    PyObject                *asyncio, *loop, *obj, *runner_ref, *runner, *loop_ref;
+    const char              *event_loop_func;
+    const char              *runner_class;
+    nxt_py_asgi_ctx_data_t  *ctx_data;
+
+    ctx_data = nxt_unit_malloc(NULL, sizeof(nxt_py_asgi_ctx_data_t));
+    if (nxt_slow_path(ctx_data == NULL)) {
+        nxt_unit_alert(NULL, "Failed to allocate context data");
+        return NXT_UNIT_ERROR;
+    }
+
+    memset(ctx_data, 0, sizeof(nxt_py_asgi_ctx_data_t));
+
+    nxt_queue_init(&ctx_data->drain_queue);
+
+    struct {
+        const char  *key;
+        PyObject    **handler;
+
+    } handlers[] = {
+        { "create_task",        &ctx_data->loop_create_task },
+        { "add_reader",         &ctx_data->loop_add_reader },
+        { "remove_reader",      &ctx_data->loop_remove_reader },
+        { "call_soon",          &ctx_data->loop_call_soon },
+        { "run_until_complete", &ctx_data->loop_run_until_complete },
+        { "create_future",      &ctx_data->loop_create_future },
+    };
+
+    loop = NULL;
+
+    asyncio = PyImport_ImportModule("asyncio");
+    if (nxt_slow_path(asyncio == NULL)) {
+        nxt_unit_alert(NULL, "Python failed to import module 'asyncio'");
+        nxt_python_print_exception();
+        goto fail;
+    }
 
+    event_loop_func = "get_loop";
+    runner_class = "Runner";
+    runner_ref = PyDict_GetItemString(PyModule_GetDict(asyncio),
+                                      runner_class);
+    if (nxt_slow_path(runner_ref == NULL)) {
+        nxt_unit_alert(NULL,
+                       "Python failed to get '%s' from module 'asyncio'",
+                       runner_class);
+        goto fail;
+    }
+
+    runner = PyObject_CallObject(runner_ref, NULL);
+    if (nxt_slow_path(runner == NULL)) {
+        nxt_unit_alert(NULL, "Python failed to call 'asyncio.%s'",
+                       runner_class);
+        goto fail;
+    }
+    ctx_data->runner = runner;
+    loop_ref = PyObject_GetAttrString(runner, event_loop_func);
+    loop = PyObject_CallObject(loop_ref, NULL);
+        if (nxt_slow_path(runner == NULL)) {
+        nxt_unit_alert(NULL, "Python failed to call 'Runner.%s'",
+                       event_loop_func);
+        goto fail;
+    }
+    for (i = 0; i < nxt_nitems(handlers); i++) {
+        obj = PyObject_GetAttrString(loop, handlers[i].key);
+        if (nxt_slow_path(obj == NULL)) {
+            nxt_unit_alert(NULL, "Python failed to get 'loop.%s'",
+                                 handlers[i].key);
+            goto fail;
+        }
+
+        *handlers[i].handler = obj;
+
+        if (nxt_slow_path(PyCallable_Check(obj) == 0)) {
+            nxt_unit_alert(NULL, "'loop.%s' is not a callable object",
+                                 handlers[i].key);
+            goto fail;
+        }
+    }
+
+    obj = PyObject_CallObject(ctx_data->loop_create_future, NULL);
+    if (nxt_slow_path(obj == NULL)) {
+        nxt_unit_alert(NULL, "Python failed to create Future ");
+        nxt_python_print_exception();
+        goto fail;
+    }
+
+    ctx_data->quit_future = obj;
+
+    obj = PyObject_GetAttrString(ctx_data->quit_future, "set_result");
+    if (nxt_slow_path(obj == NULL)) {
+        nxt_unit_alert(NULL, "Python failed to get 'future.set_result'");
+        goto fail;
+    }
+
+    ctx_data->quit_future_set_result = obj;
+
+    if (nxt_slow_path(PyCallable_Check(obj) == 0)) {
+        nxt_unit_alert(NULL, "'future.set_result' is not a callable object");
+        goto fail;
+    }
+
+    Py_DECREF(loop);
+    Py_DECREF(asyncio);
+
+    *pdata = ctx_data;
+
+    return NXT_UNIT_OK;
+
+fail:
+
+    nxt_python_asgi_ctx_data_free(ctx_data);
+
+    Py_XDECREF(loop);
+    Py_XDECREF(asyncio);
+
+    return NXT_UNIT_ERROR;
+}
+
+#else
 
 static int
-nxt_python_asgi_ctx_data_alloc(void **pdata, int main)
+nxt_python_asgi_ctx_data_alloc(void **pdata)
 {
     uint32_t                i;
     PyObject                *asyncio, *loop, *event_loop, *obj;
@@ -241,7 +364,7 @@ nxt_python_asgi_ctx_data_alloc(void **pdata, int main)
         goto fail;
     }
 
-    event_loop_func = main ? "get_event_loop" : "new_event_loop";
+    event_loop_func =  "new_event_loop";
 
     event_loop = PyDict_GetItemString(PyModule_GetDict(asyncio),
                                       event_loop_func);
@@ -321,7 +444,7 @@ nxt_python_asgi_ctx_data_alloc(void **pdata, int main)
 
     return NXT_UNIT_ERROR;
 }
-
+#endif
 
 static void
 nxt_python_asgi_ctx_data_free(void *data)
@@ -338,7 +461,12 @@ nxt_python_asgi_ctx_data_free(void *data)
     Py_XDECREF(ctx_data->loop_remove_reader);
     Py_XDECREF(ctx_data->quit_future);
     Py_XDECREF(ctx_data->quit_future_set_result);
-
+    #if (NXT_HAVE_ASYNCIO_RUNNER)
+    PyObject *close_loop_func;
+    close_loop_func = PyObject_GetAttrString(ctx_data->runner, "close");
+    PyObject_CallObject(close_loop_func, NULL);
+    Py_XDECREF(ctx_data->runner);
+    #endif
     nxt_unit_free(NULL, ctx_data);
 }
 

src/python/nxt_python_asgi.h

@@ -5,7 +5,7 @@
 
 #ifndef _NXT_PYTHON_ASGI_H_INCLUDED_
 #define _NXT_PYTHON_ASGI_H_INCLUDED_
-
+#include <python/nxt_python.h>
 
 typedef PyObject * (*nxt_py_asgi_enum_header_cb)(void *ctx, int i,
     PyObject *name, PyObject *val);
@@ -34,8 +34,12 @@ typedef struct {
     PyObject         *quit_future;
     PyObject         *quit_future_set_result;
     PyObject         **target_lifespans;
+    #if (NXT_HAVE_ASYNCIO_RUNNER)
+    PyObject         *runner;
+    #endif
 } nxt_py_asgi_ctx_data_t;
 
+
 PyObject *nxt_py_asgi_enum_headers(PyObject *headers,
     nxt_py_asgi_enum_header_cb cb, void *data);
 

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, int main);
+static int nxt_python_wsgi_ctx_data_alloc(void **pdata);
 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);
@@ -217,7 +217,7 @@ nxt_python_wsgi_init(nxt_unit_init_t *init, nxt_python_proto_t *proto)
 
 
 static int
-nxt_python_wsgi_ctx_data_alloc(void **pdata, int main)
+nxt_python_wsgi_ctx_data_alloc(void **pdata)
 {
     nxt_python_ctx_t  *pctx;