External Go app request processing.

1 files changed, 150 insertions, 0 deletions

diff --git a/src/nginext/nxt_go_process.c b/src/nginext/nxt_go_process.c
new file mode 100644
index 00000000..c7ce052d
--- /dev/null
+++ b/src/nginext/nxt_go_process.c
@@ -0,0 +1,150 @@
+
+/*
+ * Copyright (C) Max Romanov
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef NXT_CONFIGURE
+
+
+#include "nxt_go_process.h"
+#include "nxt_go_array.h"
+#include "nxt_go_mutex.h"
+#include "nxt_go_log.h"
+
+#include <nxt_port_memory_int.h>
+
+
+static nxt_array_t processes; /* of nxt_go_process_t */
+
+static nxt_go_process_t *
+nxt_go_find_process(nxt_pid_t pid, uint32_t *pos)
+{
+    uint32_t          l, r, i;
+    nxt_go_process_t  *process;
+
+    if (nxt_slow_path(processes.size == 0)) {
+        nxt_go_array_init(&processes, 1, sizeof(nxt_go_process_t));
+    }
+
+    l = 0;
+    r = processes.nelts;
+    i = (l + r) / 2;
+
+    while (r > l) {
+        process = nxt_go_array_at(&processes, i);
+
+        nxt_go_debug("compare process #%d (%p) at %d",
+                     (int)process->pid, process, (int)i);
+
+        if (pid == process->pid) {
+            nxt_go_debug("found process %d at %d", (int)pid, (int)i);
+
+            if (pos != NULL) {
+                *pos = i;
+            }
+
+            return process;
+        }
+
+        if (pid < process->pid) {
+            r = i;
+        } else {
+            l = i + 1;
+        }
+
+        i = (l + r) / 2;
+    }
+
+    if (pos != NULL) {
+        *pos = i;
+    }
+
+    nxt_go_debug("process %d not found, best pos %d", (int)pid, (int)i);
+
+    return NULL;
+}
+
+
+nxt_go_process_t *
+nxt_go_get_process(nxt_pid_t pid)
+{
+    uint32_t          pos;
+    nxt_go_process_t  *process;
+
+    process = nxt_go_find_process(pid, &pos);
+
+    if (process == NULL) {
+        nxt_go_array_add(&processes);
+        process = nxt_go_array_at(&processes, pos);
+
+        nxt_go_debug("init process #%d (%p) at %d", (int)pid, process,
+                     (int)pos);
+
+        if (pos < processes.nelts - 1) {
+            memmove(process + 1, process,
+                    processes.size * (processes.nelts - 1 - pos));
+        }
+
+        process->pid = pid;
+        nxt_go_mutex_create(&process->incoming_mutex);
+        nxt_go_array_init(&process->incoming, 1, sizeof(nxt_port_mmap_t));
+        nxt_go_mutex_create(&process->outgoing_mutex);
+        nxt_go_array_init(&process->outgoing, 1, sizeof(nxt_port_mmap_t));
+    }
+
+    return process;
+}
+
+
+void
+nxt_go_new_incoming_mmap(nxt_pid_t pid, nxt_fd_t fd)
+{
+    void             *mem;
+    struct stat      mmap_stat;
+    nxt_port_mmap_t  *port_mmap;
+    nxt_go_process_t *process;
+
+    process = nxt_go_get_process(pid);
+
+    nxt_go_debug("got new mmap fd #%d from process %d",
+                 (int)fd, (int)pid);
+
+    if (fstat(fd, &mmap_stat) == -1) {
+        nxt_go_warn("fstat(%d) failed %d", (int)fd, errno);
+
+        return;
+    }
+
+    nxt_go_mutex_lock(&process->incoming_mutex);
+
+    port_mmap = nxt_go_array_zero_add(&process->incoming);
+    if (nxt_slow_path(port_mmap == NULL)) {
+        nxt_go_warn("failed to add mmap to incoming array");
+
+        goto fail;
+    }
+
+    mem = mmap(NULL, mmap_stat.st_size,
+               PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+
+    if (nxt_slow_path(mem == MAP_FAILED)) {
+        nxt_go_warn("mmap() failed %d", errno);
+
+        goto fail;
+    }
+
+    port_mmap->hdr = mem;
+
+    if (nxt_slow_path(port_mmap->hdr->id != process->incoming.nelts - 1)) {
+        nxt_go_warn("port mmap id mismatch (%d != %d)",
+                    port_mmap->hdr->id, process->incoming.nelts - 1);
+    }
+
+fail:
+
+    nxt_go_mutex_unlock(&process->incoming_mutex);
+}
+
+
+#endif /* NXT_CONFIGURE */