port.go

/*
 * Copyright (C) Max Romanov
 * Copyright (C) NGINX, Inc.
 */

package unit

/*
#include "nxt_cgo_lib.h"
*/
import"C"

import (
"io"
"net"
"os"
"sync"
"unsafe"
)

typeport_keystruct {
pidint
idint
}

typeportstruct {
keyport_key
rcv*net.UnixConn
snd*net.UnixConn
}

typeport_registrystruct {
 sync.RWMutex
mmap[port_key]*port
}

varport_registry_port_registry

funcfind_port(keyport_key) *port {
port_registry_.RLock()
res:=port_registry_.m[key]
port_registry_.RUnlock()

returnres
}

funcadd_port(p*port) {

port_registry_.Lock()
ifport_registry_.m==nil {
port_registry_.m=make(map[port_key]*port)
 }

old:=port_registry_.m[p.key]

ifold==nil {
port_registry_.m[p.key] =p
 }

port_registry_.Unlock()
}

func (p*port) Close() {
ifp.rcv!=nil {
p.rcv.Close()
 }

ifp.snd!=nil {
p.snd.Close()
 }
}

funcgetUnixConn(fdint) *net.UnixConn {
iffd<0 {
returnnil
 }

f:=os.NewFile(uintptr(fd), "sock")
deferf.Close()

c, err:=net.FileConn(f)
iferr!=nil {
nxt_go_alert("FileConn error %s", err)
returnnil
 }

uc, ok:=c.(*net.UnixConn)
if!ok {
nxt_go_alert("Not a Unix-domain socket %d", fd)
returnnil
 }

returnuc
}

//export nxt_go_add_port
funcnxt_go_add_port(ctx*C.nxt_unit_ctx_t, p*C.nxt_unit_port_t) C.int {

new_port:=&port{
key: port_key{
pid: int(p.id.pid),
id: int(p.id.id),
 },
rcv: getUnixConn(int(p.in_fd)),
snd: getUnixConn(int(p.out_fd)),
 }

add_port(new_port)

p.in_fd=-1
p.out_fd=-1

returnC.NXT_UNIT_OK
}

//export nxt_go_ready
funcnxt_go_ready(ctx*C.nxt_unit_ctx_t) C.int {
gofunc(ctx*C.nxt_unit_ctx_t) {
C.nxt_unit_run_shared(ctx)
 }(ctx)

returnC.NXT_UNIT_OK
}

//export nxt_go_remove_port
funcnxt_go_remove_port(unit*C.nxt_unit_t, ctx*C.nxt_unit_ctx_t,
p*C.nxt_unit_port_t) {

key:=port_key{
pid: int(p.id.pid),
id: int(p.id.id),
 }

port_registry_.Lock()
ifport_registry_.m!=nil {
delete(port_registry_.m, key)
 }

port_registry_.Unlock()
}

//export nxt_go_port_send
funcnxt_go_port_send(pid C.int, id C.int, buf unsafe.Pointer, buf_size C.int,
oob unsafe.Pointer, oob_size C.int) C.ssize_t {

key:=port_key{
pid: int(pid),
id: int(id),
 }

p:=find_port(key)

ifp==nil {
nxt_go_alert("port %d:%d not found", pid, id)
return0
 }

n, oobn, err:=p.snd.WriteMsgUnix(GoBytes(buf, buf_size),
GoBytes(oob, oob_size), nil)

iferr!=nil {
nxt_go_warn("write result %d (%d), %s", n, oobn, err)

n=-1
 }

returnC.ssize_t(n)
}

//export nxt_go_port_recv
funcnxt_go_port_recv(pid C.int, id C.int, buf unsafe.Pointer, buf_size C.int,
oob unsafe.Pointer, oob_size*C.size_t) C.ssize_t {

key:=port_key{
pid: int(pid),
id: int(id),
 }

p:=find_port(key)

ifp==nil {
nxt_go_alert("port %d:%d not found", pid, id)
return0
 }

n, oobn, _, _, err:=p.rcv.ReadMsgUnix(GoBytes(buf, buf_size),
GoBytes(oob, C.int(*oob_size)))

iferr!=nil {
ifnerr, ok:=err.(*net.OpError); ok {
ifnerr.Err==io.EOF {
return0
 }
 }

nxt_go_warn("read result %d (%d), %s", n, oobn, err)

n=-1

 } else {
*oob_size=C.size_t(oobn)
 }

returnC.ssize_t(n)
}