Go: worker initialization. READY message to master.
This commit is contained in:
Max Romanov
parent
51bbdd3338
commit
a97a5e8a0b
8 changed files with 138 additions and 95 deletions
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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_ */
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
||||
|
|
|
|||
87
src/nxt_go.c
87
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;
|
||||
nxt_app_modules[NXT_APP_GO] = &nxt_go_module;
|
||||
|
||||
return NXT_OK;
|
||||
}
|
||||
}
|
||||
|
||||
nxt_log_error(NXT_LOG_ERR, thr->log, "no option \"--go\" specified");
|
||||
|
||||
return NXT_ERROR;
|
||||
}
|
||||
|
||||
extern char **environ;
|
||||
|
|
@ -77,23 +45,36 @@ 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) {
|
||||
char *go_path;
|
||||
char *argv[2];
|
||||
u_char buf[256];
|
||||
u_char *p = buf;
|
||||
|
||||
nxt_runtime_t *rt = task->thread->runtime;
|
||||
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;
|
||||
}
|
||||
|
||||
if (port->pid == nxt_pid) {
|
||||
nxt_sprintf(stream_buf, stream_buf + sizeof(stream_buf),
|
||||
"%uD", port->process->init->stream);
|
||||
|
||||
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]);
|
||||
|
||||
|
|
@ -116,20 +97,18 @@ nxt_go_init(nxt_task_t *task)
|
|||
|
||||
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);
|
||||
|
||||
nxt_log(task, NXT_LOG_WARN, "execve(%s) failed %E", go_path, nxt_errno);
|
||||
|
||||
return NXT_ERROR;
|
||||
}
|
||||
|
||||
return NXT_OK;
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue