Introducing extended app process management.

- Pre-fork 'processes.spare' application processes;
- fork more processes to keep 'processes.spare' idle processes;
- fork on-demand up to 'processes.max' count;
- scale down idle application processes above 'processes.spare' after
  'processes.idle_timeout';
- number of concurrently started application processes also limited by
  'processes.spare' (or 1, if spare is 0).

1 files changed, 545 insertions, 85 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index 1ba7072d..19a37c69 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -12,11 +12,15 @@
 
 typedef struct {
     nxt_str_t         type;
-    uint32_t          workers;
+    uint32_t          processes;
+    uint32_t          max_processes;
+    uint32_t          spare_processes;
     nxt_msec_t        timeout;
     nxt_msec_t        res_timeout;
+    nxt_msec_t        idle_timeout;
     uint32_t          requests;
     nxt_conf_value_t  *limits_value;
+    nxt_conf_value_t  *processes_value;
 } nxt_router_app_conf_t;
 
 
@@ -76,6 +80,12 @@ typedef struct {
 } nxt_socket_rpc_t;
 
 
+typedef struct {
+    nxt_app_t               *app;
+    nxt_router_temp_conf_t  *temp_conf;
+} nxt_app_rpc_t;
+
+
 struct nxt_port_select_state_s {
     nxt_app_t           *app;
     nxt_req_app_link_t  *ra;
@@ -83,7 +93,7 @@ struct nxt_port_select_state_s {
     nxt_port_t          *failed_port;
     int                 failed_port_use_delta;
 
-    nxt_bool_t          can_start_worker;
+    uint8_t             start_process;    /* 1 bit */
     nxt_req_app_link_t  *shared_ra;
     nxt_port_t          *port;
 };
@@ -96,7 +106,7 @@ static void nxt_router_port_select(nxt_task_t *task,
 static nxt_int_t nxt_router_port_post_select(nxt_task_t *task,
     nxt_port_select_state_t *state);
 
-static nxt_int_t nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app);
+static nxt_int_t nxt_router_start_app_process(nxt_task_t *task, nxt_app_t *app);
 
 nxt_inline void
 nxt_router_ra_inc_use(nxt_req_app_link_t *ra)
@@ -136,6 +146,12 @@ static void nxt_router_listen_socket_ready(nxt_task_t *task,
     nxt_port_recv_msg_t *msg, void *data);
 static void nxt_router_listen_socket_error(nxt_task_t *task,
     nxt_port_recv_msg_t *msg, void *data);
+static void nxt_router_app_rpc_create(nxt_task_t *task,
+    nxt_router_temp_conf_t *tmcf, nxt_app_t *app);
+static void nxt_router_app_prefork_ready(nxt_task_t *task,
+    nxt_port_recv_msg_t *msg, void *data);
+static void nxt_router_app_prefork_error(nxt_task_t *task,
+    nxt_port_recv_msg_t *msg, void *data);
 static nxt_socket_conf_t *nxt_router_socket_conf(nxt_task_t *task,
     nxt_router_temp_conf_t *tmcf, nxt_str_t *name);
 static nxt_int_t nxt_router_listen_socket_find(nxt_router_temp_conf_t *tmcf,
@@ -193,7 +209,7 @@ static void nxt_router_app_port_ready(nxt_task_t *task,
 static void nxt_router_app_port_error(nxt_task_t *task,
     nxt_port_recv_msg_t *msg, void *data);
 
-static nxt_port_t * nxt_router_app_get_idle_port(nxt_app_t *app);
+static void nxt_router_app_quit(nxt_task_t *task, nxt_app_t *app);
 static void nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
     uint32_t request_failed, uint32_t got_response);
 static nxt_int_t nxt_router_app_port(nxt_task_t *task, nxt_app_t *app,
@@ -209,6 +225,12 @@ static nxt_int_t nxt_go_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
     nxt_app_wmsg_t *wmsg);
 static void nxt_router_conn_free(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data);
+static void nxt_router_adjust_idle_timer(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_app_idle_timeout(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_app_release_handler(nxt_task_t *task, void *obj,
+    void *data);
 
 static const nxt_http_request_state_t  nxt_http_request_send_state;
 static void nxt_http_request_send_body(nxt_task_t *task, void *obj, void *data);
@@ -253,7 +275,8 @@ nxt_router_start(nxt_task_t *task, void *data)
 
 
 static void
-nxt_router_start_worker_handler(nxt_task_t *task, nxt_port_t *port, void *data)
+nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
+    void *data)
 {
     size_t         size;
     uint32_t       stream;
@@ -268,7 +291,7 @@ nxt_router_start_worker_handler(nxt_task_t *task, nxt_port_t *port, void *data)
     rt = task->thread->runtime;
     main_port = rt->port_by_type[NXT_PROCESS_MAIN];
 
-    nxt_debug(task, "app '%V' %p start worker", &app->name, app);
+    nxt_debug(task, "app '%V' %p start process", &app->name, app);
 
     size = app->name.length + 1 + app->conf.length;
 
@@ -304,7 +327,7 @@ failed:
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    app->pending_workers--;
+    app->pending_processes--;
 
     nxt_thread_mutex_unlock(&app->mutex);
 
@@ -313,7 +336,7 @@ failed:
 
 
 static nxt_int_t
-nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app)
+nxt_router_start_app_process(nxt_task_t *task, nxt_app_t *app)
 {
     nxt_int_t      res;
     nxt_port_t     *router_port;
@@ -324,7 +347,7 @@ nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app)
 
     nxt_router_app_use(task, app, 1);
 
-    res = nxt_port_post(task, router_port, nxt_router_start_worker_handler,
+    res = nxt_port_post(task, router_port, nxt_router_start_app_process_handler,
                         app);
 
     if (res == NXT_OK) {
@@ -333,7 +356,7 @@ nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app)
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    app->pending_workers--;
+    app->pending_processes--;
 
     nxt_thread_mutex_unlock(&app->mutex);
 
@@ -737,7 +760,8 @@ nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 
 
 static void
-nxt_router_worker_remove_pid(nxt_task_t *task, nxt_port_t *port, void *data)
+nxt_router_app_process_remove_pid(nxt_task_t *task, nxt_port_t *port,
+    void *data)
 {
     union {
         nxt_pid_t  removed_pid;
@@ -763,7 +787,7 @@ nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 
     nxt_queue_each(engine, &nxt_router->engines, nxt_event_engine_t, link0)
     {
-        nxt_port_post(task, engine->port, nxt_router_worker_remove_pid,
+        nxt_port_post(task, engine->port, nxt_router_app_process_remove_pid,
                       msg->u.data);
     }
     nxt_queue_loop;
@@ -837,10 +861,27 @@ fail:
 }
 
 
+nxt_inline nxt_bool_t
+nxt_router_app_can_start(nxt_app_t *app)
+{
+    return app->processes + app->pending_processes < app->max_processes
+            && app->pending_processes < app->max_pending_processes;
+}
+
+
+nxt_inline nxt_bool_t
+nxt_router_app_need_start(nxt_app_t *app)
+{
+    return app->idle_processes + app->pending_processes
+            < app->spare_processes;
+}
+
+
 static void
 nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data)
 {
     nxt_int_t                    ret;
+    nxt_app_t                    *app;
     nxt_router_t                 *router;
     nxt_runtime_t                *rt;
     nxt_queue_link_t             *qlk;
@@ -863,6 +904,15 @@ nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data)
         return;
     }
 
+    nxt_queue_each(app, &tmcf->apps, nxt_app_t, link) {
+
+        if (nxt_router_app_need_start(app)) {
+            nxt_router_app_rpc_create(task, tmcf, app);
+            return;
+        }
+
+    } nxt_queue_loop;
+
     rt = task->thread->runtime;
 
     interface = nxt_service_get(rt->services, "engine", NULL);
@@ -923,6 +973,7 @@ nxt_router_conf_ready(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
 static void
 nxt_router_conf_error(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
 {
+    nxt_app_t          *app;
     nxt_socket_t       s;
     nxt_router_t       *router;
     nxt_queue_link_t   *qlk;
@@ -944,6 +995,12 @@ nxt_router_conf_error(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
         nxt_free(skcf->listen);
     }
 
+    nxt_queue_each(app, &tmcf->apps, nxt_app_t, link) {
+
+        nxt_router_app_quit(task, app);
+
+    } nxt_queue_loop;
+
     router = tmcf->conf->router;
 
     nxt_queue_add(&router->sockets, &tmcf->keeping);
@@ -984,15 +1041,21 @@ static nxt_conf_map_t  nxt_router_app_conf[] = {
     },
 
     {
-        nxt_string("workers"),
+        nxt_string("limits"),
+        NXT_CONF_MAP_PTR,
+        offsetof(nxt_router_app_conf_t, limits_value),
+    },
+
+    {
+        nxt_string("processes"),
         NXT_CONF_MAP_INT32,
-        offsetof(nxt_router_app_conf_t, workers),
+        offsetof(nxt_router_app_conf_t, processes),
     },
 
     {
-        nxt_string("limits"),
+        nxt_string("processes"),
         NXT_CONF_MAP_PTR,
-        offsetof(nxt_router_app_conf_t, limits_value),
+        offsetof(nxt_router_app_conf_t, processes_value),
     },
 };
 
@@ -1018,6 +1081,27 @@ static nxt_conf_map_t  nxt_router_app_limits_conf[] = {
 };
 
 
+static nxt_conf_map_t  nxt_router_app_processes_conf[] = {
+    {
+        nxt_string("spare"),
+        NXT_CONF_MAP_INT32,
+        offsetof(nxt_router_app_conf_t, spare_processes),
+    },
+
+    {
+        nxt_string("max"),
+        NXT_CONF_MAP_INT32,
+        offsetof(nxt_router_app_conf_t, max_processes),
+    },
+
+    {
+        nxt_string("idle_timeout"),
+        NXT_CONF_MAP_MSEC,
+        offsetof(nxt_router_app_conf_t, idle_timeout),
+    },
+};
+
+
 static nxt_conf_map_t  nxt_router_listener_conf[] = {
     {
         nxt_string("application"),
@@ -1100,6 +1184,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
     nxt_conf_value_t            *applications, *application;
     nxt_conf_value_t            *listeners, *listener;
     nxt_socket_conf_t           *skcf;
+    nxt_event_engine_t          *engine;
     nxt_app_lang_module_t       *lang;
     nxt_router_app_conf_t       apcf;
     nxt_router_listener_conf_t  lscf;
@@ -1174,11 +1259,15 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             continue;
         }
 
-        apcf.workers = 1;
+        apcf.processes = 1;
+        apcf.max_processes = 1;
+        apcf.spare_processes = 1;
         apcf.timeout = 0;
         apcf.res_timeout = 1000;
+        apcf.idle_timeout = 15000;
         apcf.requests = 0;
         apcf.limits_value = NULL;
+        apcf.processes_value = NULL;
 
         ret = nxt_conf_map_object(mp, application, nxt_router_app_conf,
                                   nxt_nitems(nxt_router_app_conf), &apcf);
@@ -1204,10 +1293,27 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             }
         }
 
+        if (apcf.processes_value != NULL
+            && nxt_conf_type(apcf.processes_value) == NXT_CONF_OBJECT)
+        {
+            ret = nxt_conf_map_object(mp, apcf.processes_value,
+                                      nxt_router_app_processes_conf,
+                                      nxt_nitems(nxt_router_app_processes_conf),
+                                      &apcf);
+            if (ret != NXT_OK) {
+                nxt_log(task, NXT_LOG_CRIT, "application processes map error");
+                goto app_fail;
+            }
+
+        } else {
+            apcf.max_processes = apcf.processes;
+            apcf.spare_processes = apcf.processes;
+        }
+
         nxt_debug(task, "application type: %V", &apcf.type);
-        nxt_debug(task, "application workers: %D", apcf.workers);
-        nxt_debug(task, "application request timeout: %D", apcf.timeout);
-        nxt_debug(task, "application reschedule timeout: %D", apcf.res_timeout);
+        nxt_debug(task, "application processes: %D", apcf.processes);
+        nxt_debug(task, "application request timeout: %M", apcf.timeout);
+        nxt_debug(task, "application reschedule timeout: %M", apcf.res_timeout);
         nxt_debug(task, "application requests: %D", apcf.requests);
 
         lang = nxt_app_lang_module(task->thread->runtime, &apcf.type);
@@ -1226,6 +1332,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
         }
 
         nxt_queue_init(&app->ports);
+        nxt_queue_init(&app->spare_ports);
+        nxt_queue_init(&app->idle_ports);
         nxt_queue_init(&app->requests);
         nxt_queue_init(&app->pending);
 
@@ -1233,14 +1341,32 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
         nxt_memcpy(app->name.start, name.start, name.length);
 
         app->type = lang->type;
-        app->max_workers = apcf.workers;
+        app->max_processes = apcf.max_processes;
+        app->spare_processes = apcf.spare_processes;
+        app->max_pending_processes = apcf.spare_processes
+                                      ? apcf.spare_processes : 1;
         app->timeout = apcf.timeout;
         app->res_timeout = apcf.res_timeout * 1000000;
+        app->idle_timeout = apcf.idle_timeout;
         app->live = 1;
         app->max_pending_responses = 2;
         app->max_requests = apcf.requests;
         app->prepare_msg = nxt_app_prepare_msg[lang->type];
 
+        engine = task->thread->engine;
+
+        app->engine = engine;
+
+        app->idle_timer.precision = NXT_TIMER_DEFAULT_PRECISION;
+        app->idle_timer.work_queue = &engine->fast_work_queue;
+        app->idle_timer.handler = nxt_router_app_idle_timeout;
+        app->idle_timer.task = &engine->task;
+        app->idle_timer.log = app->idle_timer.task->log;
+
+        app->adjust_idle_work.handler = nxt_router_adjust_idle_timer;
+        app->adjust_idle_work.task = &engine->task;
+        app->adjust_idle_work.obj = app;
+
         nxt_queue_insert_tail(&tmcf->apps, &app->link);
 
         nxt_router_app_use(task, app, 1);
@@ -1624,6 +1750,119 @@ nxt_router_listen_socket_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 }
 
 
+static void
+nxt_router_app_rpc_create(nxt_task_t *task,
+    nxt_router_temp_conf_t *tmcf, nxt_app_t *app)
+{
+    size_t         size;
+    uint32_t       stream;
+    nxt_buf_t      *b;
+    nxt_port_t     *main_port, *router_port;
+    nxt_runtime_t  *rt;
+    nxt_app_rpc_t  *rpc;
+
+    rpc = nxt_mp_alloc(tmcf->mem_pool, sizeof(nxt_app_rpc_t));
+    if (rpc == NULL) {
+        goto fail;
+    }
+
+    rpc->app = app;
+    rpc->temp_conf = tmcf;
+
+    nxt_debug(task, "app '%V' prefork", &app->name);
+
+    size = app->name.length + 1 + app->conf.length;
+
+    b = nxt_buf_mem_alloc(tmcf->mem_pool, size, 0);
+    if (nxt_slow_path(b == NULL)) {
+        goto fail;
+    }
+
+    nxt_buf_cpystr(b, &app->name);
+    *b->mem.free++ = '\0';
+    nxt_buf_cpystr(b, &app->conf);
+
+    rt = task->thread->runtime;
+    main_port = rt->port_by_type[NXT_PROCESS_MAIN];
+    router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
+
+    stream = nxt_port_rpc_register_handler(task, router_port,
+                                           nxt_router_app_prefork_ready,
+                                           nxt_router_app_prefork_error,
+                                           -1, rpc);
+    if (nxt_slow_path(stream == 0)) {
+        goto fail;
+    }
+
+    app->pending_processes++;
+
+    nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_WORKER, -1,
+                          stream, router_port->id, b);
+
+    return;
+
+fail:
+
+    nxt_router_conf_error(task, tmcf);
+}
+
+
+static void
+nxt_router_app_prefork_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+    void *data)
+{
+    nxt_app_t           *app;
+    nxt_port_t          *port;
+    nxt_app_rpc_t       *rpc;
+    nxt_event_engine_t  *engine;
+
+    rpc = data;
+    app = rpc->app;
+
+    port = msg->u.new_port;
+    port->app = app;
+
+    nxt_router_app_use(task, app, 1);
+
+    app->pending_processes--;
+    app->processes++;
+    app->idle_processes++;
+
+    engine = task->thread->engine;
+
+    nxt_queue_insert_tail(&app->ports, &port->app_link);
+    nxt_queue_insert_tail(&app->spare_ports, &port->idle_link);
+
+    port->idle_start = 0;
+
+    nxt_port_inc_use(port);
+
+    nxt_work_queue_add(&engine->fast_work_queue,
+                       nxt_router_conf_apply, task, rpc->temp_conf, NULL);
+}
+
+
+static void
+nxt_router_app_prefork_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+    void *data)
+{
+    nxt_app_t               *app;
+    nxt_app_rpc_t           *rpc;
+    nxt_router_temp_conf_t  *tmcf;
+
+    rpc = data;
+    app = rpc->app;
+    tmcf = rpc->temp_conf;
+
+    nxt_log(task, NXT_LOG_WARN, "failed to start application \"%V\"",
+            &app->name);
+
+    app->pending_processes--;
+
+    nxt_router_conf_error(task, tmcf);
+}
+
+
 static nxt_int_t
 nxt_router_engines_create(nxt_task_t *task, nxt_router_t *router,
     nxt_router_temp_conf_t *tmcf, const nxt_event_interface_t *interface)
@@ -1933,32 +2172,11 @@ static void
 nxt_router_apps_sort(nxt_task_t *task, nxt_router_t *router,
     nxt_router_temp_conf_t *tmcf)
 {
-    nxt_app_t   *app;
-    nxt_port_t  *port;
+    nxt_app_t  *app;
 
     nxt_queue_each(app, &router->apps, nxt_app_t, link) {
 
-        nxt_queue_remove(&app->link);
-
-        nxt_debug(task, "about to free app '%V' %p", &app->name, app);
-
-        app->live = 0;
-
-        do {
-            port = nxt_router_app_get_idle_port(app);
-            if (port == NULL) {
-                break;
-            }
-
-            nxt_debug(task, "port %p send quit", port);
-
-            nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0,
-                                  NULL);
-
-            nxt_port_use(task, port, -1);
-        } while (1);
-
-        nxt_router_app_use(task, app, -1);
+        nxt_router_app_quit(task, app);
 
     } nxt_queue_loop;
 
@@ -2454,6 +2672,7 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         if (nxt_buf_mem_used_size(&b->mem) == 0) {
             nxt_work_queue_add(&task->thread->engine->fast_work_queue,
                                b->completion_handler, task, b, b->parent);
+
         } else {
             nxt_buf_chain_add(&r->out, b);
         }
@@ -2561,14 +2780,15 @@ nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    nxt_assert(app->pending_workers != 0);
+    nxt_assert(app->pending_processes != 0);
 
-    app->pending_workers--;
-    app->workers++;
+    app->pending_processes--;
+    app->processes++;
 
     nxt_thread_mutex_unlock(&app->mutex);
 
-    nxt_debug(task, "app '%V' %p new port ready", &app->name, app);
+    nxt_debug(task, "app '%V' new port ready, pid %PI, %d/%d",
+              &app->name, port->pid, app->processes, app->pending_processes);
 
     nxt_router_app_port_release(task, port, 0, 0);
 }
@@ -2590,9 +2810,9 @@ nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    nxt_assert(app->pending_workers != 0);
+    nxt_assert(app->pending_processes != 0);
 
-    app->pending_workers--;
+    app->pending_processes--;
 
     if (!nxt_queue_is_empty(&app->requests)) {
         lnk = nxt_queue_last(&app->requests);
@@ -2611,7 +2831,7 @@ nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         nxt_debug(task, "app '%V' %p abort next stream #%uD",
                   &app->name, app, ra->stream);
 
-        nxt_router_ra_error(ra, 500, "Failed to start application worker");
+        nxt_router_ra_error(ra, 500, "Failed to start application process");
         nxt_router_ra_use(task, ra, -1);
     }
 
@@ -2629,10 +2849,13 @@ nxt_router_app_use(nxt_task_t *task, nxt_app_t *app, int i)
     if (i < 0 && c == -i) {
 
         nxt_assert(app->live == 0);
-        nxt_assert(app->workers == 0);
-        nxt_assert(app->pending_workers == 0);
+        nxt_assert(app->processes == 0);
+        nxt_assert(app->idle_processes == 0);
+        nxt_assert(app->pending_processes == 0);
         nxt_assert(nxt_queue_is_empty(&app->requests) != 0);
         nxt_assert(nxt_queue_is_empty(&app->ports) != 0);
+        nxt_assert(nxt_queue_is_empty(&app->spare_ports) != 0);
+        nxt_assert(nxt_queue_is_empty(&app->idle_ports) != 0);
 
         nxt_thread_mutex_destroy(&app->mutex);
         nxt_free(app);
@@ -2666,6 +2889,19 @@ nxt_router_pop_first_port(nxt_app_t *app)
 
     port->app_pending_responses++;
 
+    if (nxt_queue_chk_remove(&port->idle_link)) {
+        app->idle_processes--;
+
+        if (port->idle_start == 0) {
+            nxt_assert(app->idle_processes < app->spare_processes);
+
+        } else {
+            nxt_assert(app->idle_processes >= app->spare_processes);
+
+            port->idle_start = 0;
+        }
+    }
+
     if ((app->max_pending_responses == 0
             || port->app_pending_responses < app->max_pending_responses)
         && (app->max_requests == 0
@@ -2684,8 +2920,8 @@ nxt_router_pop_first_port(nxt_app_t *app)
 }
 
 
-static nxt_port_t *
-nxt_router_app_get_idle_port(nxt_app_t *app)
+nxt_inline nxt_port_t *
+nxt_router_app_get_port_for_quit(nxt_app_t *app)
 {
     nxt_port_t  *port;
 
@@ -2701,8 +2937,16 @@ nxt_router_app_get_idle_port(nxt_app_t *app)
             continue;
         }
 
-        nxt_queue_remove(&port->app_link);
-        port->app_link.next = NULL;
+        /* Caller is responsible to decrease port use count. */
+        nxt_queue_chk_remove(&port->app_link);
+
+        if (nxt_queue_chk_remove(&port->idle_link)) {
+            app->idle_processes--;
+        }
+
+        /* Caller is responsible to decrease app use count. */
+        port->app = NULL;
+        app->processes--;
 
         break;
 
@@ -2715,6 +2959,41 @@ nxt_router_app_get_idle_port(nxt_app_t *app)
 
 
 static void
+nxt_router_app_quit(nxt_task_t *task, nxt_app_t *app)
+{
+    nxt_port_t  *port;
+
+    nxt_queue_remove(&app->link);
+
+    app->live = 0;
+
+    for ( ;; ) {
+        port = nxt_router_app_get_port_for_quit(app);
+        if (port == NULL) {
+            break;
+        }
+
+        nxt_debug(task, "send QUIT to app '%V' pid %PI", &app->name, port->pid);
+
+        nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
+
+        nxt_port_use(task, port, -1);
+        nxt_router_app_use(task, app, -1);
+    }
+
+    if (nxt_timer_is_in_tree(&app->idle_timer)) {
+        nxt_assert(app->engine == task->thread->engine);
+
+        app->idle_timer.handler = nxt_router_app_release_handler;
+        nxt_timer_add(app->engine, &app->idle_timer, 0);
+
+    } else {
+        nxt_router_app_use(task, app, -1);
+    }
+}
+
+
+static void
 nxt_router_app_process_request(nxt_task_t *task, void *obj, void *data)
 {
     nxt_app_t           *app;
@@ -2739,7 +3018,8 @@ nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
     uint32_t request_failed, uint32_t got_response)
 {
     nxt_app_t                *app;
-    nxt_bool_t               send_quit, cancelled;
+    nxt_bool_t               port_unchained;
+    nxt_bool_t               send_quit, cancelled, adjust_idle_timer;
     nxt_queue_link_t         *lnk;
     nxt_req_app_link_t       *ra, *pending_ra, *re_ra;
     nxt_port_select_state_t  state;
@@ -2856,12 +3136,52 @@ app_dead:
 
 re_ra_cancelled:
 
-    send_quit = (app->live == 0 && port->app_pending_responses > 0)
+    send_quit = (app->live == 0 && port->app_pending_responses == 0)
                 || (app->max_requests > 0 && port->app_pending_responses == 0
                     && port->app_responses >= app->max_requests);
 
+    if (send_quit) {
+        port_unchained = nxt_queue_chk_remove(&port->app_link);
+
+        port->app = NULL;
+        app->processes--;
+
+    } else {
+        port_unchained = 0;
+    }
+
+    adjust_idle_timer = 0;
+
+    if (!send_quit && port->app_pending_responses == 0) {
+        nxt_assert(port->idle_link.next == NULL);
+
+        if (app->idle_processes == app->spare_processes
+            && app->adjust_idle_work.data == NULL)
+        {
+            adjust_idle_timer = 1;
+            app->adjust_idle_work.data = app;
+            app->adjust_idle_work.next = NULL;
+        }
+
+        if (app->idle_processes < app->spare_processes) {
+            nxt_queue_insert_tail(&app->spare_ports, &port->idle_link);
+
+        } else {
+            nxt_queue_insert_tail(&app->idle_ports, &port->idle_link);
+
+            port->idle_start = task->thread->engine->timers.now;
+        }
+
+        app->idle_processes++;
+    }
+
     nxt_thread_mutex_unlock(&app->mutex);
 
+    if (adjust_idle_timer) {
+        nxt_router_app_use(task, app, 1);
+        nxt_event_engine_post(app->engine, &app->adjust_idle_work);
+    }
+
     if (pending_ra != NULL) {
         nxt_router_ra_use(task, pending_ra, -1);
     }
@@ -2893,12 +3213,18 @@ re_ra_cancelled:
     }
 
     if (send_quit) {
-        nxt_debug(task, "app '%V' %p is not alive, send QUIT to port",
+        nxt_debug(task, "app '%V' %p send QUIT to port",
                   &app->name, app);
 
         nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
                               -1, 0, 0, NULL);
 
+        if (port_unchained) {
+            nxt_port_use(task, port, -1);
+        }
+
+        nxt_router_app_use(task, app, -1);
+
         goto adjust_use;
     }
 
@@ -2916,8 +3242,10 @@ adjust_use:
 void
 nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port)
 {
-    nxt_app_t   *app;
-    nxt_bool_t  unchain, start_worker;
+    nxt_app_t         *app;
+    nxt_bool_t        unchain, start_process;
+    nxt_port_t        *idle_port;
+    nxt_queue_link_t  *idle_lnk;
 
     app = port->app;
 
@@ -2925,41 +3253,162 @@ nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port)
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    unchain = port->app_link.next != NULL;
+    unchain = nxt_queue_chk_remove(&port->app_link);
 
-    if (unchain) {
-        nxt_queue_remove(&port->app_link);
-        port->app_link.next = NULL;
+    if (nxt_queue_chk_remove(&port->idle_link)) {
+        app->idle_processes--;
+
+        if (port->idle_start == 0
+            && app->idle_processes >= app->spare_processes)
+        {
+            nxt_assert(!nxt_queue_is_empty(&app->idle_ports));
+
+            idle_lnk = nxt_queue_last(&app->idle_ports);
+            idle_port = nxt_queue_link_data(idle_lnk, nxt_port_t, idle_link);
+            nxt_queue_remove(idle_lnk);
+
+            nxt_queue_insert_tail(&app->spare_ports, idle_lnk);
+
+            idle_port->idle_start = 0;
+        }
     }
 
-    app->workers--;
+    app->processes--;
 
-    start_worker = app->live != 0
-                   && nxt_queue_is_empty(&app->requests) == 0
-                   && app->workers + app->pending_workers < app->max_workers;
+    start_process = app->live != 0
+                    && !task->thread->engine->shutdown
+                    && nxt_router_app_can_start(app)
+                    && (!nxt_queue_is_empty(&app->requests)
+                        || nxt_router_app_need_start(app));
 
-    if (start_worker) {
-        app->pending_workers++;
+    if (start_process) {
+        app->pending_processes++;
     }
 
     nxt_thread_mutex_unlock(&app->mutex);
 
-    nxt_debug(task, "app '%V' %p port %p close", &app->name, app, port);
+    nxt_debug(task, "app '%V' pid %PI closed", &app->name, port->pid);
 
     if (unchain) {
         nxt_port_use(task, port, -1);
     }
 
-    if (start_worker) {
-        nxt_router_start_worker(task, app);
+    if (start_process) {
+        nxt_router_start_app_process(task, app);
     }
 }
 
 
 static void
+nxt_router_adjust_idle_timer(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_app_t           *app;
+    nxt_bool_t          queued;
+    nxt_port_t          *port;
+    nxt_msec_t          timeout, threshold;
+    nxt_queue_link_t    *lnk;
+    nxt_event_engine_t  *engine;
+
+    app = obj;
+    queued = (data == app);
+
+    nxt_debug(task, "nxt_router_adjust_idle_timer: app \"%V\", queued %b",
+              &app->name, queued);
+
+    engine = task->thread->engine;
+
+    nxt_assert(app->engine == engine);
+
+    threshold = engine->timers.now + app->idle_timer.precision;
+    timeout = 0;
+
+    nxt_thread_mutex_lock(&app->mutex);
+
+    if (queued) {
+        app->adjust_idle_work.data = NULL;
+    }
+
+    while (app->idle_processes > app->spare_processes) {
+
+        nxt_assert(nxt_queue_is_empty(&app->idle_ports) == 0);
+
+        lnk = nxt_queue_first(&app->idle_ports);
+        port = nxt_queue_link_data(lnk, nxt_port_t, idle_link);
+
+        timeout = port->idle_start + app->idle_timeout;
+
+        if (timeout > threshold) {
+            break;
+        }
+
+        nxt_queue_remove(lnk);
+        lnk->next = NULL;
+
+        nxt_queue_chk_remove(&port->app_link);
+
+        app->idle_processes--;
+        app->processes--;
+        port->app = NULL;
+
+        nxt_thread_mutex_unlock(&app->mutex);
+
+        nxt_debug(task, "app '%V' send QUIT to idle port %PI",
+                  &app->name, port->pid);
+
+        nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
+
+        nxt_port_use(task, port, -1);
+        nxt_router_app_use(task, app, -1);
+
+        nxt_thread_mutex_lock(&app->mutex);
+    }
+
+    nxt_thread_mutex_unlock(&app->mutex);
+
+    if (timeout > threshold) {
+        nxt_timer_add(engine, &app->idle_timer, timeout - threshold);
+
+    } else {
+        nxt_timer_disable(engine, &app->idle_timer);
+    }
+
+    if (queued) {
+        nxt_router_app_use(task, app, -1);
+    }
+}
+
+
+static void
+nxt_router_app_idle_timeout(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_app_t    *app;
+    nxt_timer_t  *timer;
+
+    timer = obj;
+    app = nxt_container_of(timer, nxt_app_t, idle_timer);
+
+    nxt_router_adjust_idle_timer(task, app, NULL);
+}
+
+
+static void
+nxt_router_app_release_handler(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_app_t    *app;
+    nxt_timer_t  *timer;
+
+    timer = obj;
+    app = nxt_container_of(timer, nxt_app_t, idle_timer);
+
+    nxt_router_app_use(task, app, -1);
+}
+
+
+static void
 nxt_router_port_select(nxt_task_t *task, nxt_port_select_state_t *state)
 {
     nxt_app_t           *app;
+    nxt_bool_t          can_start_process;
     nxt_req_app_link_t  *ra;
 
     ra = state->ra;
@@ -2996,12 +3445,13 @@ nxt_router_port_select(nxt_task_t *task, nxt_port_select_state_t *state)
         ra->app_port = NULL;
     }
 
-    state->can_start_worker = (app->workers + app->pending_workers)
-                              < app->max_workers;
+    can_start_process = nxt_router_app_can_start(app);
+
     state->port = NULL;
+    state->start_process = 0;
 
     if (nxt_queue_is_empty(&app->ports)
-        || (state->can_start_worker && nxt_router_app_first_port_busy(app)) )
+        || (can_start_process && nxt_router_app_first_port_busy(app)) )
     {
         ra = nxt_router_ra_create(task, ra);
 
@@ -3020,8 +3470,9 @@ nxt_router_port_select(nxt_task_t *task, nxt_port_select_state_t *state)
 
         nxt_debug(task, "ra stream #%uD enqueue to app->requests", ra->stream);
 
-        if (state->can_start_worker) {
-            app->pending_workers++;
+        if (can_start_process) {
+            app->pending_processes++;
+            state->start_process = 1;
         }
 
     } else {
@@ -3038,6 +3489,11 @@ nxt_router_port_select(nxt_task_t *task, nxt_port_select_state_t *state)
 
             nxt_router_ra_pending(task, app, ra);
         }
+
+        if (can_start_process && nxt_router_app_need_start(app)) {
+            app->pending_processes++;
+            state->start_process = 1;
+        }
     }
 
 fail:
@@ -3077,20 +3533,24 @@ nxt_router_port_post_select(nxt_task_t *task, nxt_port_select_state_t *state)
 
         ra->app_port = state->port;
 
+        if (state->start_process) {
+            nxt_router_start_app_process(task, app);
+        }
+
         return NXT_OK;
     }
 
-    if (!state->can_start_worker) {
-        nxt_debug(task, "app '%V' %p too many running or pending workers",
+    if (!state->start_process) {
+        nxt_debug(task, "app '%V' %p too many running or pending processes",
                   &app->name, app);
 
         return NXT_AGAIN;
     }
 
-    res = nxt_router_start_worker(task, app);
+    res = nxt_router_start_app_process(task, app);
 
     if (nxt_slow_path(res != NXT_OK)) {
-        nxt_router_ra_error(ra, 500, "Failed to start worker");
+        nxt_router_ra_error(ra, 500, "Failed to start app process");
         nxt_router_ra_use(task, ra, -1);
 
         return NXT_ERROR;
@@ -3227,7 +3687,7 @@ nxt_router_app_prepare_request(nxt_task_t *task, nxt_req_app_link_t *ra)
         goto release_port;
     }
 
-    nxt_debug(task, "about to send %O bytes buffer to worker port %d",
+    nxt_debug(task, "about to send %O bytes buffer to app process port %d",
                     nxt_buf_used_size(wmsg.write),
                     wmsg.port->socket.fd);