Moving request limit control to libunit.

Introducting application graceful stop.  For now only used when application
process reach request limit value.

This closes #585 issue on GitHub.

6 files changed, 16 insertions, 85 deletions

diff --git a/src/python/nxt_python.c b/src/python/nxt_python.c
index abb04194..8687c869 100644
--- a/src/python/nxt_python.c
+++ b/src/python/nxt_python.c
@@ -230,10 +230,9 @@ nxt_python_start(nxt_task_t *task, nxt_process_data_t *data)
         }
     }
 
-    nxt_unit_default_init(task, &python_init);
+    nxt_unit_default_init(task, &python_init, data->app);
 
     python_init.data = c;
-    python_init.shm_limit = data->app->shm_limit;
     python_init.callbacks.ready_handler = nxt_python_ready_handler;
 
     proto = c->protocol;
@@ -522,18 +521,6 @@ nxt_python_ready_handler(nxt_unit_ctx_t *ctx)
     nxt_py_thread_info_t   *ti;
     nxt_python_app_conf_t  *c;
 
-    if (nxt_py_proto.ready != NULL) {
-        res = nxt_py_proto.ready(ctx);
-        if (nxt_slow_path(res != NXT_UNIT_OK)) {
-            return NXT_UNIT_ERROR;
-        }
-    }
-
-    /* Worker thread context. */
-    if (!nxt_unit_is_main_ctx(ctx)) {
-        return NXT_UNIT_OK;
-    }
-
     c = ctx->unit->data;
 
     if (c->threads <= 1) {
diff --git a/src/python/nxt_python.h b/src/python/nxt_python.h
index e4eac9dc..eddb1cfc 100644
--- a/src/python/nxt_python.h
+++ b/src/python/nxt_python.h
@@ -64,7 +64,6 @@ typedef struct {
     void  (*ctx_data_free)(void *data);
     int   (*startup)(void *data);
     int   (*run)(nxt_unit_ctx_t *ctx);
-    int   (*ready)(nxt_unit_ctx_t *ctx);
     void  (*done)(void);
 } nxt_python_proto_t;
 
diff --git a/src/python/nxt_python_asgi.c b/src/python/nxt_python_asgi.c
index 8bf43d94..354e3a81 100644
--- a/src/python/nxt_python_asgi.c
+++ b/src/python/nxt_python_asgi.c
@@ -33,11 +33,10 @@ static PyObject *nxt_py_asgi_create_address(nxt_unit_sptr_t *sptr, uint8_t len,
 static PyObject *nxt_py_asgi_create_header(nxt_unit_field_t *f);
 static PyObject *nxt_py_asgi_create_subprotocols(nxt_unit_field_t *f);
 
-static int nxt_python_asgi_ready(nxt_unit_ctx_t *ctx);
-
 static int nxt_py_asgi_add_port(nxt_unit_ctx_t *ctx, nxt_unit_port_t *port);
 static int nxt_py_asgi_add_reader(nxt_unit_ctx_t *ctx, nxt_unit_port_t *port);
-static void nxt_py_asgi_remove_port(nxt_unit_t *lib, nxt_unit_port_t *port);
+static void nxt_py_asgi_remove_port(nxt_unit_t *lib, nxt_unit_ctx_t *ctx,
+    nxt_unit_port_t *port);
 static void nxt_py_asgi_quit(nxt_unit_ctx_t *ctx);
 static void nxt_py_asgi_shm_ack_handler(nxt_unit_ctx_t *ctx);
 
@@ -45,7 +44,6 @@ static PyObject *nxt_py_asgi_port_read(PyObject *self, PyObject *args);
 static void nxt_python_asgi_done(void);
 
 static PyObject           *nxt_py_port_read;
-static nxt_unit_port_t    *nxt_py_shared_port;
 
 static PyMethodDef        nxt_py_port_read_method =
     {"unit_port_read", nxt_py_asgi_port_read, METH_VARARGS, ""};
@@ -55,7 +53,6 @@ static nxt_python_proto_t  nxt_py_asgi_proto = {
     .ctx_data_free  = nxt_python_asgi_ctx_data_free,
     .startup        = nxt_python_asgi_startup,
     .run            = nxt_python_asgi_run,
-    .ready          = nxt_python_asgi_ready,
     .done           = nxt_python_asgi_done,
 };
 
@@ -362,13 +359,6 @@ nxt_python_asgi_run(nxt_unit_ctx_t *ctx)
 
     Py_DECREF(res);
 
-    nxt_py_asgi_remove_reader(ctx, nxt_py_shared_port);
-    nxt_py_asgi_remove_reader(ctx, ctx_data->port);
-
-    if (ctx_data->port != NULL) {
-        ctx_data->port = NULL;
-    }
-
     nxt_py_asgi_lifespan_shutdown(ctx);
 
     return NXT_UNIT_OK;
@@ -892,27 +882,9 @@ fail:
 
 
 static int
-nxt_python_asgi_ready(nxt_unit_ctx_t *ctx)
-{
-    nxt_unit_port_t  *port;
-
-    if (nxt_slow_path(nxt_py_shared_port == NULL)) {
-        return NXT_UNIT_ERROR;
-    }
-
-    port = nxt_py_shared_port;
-
-    nxt_unit_debug(ctx, "asgi_ready %d %p %p", port->in_fd, ctx, port);
-
-    return nxt_py_asgi_add_reader(ctx, port);
-}
-
-
-static int
 nxt_py_asgi_add_port(nxt_unit_ctx_t *ctx, nxt_unit_port_t *port)
 {
-    int                     nb;
-    nxt_py_asgi_ctx_data_t  *ctx_data;
+    int  nb;
 
     if (port->in_fd == -1) {
         return NXT_UNIT_OK;
@@ -929,16 +901,6 @@ nxt_py_asgi_add_port(nxt_unit_ctx_t *ctx, nxt_unit_port_t *port)
 
     nxt_unit_debug(ctx, "asgi_add_port %d %p %p", port->in_fd, ctx, port);
 
-    if (port->id.id == NXT_UNIT_SHARED_PORT_ID) {
-        nxt_py_shared_port = port;
-
-        return NXT_UNIT_OK;
-    }
-
-    ctx_data = ctx->data;
-
-    ctx_data->port = port;
-
     return nxt_py_asgi_add_reader(ctx, port);
 }
 
@@ -1008,17 +970,16 @@ clean_fd:
 
 
 static void
-nxt_py_asgi_remove_port(nxt_unit_t *lib, nxt_unit_port_t *port)
+nxt_py_asgi_remove_port(nxt_unit_t *lib, nxt_unit_ctx_t *ctx,
+    nxt_unit_port_t *port)
 {
-    if (port->in_fd == -1) {
+    if (port->in_fd == -1 || ctx == NULL) {
         return;
     }
 
     nxt_unit_debug(NULL, "asgi_remove_port %d %p", port->in_fd, port);
 
-    if (nxt_py_shared_port == port) {
-        nxt_py_shared_port = NULL;
-    }
+    nxt_py_asgi_remove_reader(ctx, port);
 }
 
 
@@ -1032,27 +993,6 @@ nxt_py_asgi_quit(nxt_unit_ctx_t *ctx)
 
     ctx_data = ctx->data;
 
-    if (nxt_py_shared_port != NULL) {
-        p = PyLong_FromLong(nxt_py_shared_port->in_fd);
-        if (nxt_slow_path(p == NULL)) {
-            nxt_unit_alert(NULL, "Python failed to create Long");
-            nxt_python_print_exception();
-
-        } else {
-            res = PyObject_CallFunctionObjArgs(ctx_data->loop_remove_reader,
-                                               p, NULL);
-            if (nxt_slow_path(res == NULL)) {
-                nxt_unit_alert(NULL, "Python failed to remove_reader");
-                nxt_python_print_exception();
-
-            } else {
-                Py_DECREF(res);
-            }
-
-            Py_DECREF(p);
-        }
-    }
-
     p = PyLong_FromLong(0);
     if (nxt_slow_path(p == NULL)) {
         nxt_unit_alert(NULL, "Python failed to create Long");
diff --git a/src/python/nxt_python_asgi.h b/src/python/nxt_python_asgi.h
index 20702065..94478a36 100644
--- a/src/python/nxt_python_asgi.h
+++ b/src/python/nxt_python_asgi.h
@@ -34,7 +34,6 @@ typedef struct {
     PyObject         *quit_future;
     PyObject         *quit_future_set_result;
     PyObject         **target_lifespans;
-    nxt_unit_port_t  *port;
 } nxt_py_asgi_ctx_data_t;
 
 PyObject *nxt_py_asgi_enum_headers(PyObject *headers,
diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c
index c4a77d53..05c0da4f 100644
--- a/src/python/nxt_python_asgi_http.c
+++ b/src/python/nxt_python_asgi_http.c
@@ -636,9 +636,11 @@ nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req)
 
     nxt_unit_req_debug(req, "asgi_http_close_handler");
 
-    http->closed = 1;
+    if (nxt_fast_path(http != NULL)) {
+        http->closed = 1;
 
-    nxt_py_asgi_http_emit_disconnect(http);
+        nxt_py_asgi_http_emit_disconnect(http);
+    }
 }
 
 
diff --git a/src/python/nxt_python_asgi_websocket.c b/src/python/nxt_python_asgi_websocket.c
index fc7d9fa4..ab1d0324 100644
--- a/src/python/nxt_python_asgi_websocket.c
+++ b/src/python/nxt_python_asgi_websocket.c
@@ -980,6 +980,10 @@ nxt_py_asgi_websocket_close_handler(nxt_unit_request_info_t *req)
 
     nxt_unit_req_debug(req, "asgi_websocket_close_handler");
 
+    if (nxt_slow_path(ws == NULL)) {
+        return;
+    }
+
     if (ws->receive_future == NULL) {
         ws->state = NXT_WS_DISCONNECTED;