reduce container communication overhead

This commit is contained in:
criyle 2020-02-27 23:30:32 -05:00
parent 49bbe9caec
commit c21ae19d18
9 changed files with 50 additions and 62 deletions

View File

@ -219,7 +219,6 @@ func start() (*types.Result, error) {
Stack: stackLimit << 20, Stack: stackLimit << 20,
} }
debug("rlimit: ", rlims) debug("rlimit: ", rlims)
debug("defaultMount: ", mount.DefaultMounts)
actionDefault := seccomp.ActionKill actionDefault := seccomp.ActionKill
if showDetails { if showDetails {
@ -276,7 +275,7 @@ func start() (*types.Result, error) {
return nil, fmt.Errorf("cannot make temp root for new namespace") return nil, fmt.Errorf("cannot make temp root for new namespace")
} }
defer os.RemoveAll(root) defer os.RemoveAll(root)
mounts, err := mount.NewBuilder().WithMounts(mount.DefaultMounts).WithBind(root, "w", true).Build(true) mounts, err := mount.NewDefaultBuilder().WithBind(root, "w", true).Build(true)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot make rootfs mounts") return nil, fmt.Errorf("cannot make rootfs mounts")
} }

View File

@ -60,7 +60,7 @@ func (c *containerServer) handleReset() error {
func (c *containerServer) recvCmd() (*cmd, *unixsocket.Msg, error) { func (c *containerServer) recvCmd() (*cmd, *unixsocket.Msg, error) {
cm := new(cmd) cm := new(cmd)
msg, err := (*socket)(c.socket).RecvMsg(cm) msg, err := c.socket.RecvMsg(cm)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -68,7 +68,7 @@ func (c *containerServer) recvCmd() (*cmd, *unixsocket.Msg, error) {
} }
func (c *containerServer) sendReply(rep *reply, msg *unixsocket.Msg) error { func (c *containerServer) sendReply(rep *reply, msg *unixsocket.Msg) error {
return (*socket)(c.socket).SendMsg(rep, msg) return c.socket.SendMsg(rep, msg)
} }
// sendErrorReply sends error reply // sendErrorReply sends error reply

View File

@ -9,7 +9,7 @@ import (
) )
type containerServer struct { type containerServer struct {
socket *unixsocket.Socket socket *socket
containerConfig containerConfig
} }
@ -53,7 +53,7 @@ func Init() (err error) {
} }
// serve forever // serve forever
cs := &containerServer{socket: soc} cs := &containerServer{socket: newSocket(soc)}
return cs.serve() return cs.serve()
} }

View File

@ -55,9 +55,9 @@ type Environment interface {
// container manages single pre-forked container environment // container manages single pre-forked container environment
type container struct { type container struct {
pid int // underlying container init pid pid int // underlying container init pid
socket *unixsocket.Socket // host - container communication socket *socket // host - container communication
mu sync.Mutex // lock to avoid race condition mu sync.Mutex // lock to avoid race condition
} }
// Build creates new environment with underlying container // Build creates new environment with underlying container
@ -71,8 +71,7 @@ func (b *Builder) Build() (Environment, error) {
// container mount points // container mount points
mounts := b.Mounts mounts := b.Mounts
if len(mounts) == 0 { if len(mounts) == 0 {
if mounts, err = mount.NewBuilder(). if mounts, err = mount.NewDefaultBuilder().
WithMounts(mount.DefaultMounts).
WithTmpfs("w", ""). // work dir WithTmpfs("w", ""). // work dir
WithTmpfs("tmp", ""). // tmp WithTmpfs("tmp", ""). // tmp
Build(true); err != nil { Build(true); err != nil {
@ -154,7 +153,7 @@ func (b *Builder) Build() (Environment, error) {
c := &container{ c := &container{
pid: pid, pid: pid,
socket: ins, socket: newSocket(ins),
} }
// set configuration and check if container creation successful // set configuration and check if container creation successful

View File

@ -123,7 +123,7 @@ func (c *container) recvAckReply(name string) error {
func (c *container) recvReply() (*reply, *unixsocket.Msg, error) { func (c *container) recvReply() (*reply, *unixsocket.Msg, error) {
reply := new(reply) reply := new(reply)
msg, err := (*socket)(c.socket).RecvMsg(reply) msg, err := c.socket.RecvMsg(reply)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -131,5 +131,5 @@ func (c *container) recvReply() (*reply, *unixsocket.Msg, error) {
} }
func (c *container) sendCmd(cmd *cmd, msg *unixsocket.Msg) error { func (c *container) sendCmd(cmd *cmd, msg *unixsocket.Msg) error {
return (*socket)(c.socket).SendMsg(cmd, msg) return c.socket.SendMsg(cmd, msg)
} }

View File

@ -18,38 +18,50 @@ var bufferPool = sync.Pool{
}, },
} }
type socket unixsocket.Socket type socket struct {
*unixsocket.Socket
recvBuff bytes.Buffer
decoder *gob.Decoder
sendBuff bytes.Buffer
encoder *gob.Encoder
}
func newSocket(s *unixsocket.Socket) *socket {
soc := socket{
Socket: s,
}
soc.decoder = gob.NewDecoder(&soc.recvBuff)
soc.encoder = gob.NewEncoder(&soc.sendBuff)
return &soc
}
func (s *socket) RecvMsg(e interface{}) (*unixsocket.Msg, error) { func (s *socket) RecvMsg(e interface{}) (*unixsocket.Msg, error) {
soc := (*unixsocket.Socket)(s)
buff := bufferPool.Get().([]byte) buff := bufferPool.Get().([]byte)
defer bufferPool.Put(buff) defer bufferPool.Put(buff)
n, msg, err := soc.RecvMsg(buff) n, msg, err := s.Socket.RecvMsg(buff)
if err != nil { if err != nil {
return nil, fmt.Errorf("RecvMsg: %v", err) return nil, fmt.Errorf("RecvMsg: %v", err)
} }
s.recvBuff.Reset()
s.recvBuff.Write(buff[:n])
if err := gob.NewDecoder(bytes.NewReader(buff[:n])).Decode(e); err != nil { if err := s.decoder.Decode(e); err != nil {
return nil, fmt.Errorf("RecvMsg: failed to decode %v", err) return nil, fmt.Errorf("RecvMsg: failed to decode %v", err)
} }
return msg, nil return msg, nil
} }
func (s *socket) SendMsg(e interface{}, msg *unixsocket.Msg) error { func (s *socket) SendMsg(e interface{}, msg *unixsocket.Msg) error {
soc := (*unixsocket.Socket)(s) s.sendBuff.Reset()
if err := s.encoder.Encode(e); err != nil {
buf := bufferPool.Get().([]byte)
defer bufferPool.Put(buf)
// use buf pool to reduce allocation
buff := bytes.NewBuffer(buf[:0])
if err := gob.NewEncoder(buff).Encode(e); err != nil {
return fmt.Errorf("SendMsg: failed to encode %v", err) return fmt.Errorf("SendMsg: failed to encode %v", err)
} }
if err := soc.SendMsg(buff.Bytes(), msg); err != nil { if err := s.Socket.SendMsg(s.sendBuff.Bytes(), msg); err != nil {
return fmt.Errorf("SendMsg: failed to SendMsg %v", err) return fmt.Errorf("SendMsg: failed to SendMsg %v", err)
} }
return nil return nil

View File

@ -12,6 +12,15 @@ const (
mFlag = unix.MS_NOSUID | unix.MS_NOATIME | unix.MS_NODEV mFlag = unix.MS_NOSUID | unix.MS_NOATIME | unix.MS_NODEV
) )
// NewDefaultBuilder creates default builder for minimal rootfs
func NewDefaultBuilder() *Builder {
return NewBuilder().
WithBind("/usr", "usr", true).
WithBind("/lib", "lib", true).
WithBind("/lib64", "lib64", true).
WithBind("/bin", "bin", true)
}
// Build creates sequence of syscalls for fork_exec // Build creates sequence of syscalls for fork_exec
// skipNotExists skips bind mounts that source not exists // skipNotExists skips bind mounts that source not exists
func (b *Builder) Build(skipNotExists bool) ([]SyscallParams, error) { func (b *Builder) Build(skipNotExists bool) ([]SyscallParams, error) {

View File

@ -1,34 +0,0 @@
package mount
import "golang.org/x/sys/unix"
// RoBind is read-only bind mount flags
const (
RoBind = unix.MS_BIND | unix.MS_NOSUID | unix.MS_PRIVATE | unix.MS_RDONLY
)
var (
// DefaultMounts is a example of minimal rootfs
DefaultMounts = []Mount{
{
Source: "/usr",
Target: "usr",
Flags: RoBind,
},
{
Source: "/lib",
Target: "lib",
Flags: RoBind,
},
{
Source: "/lib64",
Target: "lib64",
Flags: RoBind,
},
{
Source: "/bin",
Target: "bin",
Flags: RoBind,
},
}
)

View File

@ -117,6 +117,9 @@ func (r *Runner) trace(c context.Context, runner *forkexec.Runner) (result types
case wstatus.Exited(): case wstatus.Exited():
result.Status = types.StatusNormal result.Status = types.StatusNormal
result.ExitStatus = wstatus.ExitStatus() result.ExitStatus = wstatus.ExitStatus()
if result.ExitStatus != 0 {
result.Status = types.StatusNonzeroExitStatus
}
return return
case wstatus.Signaled(): case wstatus.Signaled():