mirror of
https://github.com/criyle/go-sandbox.git
synced 2025-11-04 14:49:53 +08:00
container: allocate buffers in advance
This commit is contained in:
parent
c1d9a5572f
commit
963e027b0a
@ -1,6 +1,6 @@
|
||||
package container
|
||||
|
||||
type cmdType int
|
||||
type cmdType int8
|
||||
|
||||
const (
|
||||
cmdPing cmdType = iota + 1
|
||||
|
||||
@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"sync"
|
||||
"syscall"
|
||||
|
||||
@ -77,6 +78,11 @@ func Init() (err error) {
|
||||
// limit container resource usage
|
||||
runtime.GOMAXPROCS(containerMaxProc)
|
||||
|
||||
// ensure there's no fd leak to child process (e.g. VSCode leaks ptmx fd)
|
||||
if err := closeOnExecAllFds(); err != nil {
|
||||
return fmt.Errorf("container_init: failed to close_on_exec all fd %v", err)
|
||||
}
|
||||
|
||||
// new_container environment shared the socket at fd 3 (marked close_exec)
|
||||
const defaultFd = 3
|
||||
soc, err := unixsocket.NewSocket(defaultFd)
|
||||
@ -302,3 +308,20 @@ func (c *containerServer) sendErrorReply(ft string, v ...interface{}) error {
|
||||
}
|
||||
return c.sendReply(reply{Error: errRep}, unixsocket.Msg{})
|
||||
}
|
||||
|
||||
func closeOnExecAllFds() error {
|
||||
// get all fd from /proc/self/fd
|
||||
const fdPath = "/proc/self/fd"
|
||||
fds, err := os.ReadDir(fdPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, f := range fds {
|
||||
fd, err := strconv.Atoi(f.Name())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
syscall.CloseOnExec(fd)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -13,12 +13,12 @@ import (
|
||||
|
||||
// cmd is the control message send into container
|
||||
type cmd struct {
|
||||
Cmd cmdType // type of the cmd
|
||||
|
||||
OpenCmd []OpenCmd // open argument
|
||||
DeleteCmd *deleteCmd // delete argument
|
||||
ExecCmd *execCmd // execve argument
|
||||
ConfCmd *confCmd // to set configuration
|
||||
|
||||
Cmd cmdType // type of the cmd
|
||||
}
|
||||
|
||||
// OpenCmd correspond to a single open syscall
|
||||
|
||||
@ -4,7 +4,6 @@ import (
|
||||
"bytes"
|
||||
"encoding/gob"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/criyle/go-sandbox/pkg/unixsocket"
|
||||
)
|
||||
@ -12,26 +11,32 @@ import (
|
||||
// 16k buffsize
|
||||
const bufferSize = 16 << 10
|
||||
|
||||
var bufferPool = sync.Pool{
|
||||
New: func() interface{} {
|
||||
return make([]byte, bufferSize)
|
||||
},
|
||||
}
|
||||
|
||||
type socket struct {
|
||||
*unixsocket.Socket
|
||||
|
||||
recvBuff bytes.Buffer
|
||||
decoder *gob.Decoder
|
||||
buff []byte
|
||||
|
||||
decoder *gob.Decoder
|
||||
recvBuff bufferRotater
|
||||
|
||||
sendBuff bytes.Buffer
|
||||
encoder *gob.Encoder
|
||||
sendBuff bytes.Buffer
|
||||
}
|
||||
|
||||
// bufferRotater replace the underlying Buffers to avoid allocation
|
||||
type bufferRotater struct {
|
||||
*bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *bufferRotater) Rotate(buffer *bytes.Buffer) {
|
||||
b.Buffer = buffer
|
||||
}
|
||||
|
||||
func newSocket(s *unixsocket.Socket) *socket {
|
||||
soc := socket{
|
||||
Socket: s,
|
||||
}
|
||||
soc.buff = make([]byte, bufferSize)
|
||||
soc.decoder = gob.NewDecoder(&soc.recvBuff)
|
||||
soc.encoder = gob.NewEncoder(&soc.sendBuff)
|
||||
|
||||
@ -39,15 +44,11 @@ func newSocket(s *unixsocket.Socket) *socket {
|
||||
}
|
||||
|
||||
func (s *socket) RecvMsg(e interface{}) (msg unixsocket.Msg, err error) {
|
||||
buff := bufferPool.Get().([]byte)
|
||||
defer bufferPool.Put(buff)
|
||||
|
||||
n, msg, err := s.Socket.RecvMsg(buff)
|
||||
n, msg, err := s.Socket.RecvMsg(s.buff)
|
||||
if err != nil {
|
||||
return msg, fmt.Errorf("RecvMsg: %v", err)
|
||||
}
|
||||
s.recvBuff.Reset()
|
||||
s.recvBuff.Write(buff[:n])
|
||||
s.recvBuff.Rotate(bytes.NewBuffer(s.buff[:n]))
|
||||
|
||||
if err := s.decoder.Decode(e); err != nil {
|
||||
return msg, fmt.Errorf("RecvMsg: failed to decode %v", err)
|
||||
|
||||
@ -7,22 +7,18 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// oob size default to page size
|
||||
const oobSize = 4096
|
||||
|
||||
// use pool to minimize allocate gabage collector overhead
|
||||
var oobPool = sync.Pool{
|
||||
New: func() interface{} {
|
||||
return make([]byte, oobSize)
|
||||
},
|
||||
}
|
||||
const oobSize = 4 << 10 // 4kb
|
||||
|
||||
// Socket wrappers a unix socket connection
|
||||
type Socket net.UnixConn
|
||||
type Socket struct {
|
||||
*net.UnixConn
|
||||
sendBuff []byte
|
||||
recvBuff []byte
|
||||
}
|
||||
|
||||
// Msg is the oob msg with the message
|
||||
type Msg struct {
|
||||
@ -30,6 +26,14 @@ type Msg struct {
|
||||
Cred *syscall.Ucred // unix credential
|
||||
}
|
||||
|
||||
func newSocket(conn *net.UnixConn) *Socket {
|
||||
return &Socket{
|
||||
UnixConn: conn,
|
||||
sendBuff: make([]byte, oobSize),
|
||||
recvBuff: make([]byte, oobSize),
|
||||
}
|
||||
}
|
||||
|
||||
// NewSocket creates Socket conn struct using existing unix socket fd
|
||||
// creates by socketpair or net.DialUnix and mark it as close_on_exec (avoid fd leak)
|
||||
// it need SOCK_SEQPACKET socket for reliable transfer
|
||||
@ -55,7 +59,7 @@ func NewSocket(fd int) (*Socket, error) {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("NewSocket: %d is not a valid unix socket connection", fd)
|
||||
}
|
||||
return (*Socket)(unixConn), nil
|
||||
return newSocket(unixConn), nil
|
||||
}
|
||||
|
||||
// NewSocketPair creates connected unix socketpair using SOCK_SEQPACKET
|
||||
@ -84,7 +88,7 @@ func NewSocketPair() (*Socket, *Socket, error) {
|
||||
|
||||
// SetPassCred set sockopt for pass cred for unix socket
|
||||
func (s *Socket) SetPassCred(option int) error {
|
||||
sysconn, err := (*net.UnixConn)(s).SyscallConn()
|
||||
sysconn, err := s.SyscallConn()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -95,10 +99,7 @@ func (s *Socket) SetPassCred(option int) error {
|
||||
|
||||
// SendMsg sendmsg to unix socket and encode possible unix right / credential
|
||||
func (s *Socket) SendMsg(b []byte, m Msg) error {
|
||||
buf := oobPool.Get().([]byte)
|
||||
defer oobPool.Put(buf)
|
||||
|
||||
oob := bytes.NewBuffer(buf[:0])
|
||||
oob := bytes.NewBuffer(s.sendBuff[:0])
|
||||
if len(m.Fds) > 0 {
|
||||
oob.Write(syscall.UnixRights(m.Fds...))
|
||||
}
|
||||
@ -106,7 +107,7 @@ func (s *Socket) SendMsg(b []byte, m Msg) error {
|
||||
oob.Write(syscall.UnixCredentials(m.Cred))
|
||||
}
|
||||
|
||||
_, _, err := (*net.UnixConn)(s).WriteMsgUnix(b, oob.Bytes(), nil)
|
||||
_, _, err := s.WriteMsgUnix(b, oob.Bytes(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -115,16 +116,13 @@ func (s *Socket) SendMsg(b []byte, m Msg) error {
|
||||
|
||||
// RecvMsg recvmsg from unix socket and parse possible unix right / credential
|
||||
func (s *Socket) RecvMsg(b []byte) (int, Msg, error) {
|
||||
oob := oobPool.Get().([]byte)
|
||||
defer oobPool.Put(oob)
|
||||
|
||||
var msg Msg
|
||||
n, oobn, _, _, err := (*net.UnixConn)(s).ReadMsgUnix(b, oob)
|
||||
n, oobn, _, _, err := s.ReadMsgUnix(b, s.recvBuff)
|
||||
if err != nil {
|
||||
return 0, msg, err
|
||||
}
|
||||
// parse oob msg
|
||||
msgs, err := syscall.ParseSocketControlMessage(oob[:oobn])
|
||||
msgs, err := syscall.ParseSocketControlMessage(s.recvBuff[:oobn])
|
||||
if err != nil {
|
||||
return 0, msg, err
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user