From a97a5e8a0b7a921905994bae101d4f74f5a8aaa4 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Wed, 12 Jul 2017 20:32:18 +0300 Subject: [PATCH] Go: worker initialization. READY message to master. --- src/nginext/nginext.go | 4 +- src/nginext/nxt_go_lib.c | 35 +++++++---- src/nginext/nxt_go_lib.h | 2 +- src/nginext/nxt_go_log.h | 7 +-- src/nginext/nxt_go_run_ctx.c | 10 +++ src/nginext/nxt_go_run_ctx.h | 3 +- src/nginext/port.go | 53 ++++++++++++++-- src/nxt_go.c | 119 +++++++++++++++-------------------- 8 files changed, 138 insertions(+), 95 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 -#include -#include - -#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 -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; + nxt_app_modules[NXT_APP_GO] = &nxt_go_module; - 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"); - - 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"); + 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; - nxt_debug(task, "initialize go app, NXT_GO_PORTS=%s", - go_ports ? go_ports : "NULL"); + c = &conf->u.go; + rt = task->thread->runtime; + p = buf; - if (go_ports == NULL) { - u_char buf[256]; - u_char *p = buf; + nxt_runtime_port_each(rt, port) { - nxt_runtime_t *rt = task->thread->runtime; - nxt_port_t *port; + if (port->pid != nxt_pid && port->type != NXT_PROCESS_MASTER) { + continue; + } - nxt_runtime_port_each(rt, port) { + if (port->pid == nxt_pid) { + nxt_sprintf(stream_buf, stream_buf + sizeof(stream_buf), + "%uD", port->process->init->stream); - nxt_debug(task, "port %PI, %ud, (%d, %d)", port->pid, port->id, - port->pair[0], port->pair[1]); + setenv("NXT_GO_STREAM", (char *)stream_buf, 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]); + nxt_debug(task, "port %PI, %ud, (%d, %d)", port->pid, port->id, + 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); - } + 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[1]))) { - 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); + } - } nxt_runtime_port_loop; + if (nxt_slow_path(nxt_sock_no_cloexec(port->pair[1]))) { + nxt_log(task, NXT_LOG_WARN, "fcntl() failed %E", nxt_errno); + } - *p = '\0'; - nxt_debug(task, "update NXT_GO_PORTS=%s", buf); + } nxt_runtime_port_loop; - setenv("NXT_GO_PORTS", (char *)buf, 1); + *p = '\0'; + nxt_debug(task, "update NXT_GO_PORTS=%s", buf); - char *argv[] = { - (char *)nxt_go_path.start, - (char *)"--no-daemonize", - (char *)"--app", NULL }; + setenv("NXT_GO_PORTS", (char *)buf, 1); - (void) execve((char *)nxt_go_path.start, argv, environ); + go_path = malloc(c->executable.length + 1); + nxt_memcpy(go_path, c->executable.start, c->executable.length); + go_path[c->executable.length] = '\0'; - nxt_log(task, NXT_LOG_WARN, "execve(%V) failed %E", &nxt_go_path, - nxt_errno); + argv[0] = go_path; + argv[1] = NULL; - return NXT_ERROR; - } + (void) execve(go_path, argv, environ); - return NXT_OK; + nxt_log(task, NXT_LOG_WARN, "execve(%s) failed %E", go_path, nxt_errno); + + return NXT_ERROR; }