Go: worker initialization. READY message to master.

8 files changed, 140 insertions, 97 deletions

diff --git a/src/nginext/nginext.go b/src/nginext/nginext.go
index 2bde80a6..de0218e8 100644
--- a/src/nginext/nginext.go
+++ b/src/nginext/nginext.go
@@ -84,7 +84,6 @@ func ListenAndServe() {
 	var read_port *port
 
 	go_ports_env := os.Getenv("NXT_GO_PORTS")
-	fmt.Printf("NXT_GO_PORTS = %s\n", go_ports_env)
 
 	ports := strings.Split(go_ports_env, ";")
 	pid := os.Getpid()
@@ -95,7 +94,6 @@ func ListenAndServe() {
 		}
 
 		attrs := strings.Split(port_str, ",")
-		fmt.Printf("Port = %q\n", attrs)
 
 		var attrsN [5]int
 		var err error
@@ -119,6 +117,8 @@ func ListenAndServe() {
 	}
 
 	if read_port != nil {
+		C.nxt_go_ready()
+
 		for !nxt_go_quit {
 			read_port.read()
 		}
diff --git a/src/nginext/nxt_go_lib.c b/src/nginext/nxt_go_lib.c
index 86d5b619..57c3da8e 100644
--- a/src/nginext/nxt_go_lib.c
+++ b/src/nginext/nxt_go_lib.c
@@ -42,7 +42,7 @@ nxt_go_request_done(nxt_go_request_t r)
 }
 
 void
-nxt_go_listen_and_serve()
+nxt_go_ready()
 {
 }
 
@@ -54,16 +54,6 @@ nxt_go_process_port_msg(void *buf, size_t buf_len, void *oob, size_t oob_len)
 
 #else
 
-#if 0
-
-#include <nxt_runtime.h>
-#include <nxt_master_process.h>
-#include <nxt_application.h>
-
-#include "nxt_go_port.h"
-
-#endif
-
 #include "nxt_go_run_ctx.h"
 #include "nxt_go_log.h"
 #include "nxt_go_port.h"
@@ -184,6 +174,29 @@ nxt_go_request_done(nxt_go_request_t r)
 }
 
 
+void
+nxt_go_ready()
+{
+    char           *go_stream;
+    nxt_port_msg_t port_msg;
+
+    go_stream = getenv("NXT_GO_STREAM");
+
+    if (go_stream == NULL) {
+        return;
+    }
+
+    port_msg.stream = atol(go_stream);
+    port_msg.pid = getpid();
+    port_msg.reply_port = 0;
+    port_msg.type = NXT_PORT_MSG_READY;
+    port_msg.last = 0;
+    port_msg.mmap = 0;
+
+    nxt_go_master_send(&port_msg, sizeof(port_msg), NULL, 0);
+}
+
+
 nxt_go_request_t
 nxt_go_process_port_msg(void *buf, size_t buf_len, void *oob, size_t oob_len)
 {
diff --git a/src/nginext/nxt_go_lib.h b/src/nginext/nxt_go_lib.h
index 1c7f61c0..220fc9b3 100644
--- a/src/nginext/nxt_go_lib.h
+++ b/src/nginext/nxt_go_lib.h
@@ -31,7 +31,7 @@ int nxt_go_request_close(nxt_go_request_t r);
 
 int nxt_go_request_done(nxt_go_request_t r);
 
-void nxt_go_listen_and_serve();
+void nxt_go_ready();
 
 nxt_go_request_t nxt_go_process_port_msg(void *buf, size_t buf_len,
     void *oob, size_t oob_len);
diff --git a/src/nginext/nxt_go_log.h b/src/nginext/nxt_go_log.h
index da1ce6a0..d596cfb3 100644
--- a/src/nginext/nxt_go_log.h
+++ b/src/nginext/nxt_go_log.h
@@ -16,8 +16,7 @@
 #if (NXT_DEBUG)
 
 #define nxt_go_debug(fmt, ARGS...) \
-    fprintf(stdout, "go debug[%p]: " fmt "\n", \
-            (void *) (intptr_t) pthread_self(), ##ARGS)
+    fprintf(stderr, "[go debug] " fmt "\n", ##ARGS)
 
 #else
 
@@ -26,10 +25,10 @@
 #endif
 
 #define nxt_go_warn(fmt, ARGS...) \
-    fprintf(stdout, "go warn: " fmt "\n", ##ARGS)
+    fprintf(stderr, "[go warn] " fmt "\n", ##ARGS)
 
 #define nxt_go_error(fmt, ARGS...) \
-    fprintf(stdout, "go error: " fmt "\n", ##ARGS)
+    fprintf(stderr, "[go error] " fmt "\n", ##ARGS)
 
 
 #endif /* _NXT_GO_LOG_H_INCLUDED_ */
diff --git a/src/nginext/nxt_go_run_ctx.c b/src/nginext/nxt_go_run_ctx.c
index 2d284bc5..0788b0d1 100644
--- a/src/nginext/nxt_go_run_ctx.c
+++ b/src/nginext/nxt_go_run_ctx.c
@@ -168,6 +168,8 @@ nxt_go_ctx_init(nxt_go_run_ctx_t *ctx, nxt_port_msg_t *port_msg,
     ctx->wport_msg.type = NXT_PORT_MSG_DATA;
     ctx->wport_msg.mmap = 1;
 
+    ctx->wmmap_msg = (nxt_port_mmap_msg_t *) ( &ctx->wport_msg + 1 );
+
     return nxt_go_ctx_init_rbuf(ctx);
 }
 
@@ -197,6 +199,7 @@ nxt_go_ctx_add_msg(nxt_go_run_ctx_t *ctx, nxt_port_msg_t *port_msg, size_t size)
 nxt_int_t
 nxt_go_ctx_flush(nxt_go_run_ctx_t *ctx, int last)
 {
+    int       i;
     nxt_int_t rc;
 
     if (last != 0) {
@@ -205,6 +208,13 @@ nxt_go_ctx_flush(nxt_go_run_ctx_t *ctx, int last)
 
     nxt_go_debug("flush buffers (%d)", last);
 
+    for (i = 0; i < ctx->nwbuf; i++) {
+        nxt_port_mmap_msg_t *m = ctx->wmmap_msg + i;
+
+        nxt_go_debug("  mmap_msg[%d]={%d, %d, %d}", i,
+                     m->mmap_id, m->chunk_id, m->size);
+    }
+
     rc = nxt_go_port_send(ctx->msg.port_msg->pid, ctx->msg.port_msg->reply_port,
                           &ctx->wport_msg, sizeof(nxt_port_msg_t) +
                           ctx->nwbuf * sizeof(nxt_port_mmap_msg_t), NULL, 0);
diff --git a/src/nginext/nxt_go_run_ctx.h b/src/nginext/nxt_go_run_ctx.h
index 244bd7c3..c7c3da15 100644
--- a/src/nginext/nxt_go_run_ctx.h
+++ b/src/nginext/nxt_go_run_ctx.h
@@ -36,6 +36,7 @@ typedef struct {
     nxt_go_msg_t         msg;
 
     nxt_go_process_t     *process;
+    nxt_port_mmap_msg_t  *wmmap_msg;
 
     uint32_t             nrbuf;
     nxt_buf_t            rbuf;
@@ -43,7 +44,7 @@ typedef struct {
     uint32_t             nwbuf;
     nxt_buf_t            wbuf;
     nxt_port_msg_t       wport_msg;
-    nxt_port_mmap_msg_t  wmmap_msg[8];
+    char                 wmmap_msg_buf[ sizeof(nxt_port_mmap_msg_t) * 8 ];
 
     nxt_app_request_t    r;
 
diff --git a/src/nginext/port.go b/src/nginext/port.go
index 01921b5a..639b73d5 100644
--- a/src/nginext/port.go
+++ b/src/nginext/port.go
@@ -25,6 +25,7 @@ type port_key struct {
 
 type port struct {
 	key port_key
+	t   int
 	rcv *net.UnixConn
 	snd *net.UnixConn
 }
@@ -32,6 +33,7 @@ type port struct {
 type port_registry struct {
 	sync.RWMutex
 	m map[port_key]*port
+	t [5]*port
 }
 
 var port_registry_ port_registry
@@ -44,6 +46,31 @@ func find_port(key port_key) *port {
 	return res
 }
 
+func remove_by_pid(pid int) {
+	port_registry_.Lock()
+	if port_registry_.m != nil {
+		for k, p := range port_registry_.m {
+			if k.pid == pid {
+				if port_registry_.t[p.t] == p {
+					port_registry_.t[p.t] = nil
+				}
+
+				delete(port_registry_.m, k)
+			}
+		}
+	}
+
+	port_registry_.Unlock()
+}
+
+func master_port() *port {
+	port_registry_.RLock()
+	res := port_registry_.t[1]
+	port_registry_.RUnlock()
+
+	return res
+}
+
 func add_port(p *port) {
 	port_registry_.Lock()
 	if port_registry_.m == nil {
@@ -51,6 +78,7 @@ func add_port(p *port) {
 	}
 
 	port_registry_.m[p.key] = p
+	port_registry_.t[p.t] = p
 
 	port_registry_.Unlock()
 }
@@ -85,7 +113,6 @@ func getUnixConn(fd int) *net.UnixConn {
 		return nil
 	}
 
-	fmt.Printf("Unix-domain socket %d\n", fd)
 	return uc
 }
 
@@ -116,20 +143,36 @@ func nxt_go_port_send(pid C.int, id C.int, buf unsafe.Pointer, buf_size C.int, o
 	return 0
 }
 
+//export nxt_go_master_send
+func nxt_go_master_send(buf unsafe.Pointer, buf_size C.int, oob unsafe.Pointer, oob_size C.int) C.int {
+	p := master_port()
+
+	if p != nil {
+		n, oobn, err := p.snd.WriteMsgUnix(C.GoBytes(buf, buf_size), C.GoBytes(oob, oob_size), nil)
+
+		if err != nil {
+			fmt.Printf("write result %d (%d), %s\n", n, oobn, err)
+		}
+
+		return C.int(n)
+	}
+
+	return 0
+}
+
 func new_port(pid int, id int, t int, rcv int, snd int) *port {
 	p := &port{
 		key: port_key{
 			pid: pid,
 			id:  id,
 		},
+		t: t,
 		rcv: getUnixConn(rcv),
 		snd: getUnixConn(snd),
 	}
 
 	add_port(p)
 
-	fmt.Printf("new_port: %d, %d, %d, %d\n", pid, id, rcv, snd)
-
 	return p
 }
 
@@ -137,9 +180,7 @@ func (p *port) read() {
 	var buf [16384]byte
 	var oob [1024]byte
 
-	n, oobn, _, _, err := p.rcv.ReadMsgUnix(buf[:], oob[:])
-
-	fmt.Printf("read result %d (%d), %s\n", n, oobn, err)
+	n, oobn, _, _, _ := p.rcv.ReadMsgUnix(buf[:], oob[:])
 
 	m := new_cmsg(buf[:n], oob[:oobn])
 
diff --git a/src/nxt_go.c b/src/nxt_go.c
index 4c77de0d..17240804 100644
--- a/src/nxt_go.c
+++ b/src/nxt_go.c
@@ -8,7 +8,7 @@
 #include <nxt_application.h>
 
 
-static nxt_int_t nxt_go_init(nxt_task_t *task);
+static nxt_int_t nxt_go_init(nxt_task_t *task, nxt_common_app_conf_t *conf);
 
 static nxt_int_t nxt_go_prepare_msg(nxt_task_t *task,
                       nxt_app_request_t *r, nxt_app_wmsg_t *msg);
@@ -16,8 +16,6 @@ static nxt_int_t nxt_go_prepare_msg(nxt_task_t *task,
 static nxt_int_t nxt_go_run(nxt_task_t *task,
                       nxt_app_rmsg_t *rmsg, nxt_app_wmsg_t *msg);
 
-static nxt_str_t nxt_go_path;
-
 nxt_application_module_t  nxt_go_module = {
     nxt_go_init,
     nxt_go_prepare_msg,
@@ -31,39 +29,9 @@ nxt_go_module_init(nxt_thread_t *thr, nxt_runtime_t *rt);
 nxt_int_t
 nxt_go_module_init(nxt_thread_t *thr, nxt_runtime_t *rt)
 {
-    char        **argv;
-    u_char      *p;
-
-    argv = nxt_process_argv;
-
-    while (*argv != NULL) {
-        p = (u_char *) *argv++;
-
-        if (nxt_strcmp(p, "--go") == 0) {
-            if (*argv == NULL) {
-                nxt_log_error(NXT_LOG_ERR, thr->log,
-                              "no argument for option \"--go\"");
-                return NXT_ERROR;
-            }
-
-            p = (u_char *) *argv;
-
-            nxt_go_path.start = p;
-            nxt_go_path.length = nxt_strlen(p);
-
-            nxt_log_error(NXT_LOG_INFO, thr->log,
-                          "go program \"%V\"",
-                          &nxt_go_path);
-
-            nxt_app = &nxt_go_module;
-
-            return NXT_OK;
-        }
-    }
-
-    nxt_log_error(NXT_LOG_ERR, thr->log, "no option \"--go\" specified");
+    nxt_app_modules[NXT_APP_GO] = &nxt_go_module;
 
-    return NXT_ERROR;
+    return NXT_OK;
 }
 
 extern char  **environ;
@@ -77,59 +45,70 @@ nxt_sock_no_cloexec(nxt_socket_t fd)
     return fcntl(fd, F_SETFD, 0);
 }
 
+
 static nxt_int_t
-nxt_go_init(nxt_task_t *task)
+nxt_go_init(nxt_task_t *task, nxt_common_app_conf_t *conf)
 {
-    char *go_ports = getenv("NXT_GO_PORTS");
-
-    nxt_debug(task, "initialize go app, NXT_GO_PORTS=%s",
-              go_ports ? go_ports : "NULL");
-
-    if (go_ports == NULL) {
-        u_char buf[256];
-        u_char *p = buf;
+    char               *go_path;
+    char               *argv[2];
+    u_char             buf[256];
+    u_char             *p;
+    u_char             stream_buf[32];
+    nxt_port_t         *port;
+    nxt_runtime_t      *rt;
+    nxt_go_app_conf_t  *c;
+
+    c = &conf->u.go;
+    rt = task->thread->runtime;
+    p = buf;
+
+    nxt_runtime_port_each(rt, port) {
+
+        if (port->pid != nxt_pid && port->type != NXT_PROCESS_MASTER) {
+            continue;
+        }
 
-        nxt_runtime_t *rt = task->thread->runtime;
-        nxt_port_t *port;
+        if (port->pid == nxt_pid) {
+            nxt_sprintf(stream_buf, stream_buf + sizeof(stream_buf),
+                        "%uD", port->process->init->stream);
 
-        nxt_runtime_port_each(rt, port) {
+            setenv("NXT_GO_STREAM", (char *)stream_buf, 1);
+        }
 
-            nxt_debug(task, "port %PI, %ud, (%d, %d)", port->pid, port->id,
-                      port->pair[0], port->pair[1]);
+        nxt_debug(task, "port %PI, %ud, (%d, %d)", port->pid, port->id,
+                  port->pair[0], port->pair[1]);
 
-            p = nxt_sprintf(p, buf + sizeof(buf), "%PI,%ud,%d,%d,%d;",
-                            port->pid, port->id, (int)port->type,
-                            port->pair[0], port->pair[1]);
+        p = nxt_sprintf(p, buf + sizeof(buf), "%PI,%ud,%d,%d,%d;",
+                        port->pid, port->id, (int)port->type,
+                        port->pair[0], port->pair[1]);
 
-            if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[0]))) {
-                nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
-            }
+        if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[0]))) {
+            nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
+        }
 
-            if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[1]))) {
-                nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
-            }
+        if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[1]))) {
+            nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
+        }
 
-        } nxt_runtime_port_loop;
+    } nxt_runtime_port_loop;
 
-        *p = '\0';
-        nxt_debug(task, "update NXT_GO_PORTS=%s", buf);
+    *p = '\0';
+    nxt_debug(task, "update NXT_GO_PORTS=%s", buf);
 
-        setenv("NXT_GO_PORTS", (char *)buf, 1);
+    setenv("NXT_GO_PORTS", (char *)buf, 1);
 
-        char *argv[] = {
-            (char *)nxt_go_path.start,
-            (char *)"--no-daemonize",
-            (char *)"--app", NULL };
+    go_path = malloc(c->executable.length + 1);
+    nxt_memcpy(go_path, c->executable.start, c->executable.length);
+    go_path[c->executable.length] = '\0';
 
-        (void) execve((char *)nxt_go_path.start, argv, environ);
+    argv[0] = go_path;
+    argv[1] = NULL;
 
-        nxt_log(task, NXT_LOG_WARN, "execve(%V) failed %E", &nxt_go_path,
-                nxt_errno);
+    (void) execve(go_path, argv, environ);
 
-        return NXT_ERROR;
-    }
+    nxt_log(task, NXT_LOG_WARN, "execve(%s) failed %E", go_path, nxt_errno);
 
-    return NXT_OK;
+    return NXT_ERROR;
 }