Introducing Unit version check in Go package.

To communicate with the Go program, Unit setup environment variable
named NXT_GO_PORTS with value contains Unit version, stream id to confirm
application is started, and Unit ports information.  Go Unit package parses
this string and compares runtime version with compile time version.  In case
of parse error or version mismatch, ListenAndServe() returns with the error.

src/go/unit/nxt_go_lib.c

@@ -112,18 +112,11 @@ nxt_go_request_done(nxt_go_request_t r)
 
 
 void
-nxt_go_ready()
+nxt_go_ready(uint32_t stream)
 {
-    char           *go_stream;
+    nxt_port_msg_t  port_msg;
-    nxt_port_msg_t port_msg;
 
-    go_stream = getenv("NXT_GO_STREAM");
+    port_msg.stream = 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_PROCESS_READY;
@@ -141,3 +134,10 @@ nxt_go_process_port_msg(uintptr_t buf, size_t buf_len, uintptr_t oob, size_t oob
 {
     return nxt_go_port_on_read((void *) buf, buf_len, (void *) oob, oob_len);
 }
+
+
+const char *
+nxt_go_version()
+{
+    return NXT_VERSION;
+}

src/go/unit/nxt_go_lib.h

@@ -30,10 +30,12 @@ int nxt_go_request_close(nxt_go_request_t r);
 
 int nxt_go_request_done(nxt_go_request_t r);
 
-void nxt_go_ready();
+void nxt_go_ready(uint32_t stream);
 
 nxt_go_request_t nxt_go_process_port_msg(uintptr_t buf, size_t buf_len,
     uintptr_t oob, size_t oob_len);
 
+const char *nxt_go_version();
+
 
 #endif /* _NXT_GO_LIB_H_INCLUDED_ */

src/go/unit/unit.go

@@ -11,6 +11,7 @@ package unit
 import "C"
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -65,31 +66,53 @@ func ListenAndServe(addr string, handler http.Handler) error {
 	var read_port *port
 
 	go_ports_env := os.Getenv("NXT_GO_PORTS")
+	if go_ports_env == "" {
+		return http.ListenAndServe(addr, handler)
+	}
+
+	nxt_go_debug("NXT_GO_PORTS=%s", go_ports_env)
 
 	ports := strings.Split(go_ports_env, ";")
 	pid := os.Getpid()
 
-	for _, port_str := range ports {
+	if len(ports) != 4 {
-		if len(port_str) <= 0 {
+		return errors.New("Invalid NXT_GO_PORTS format")
-			continue
+	}
-		}
 
+	nxt_go_debug("version=%s", ports[0])
+
+	builtin_version := C.GoString(C.nxt_go_version())
+
+	if ports[0] != builtin_version {
+		return fmt.Errorf("Versions mismatch: Unit %s, while application is built with %s",
+			ports[0], builtin_version)
+	}
+
+	stream, stream_err := strconv.Atoi(ports[1])
+	if stream_err != nil {
+		return stream_err
+	}
+
+	read_port = nil
+
+	for _, port_str := range ports[2:] {
 		attrs := strings.Split(port_str, ",")
 
+		if len(attrs) != 5 {
+			return fmt.Errorf("Invalid port format: unexpected port attributes number %d, while 5 expected",
+				len(attrs))
+		}
+
 		var attrsN [5]int
 		var err error
 		for i, attr := range attrs {
 			attrsN[i], err = strconv.Atoi(attr)
 			if err != nil {
-				fmt.Printf("err %s\n", err)
+				return fmt.Errorf("Invalid port format: number attribute expected at %d position instead of '%s'",
-				break
+					i, attr);
 			}
 		}
 
-		if err != nil {
-			continue
-		}
-
 		p := new_port(attrsN[0], attrsN[1], attrsN[2], attrsN[3], attrsN[4])
 
 		if attrsN[0] == pid {
@@ -97,17 +120,17 @@ func ListenAndServe(addr string, handler http.Handler) error {
 		}
 	}
 
-	if read_port != nil {
+	if read_port == nil {
-		C.nxt_go_ready()
+		return errors.New("Application read port not found");
+	}
 
-		for !nxt_go_quit {
+	C.nxt_go_ready(C.uint32_t(stream))
-			err := read_port.read(handler)
+
-			if err != nil {
+	for !nxt_go_quit {
-				return err
+		err := read_port.read(handler)
-			}
+		if err != nil {
+			return err
 		}
-	} else {
-		return http.ListenAndServe(addr, handler)
 	}
 
 	return nil

src/nxt_go.c

@@ -25,13 +25,34 @@ nxt_application_module_t  nxt_go_module = {
 
 extern char  **environ;
 
-nxt_inline int
+nxt_inline nxt_int_t
-nxt_sock_no_cloexec(nxt_socket_t fd)
+nxt_go_fd_no_cloexec(nxt_task_t *task, nxt_socket_t fd)
 {
+    int  res, flags;
+
     if (fd == -1) {
-        return 0;
+        return NXT_OK;
     }
-    return fcntl(fd, F_SETFD, 0);
+
+    flags = fcntl(fd, F_GETFD);
+
+    if (nxt_slow_path(flags == -1)) {
+        nxt_log(task, NXT_LOG_CRIT, "fcntl(%d, F_GETFD) failed %E",
+                fd, nxt_errno);
+        return NXT_ERROR;
+    }
+
+    flags &= ~FD_CLOEXEC;
+
+    res = fcntl(fd, F_SETFD, flags);
+
+    if (nxt_slow_path(res == -1)) {
+        nxt_log(task, NXT_LOG_CRIT, "fcntl(%d, F_SETFD) failed %E",
+                fd, nxt_errno);
+        return NXT_ERROR;
+    }
+
+    return NXT_OK;
 }
 
 
@@ -40,58 +61,69 @@ nxt_go_init(nxt_task_t *task, nxt_common_app_conf_t *conf)
 {
     char               *argv[2];
     u_char             buf[256];
-    u_char             *p;
+    u_char             *p, *end;
-    u_char             stream_buf[32];
+    nxt_int_t          rc;
-    nxt_port_t         *port;
+    nxt_port_t         *my_port, *main_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) {
+    main_port = rt->port_by_type[NXT_PROCESS_MAIN];
+    my_port = nxt_runtime_port_find(rt, nxt_pid, 0);
 
-        if (port->pid != nxt_pid && port->type != NXT_PROCESS_MAIN) {
+    if (nxt_slow_path(main_port == NULL || my_port == NULL)) {
-            continue;
+        return NXT_ERROR;
-        }
+    }
 
-        if (port->pid == nxt_pid) {
+    rc = nxt_go_fd_no_cloexec(task, main_port->pair[1]);
-            nxt_sprintf(stream_buf, stream_buf + sizeof(stream_buf),
+    if (nxt_slow_path(rc != NXT_OK)) {
-                        "%uD", port->process->init->stream);
+        return NXT_ERROR;
+    }
 
-            setenv("NXT_GO_STREAM", (char *) stream_buf, 1);
+    rc = nxt_go_fd_no_cloexec(task, my_port->pair[0]);
-        }
+    if (nxt_slow_path(rc != NXT_OK)) {
+        return NXT_ERROR;
+    }
 
-        nxt_debug(task, "port %PI, %ud, (%d, %d)", port->pid, port->id,
+    end = buf + sizeof(buf);
-                  port->pair[0], port->pair[1]);
 
-        p = nxt_sprintf(p, buf + sizeof(buf), "%PI,%ud,%d,%d,%d;",
+    p = nxt_sprintf(buf, end,
-                        port->pid, port->id, (int) port->type,
+                    "%s;%uD;"
-                        port->pair[0], port->pair[1]);
+                    "%PI,%ud,%d,%d,%d;"
+                    "%PI,%ud,%d,%d,%d%Z",
+                    NXT_VERSION, my_port->process->init->stream,
+                    main_port->pid, main_port->id, (int) main_port->type,
+                    -1, main_port->pair[1],
+                    my_port->pid, my_port->id, (int) my_port->type,
+                    my_port->pair[0], -1);
 
-        if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[0]))) {
+    if (nxt_slow_path(p == end)) {
-            nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
+        nxt_log(task, NXT_LOG_ALERT,
-        }
+                "internal error: buffer too small for NXT_GO_PORTS");
 
-        if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[1]))) {
+        return NXT_ERROR;
-            nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno);
+    }
-        }
 
-    } nxt_runtime_port_loop;
-
-    *p = '\0';
     nxt_debug(task, "update NXT_GO_PORTS=%s", buf);
 
-    setenv("NXT_GO_PORTS", (char *) buf, 1);
+    rc = setenv("NXT_GO_PORTS", (char *) buf, 1);
+    if (nxt_slow_path(rc == -1)) {
+        nxt_log(task, NXT_LOG_CRIT, "setenv(NXT_GO_PORTS, %s) failed %E",
+                buf, nxt_errno);
+
+        return NXT_ERROR;
+    }
+
+    c = &conf->u.go;
 
     argv[0] = c->executable;
     argv[1] = NULL;
 
     (void) execve(c->executable, argv, environ);
 
-    nxt_log(task, NXT_LOG_WARN, "execve(%s) failed %E", c->executable,
+    nxt_log(task, NXT_LOG_CRIT, "execve(%s) failed %E",
-            nxt_errno);
+            c->executable, nxt_errno);
 
     return NXT_ERROR;
 }