Libunit: closing active requests on quit.

src/python/nxt_python_asgi.c

@@ -25,6 +25,7 @@ static int nxt_python_asgi_run(nxt_unit_ctx_t *ctx);
 static void nxt_py_asgi_remove_reader(nxt_unit_ctx_t *ctx,
     nxt_unit_port_t *port);
 static void nxt_py_asgi_request_handler(nxt_unit_request_info_t *req);
+static void nxt_py_asgi_close_handler(nxt_unit_request_info_t *req);
 
 static PyObject *nxt_py_asgi_create_http_scope(nxt_unit_request_info_t *req);
 static PyObject *nxt_py_asgi_create_address(nxt_unit_sptr_t *sptr, uint8_t len,
@@ -179,7 +180,7 @@ nxt_python_asgi_init(nxt_unit_init_t *init, nxt_python_proto_t *proto)
     init->callbacks.request_handler = nxt_py_asgi_request_handler;
     init->callbacks.data_handler = nxt_py_asgi_http_data_handler;
     init->callbacks.websocket_handler = nxt_py_asgi_websocket_handler;
-    init->callbacks.close_handler = nxt_py_asgi_websocket_close_handler;
+    init->callbacks.close_handler = nxt_py_asgi_close_handler;
     init->callbacks.quit = nxt_py_asgi_quit;
     init->callbacks.shm_ack_handler = nxt_py_asgi_shm_ack_handler;
     init->callbacks.add_port = nxt_py_asgi_add_port;
@@ -551,6 +552,18 @@ release_asgi:
 }
 
 
+static void
+nxt_py_asgi_close_handler(nxt_unit_request_info_t *req)
+{
+    if (req->request->websocket_handshake) {
+        nxt_py_asgi_websocket_close_handler(req);
+
+    } else {
+        nxt_py_asgi_http_close_handler(req);
+    }
+}
+
+
 static PyObject *
 nxt_py_asgi_create_http_scope(nxt_unit_request_info_t *req)
 {

src/python/nxt_python_asgi.h

@@ -58,6 +58,7 @@ int nxt_py_asgi_http_init(void);
 PyObject *nxt_py_asgi_http_create(nxt_unit_request_info_t *req);
 void nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req);
 int nxt_py_asgi_http_drain(nxt_queue_link_t *lnk);
+void nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req);
 
 int nxt_py_asgi_websocket_init(void);
 PyObject *nxt_py_asgi_websocket_create(nxt_unit_request_info_t *req);

src/python/nxt_python_asgi_http.c

@@ -24,6 +24,7 @@ typedef struct {
     uint64_t                 content_length;
     uint64_t                 bytes_sent;
     int                      complete;
+    int                      closed;
     PyObject                 *send_body;
     Py_ssize_t               send_body_off;
 } nxt_py_asgi_http_t;
@@ -94,6 +95,7 @@ nxt_py_asgi_http_create(nxt_unit_request_info_t *req)
         http->content_length = -1;
         http->bytes_sent = 0;
         http->complete = 0;
+        http->closed = 0;
         http->send_body = NULL;
         http->send_body_off = 0;
     }
@@ -115,7 +117,13 @@ nxt_py_asgi_http_receive(PyObject *self, PyObject *none)
 
     nxt_unit_req_debug(req, "asgi_http_receive");
 
-    msg = nxt_py_asgi_http_read_msg(http);
+    if (nxt_slow_path(http->closed || nxt_unit_response_is_sent(req))) {
+        msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str);
+
+    } else {
+        msg = nxt_py_asgi_http_read_msg(http);
+    }
+
     if (nxt_slow_path(msg == NULL)) {
         return NULL;
     }
@@ -561,6 +569,48 @@ 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;
+
+    nxt_unit_req_debug(req, "asgi_http_close_handler");
+
+    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);
+}
+
+
 static PyObject *
 nxt_py_asgi_http_done(PyObject *self, PyObject *future)
 {