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, 19 insertions, 0 deletions

diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c
index b7068c88..24a40406 100644
--- a/src/nxt_port_memory.c
+++ b/src/nxt_port_memory.c
@@ -112,6 +112,8 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
     u_char                   *p;
     nxt_mp_t                 *mp;
     nxt_buf_t                *b;
+    nxt_port_t               *port;
+    nxt_process_t            *process;
     nxt_chunk_id_t           c;
     nxt_port_mmap_header_t   *hdr;
     nxt_port_mmap_handler_t  *mmap_handler;
@@ -163,6 +165,21 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
         c++;
     }
 
+    if (hdr->dst_pid == nxt_pid
+        && nxt_atomic_cmp_set(&hdr->oosm, 1, 0))
+    {
+        process = nxt_runtime_process_find(task->thread->runtime, hdr->src_pid);
+
+        if (process != NULL && !nxt_queue_is_empty(&process->ports)) {
+            port = nxt_process_port_first(process);
+
+            if (port->type == NXT_PROCESS_WORKER) {
+                (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_SHM_ACK,
+                                             -1, 0, 0, NULL);
+            }
+        }
+    }
+
 release_buf:
 
     nxt_port_mmap_handler_use(mmap_handler, -1);
@@ -454,6 +471,8 @@ nxt_port_mmap_get(nxt_task_t *task, nxt_port_t *port, nxt_chunk_id_t *c,
                 goto unlock_return;
             }
         }
+
+        hdr->oosm = 1;
     }
 
     /* TODO introduce port_mmap limit and release wait. */