Introducing port messages to notify about out of shared memory.

- OOSM (out of shared memory).  Sent by application process to router
  when application reaches the limit of allocated shared memory and
  needs more.
- SHM_ACK.  Sent by router to application when the application's shared
  memory is released and the OOSM flag is enabled for the segment.

This implements blocking mode (the library waits for SHM_ACK in case of
out of shared memory condition and retries allocating the required memory
amount) and non-blocking mode (the library notifies the application that
it's out of shared memory and returns control to the application module
that sets up the output queue and puts SHM_ACK in the main message loop).

1 files changed, 56 insertions, 0 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index 38396e86..6a1f3792 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -248,6 +248,7 @@ static nxt_int_t nxt_router_http_request_done(nxt_task_t *task,
     nxt_http_request_t *r);
 static void nxt_router_http_request_release(nxt_task_t *task, void *obj,
     void *data);
+static void nxt_router_oosm_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 
 extern const nxt_http_request_state_t  nxt_http_websocket;
 
@@ -276,6 +277,7 @@ nxt_port_handlers_t  nxt_router_process_port_handlers = {
     .access_log   = nxt_router_access_log_reopen_handler,
     .rpc_ready    = nxt_port_rpc_handler,
     .rpc_error    = nxt_port_rpc_handler,
+    .oosm         = nxt_router_oosm_handler,
 };
 
 
@@ -2748,6 +2750,7 @@ static nxt_port_handlers_t  nxt_router_app_port_handlers = {
     .rpc_error = nxt_port_rpc_handler,
     .mmap      = nxt_port_mmap_handler,
     .data      = nxt_port_rpc_handler,
+    .oosm      = nxt_router_oosm_handler,
 };
 
 
@@ -5241,3 +5244,56 @@ nxt_router_http_request_release(nxt_task_t *task, void *obj, void *data)
 
     nxt_mp_release(r->mem_pool);
 }
+
+
+static void
+nxt_router_oosm_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
+{
+    size_t                   mi;
+    uint32_t                 i;
+    nxt_bool_t               ack;
+    nxt_process_t            *process;
+    nxt_free_map_t           *m;
+    nxt_port_mmap_header_t   *hdr;
+
+    nxt_debug(task, "oosm in %PI", msg->port_msg.pid);
+
+    process = nxt_runtime_process_find(task->thread->runtime,
+                                       msg->port_msg.pid);
+    if (nxt_slow_path(process == NULL)) {
+        return;
+    }
+
+    ack = 0;
+
+    /*
+     * To mitigate possible racing condition (when OOSM message received
+     * after some of the memory was already freed), need to try to find
+     * first free segment in shared memory and send ACK if found.
+     */
+
+    nxt_thread_mutex_lock(&process->incoming.mutex);
+
+    for (i = 0; i < process->incoming.size; i++) {
+        hdr = process->incoming.elts[i].mmap_handler->hdr;
+        m = hdr->free_map;
+
+        for (mi = 0; mi < MAX_FREE_IDX; mi++) {
+            if (m[mi] != 0) {
+                ack = 1;
+
+                nxt_debug(task, "oosm: already free #%uD %uz = 0x%08xA",
+                          i, mi, m[mi]);
+
+                break;
+            }
+        }
+    }
+
+    nxt_thread_mutex_unlock(&process->incoming.mutex);
+
+    if (ack) {
+        (void) nxt_port_socket_write(task, msg->port, NXT_PORT_MSG_SHM_ACK,
+                                     -1, 0, 0, NULL);
+    }
+}