simplifies environment interface

This commit is contained in:
criyle 2020-04-16 20:15:59 -04:00
parent 008f9dfa41
commit a2a6c0f97b
17 changed files with 244 additions and 154 deletions

View File

@ -87,8 +87,8 @@ func Init(i *C.char) C.int {
return -1 return -1
} }
envPool = pool.NewEnvPool(b) cgroupPool := pool.NewFakeCgroupPool(cgb)
cgroupPool = pool.NewFakeCgroupPool(cgb) envPool = pool.NewEnvPool(b, cgroupPool)
startWorkers() startWorkers()
return 0 return 0

View File

@ -28,7 +28,6 @@ var (
mountConf = flag.String("mount", "mount.yaml", "specifics mount configuration file") mountConf = flag.String("mount", "mount.yaml", "specifics mount configuration file")
envPool envexec.EnvironmentPool envPool envexec.EnvironmentPool
cgroupPool envexec.CgroupPool
fs fileStore fs fileStore
@ -90,8 +89,8 @@ func main() {
} }
printLog("Created cgroup builder with:", cgb) printLog("Created cgroup builder with:", cgb)
envPool = pool.NewEnvPool(b) cgroupPool := pool.NewFakeCgroupPool(cgb)
cgroupPool = pool.NewFakeCgroupPool(cgb) envPool = pool.NewEnvPool(b, cgroupPool)
printLog("Starting worker with parallism", *parallism) printLog("Starting worker with parallism", *parallism)
startWorkers() startWorkers()

View File

@ -12,7 +12,7 @@ type waiter struct {
realTimeLimit time.Duration realTimeLimit time.Duration
} }
func (w *waiter) Wait(ctx context.Context, u envexec.CPUUsager) bool { func (w *waiter) Wait(ctx context.Context, u envexec.Process) bool {
if w.realTimeLimit < w.timeLimit { if w.realTimeLimit < w.timeLimit {
w.realTimeLimit = w.timeLimit w.realTimeLimit = w.timeLimit
} }

View File

@ -72,7 +72,6 @@ func workDoSingle(rc cmd) (rt result) {
return return
} }
s := &envexec.Single{ s := &envexec.Single{
CgroupPool: cgroupPool,
EnvironmentPool: envPool, EnvironmentPool: envPool,
Cmd: c, Cmd: c,
} }
@ -101,7 +100,6 @@ func workDoGroup(rc []cmd, pm []pipeMap) (rt result) {
copyOutSets = append(copyOutSets, os) copyOutSets = append(copyOutSets, os)
} }
g := envexec.Group{ g := envexec.Group{
CgroupPool: cgroupPool,
EnvironmentPool: envPool, EnvironmentPool: envPool,
Cmd: cs, Cmd: cs,

View File

@ -21,7 +21,8 @@ type Cmd struct {
// PipeCollector: a pipe write end will be passed to runner and collected as a copyout file // PipeCollector: a pipe write end will be passed to runner and collected as a copyout file
Files []interface{} Files []interface{}
// cgroup limits // resource limits
TimeLimit time.Duration
MemoryLimit runner.Size // in bytes MemoryLimit runner.Size // in bytes
ProcLimit uint64 ProcLimit uint64
@ -37,7 +38,7 @@ type Cmd struct {
// Waiter is called after cmd starts and it should return // Waiter is called after cmd starts and it should return
// once time limit exceeded. // once time limit exceeded.
// return true to as TLE and false as normal exits (context finished) // return true to as TLE and false as normal exits (context finished)
Waiter func(context.Context, CPUUsager) bool Waiter func(context.Context, Process) bool
} }
// PipeCollector can be used in Cmd.Files paramenter // PipeCollector can be used in Cmd.Files paramenter

View File

@ -19,7 +19,7 @@ func copyOutAndCollect(m Environment, c *Cmd, ptc []pipeCollector) (map[string]f
for _, n := range c.CopyOut { for _, n := range c.CopyOut {
n := n n := n
g.Go(func() error { g.Go(func() error {
cf, err := m.OpenAtWorkDir(n, os.O_RDONLY, 0777) cf, err := m.Open(n, os.O_RDONLY, 0777)
if err != nil { if err != nil {
return err return err
} }

View File

@ -14,7 +14,7 @@ func copyIn(m Environment, copyIn map[string]file.File) error {
for n, f := range copyIn { for n, f := range copyIn {
n, f := n, f n, f := n, f
g.Go(func() error { g.Go(func() error {
cf, err := m.OpenAtWorkDir(n, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0777) cf, err := m.Open(n, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0777)
if err != nil { if err != nil {
return err return err
} }

View File

@ -9,9 +9,6 @@ import (
// Group defines the running instruction to run multiple // Group defines the running instruction to run multiple
// exec in parallel restricted within cgroup // exec in parallel restricted within cgroup
type Group struct { type Group struct {
// CgroupPool defines pool of cgroup used for Cmd
CgroupPool CgroupPool
// EnvironmentPool defines pool used for runner environment // EnvironmentPool defines pool used for runner environment
EnvironmentPool EnvironmentPool EnvironmentPool EnvironmentPool
@ -55,24 +52,13 @@ func (r *Group) Run() ([]Result, error) {
ms = append(ms, m) ms = append(ms, m)
} }
// prepare cgroup
cgs := make([]Cgroup, 0, len(r.Cmd))
for range r.Cmd {
cg, err := r.CgroupPool.Get()
if err != nil {
return nil, fmt.Errorf("failed to get cgroup %v", err)
}
defer r.CgroupPool.Put(cg)
cgs = append(cgs, cg)
}
// wait all cmd to finish // wait all cmd to finish
var g errgroup.Group var g errgroup.Group
result := make([]Result, len(r.Cmd)) result := make([]Result, len(r.Cmd))
for i, c := range r.Cmd { for i, c := range r.Cmd {
i, c := i, c i, c := i, c
g.Go(func() error { g.Go(func() error {
r, err := runSingle(ms[i], cgs[i], c, fds[i], pipeToCollect[i]) r, err := runSingle(ms[i], c, fds[i], pipeToCollect[i])
result[i] = r result[i] = r
if err != nil { if err != nil {
result[i].Status = StatusInternalError result[i].Status = StatusInternalError

View File

@ -1,22 +1,58 @@
package envexec package envexec
import ( import (
"context"
"os" "os"
"time" "time"
"github.com/criyle/go-sandbox/container"
"github.com/criyle/go-sandbox/runner" "github.com/criyle/go-sandbox/runner"
) )
// ExecveParam is parameters to run process inside environment
type ExecveParam struct {
// Args holds command line arguments
Args []string
// Env specifies the environment of the process
Env []string
// Files specifies file descriptors for the child process
Files []uintptr
// ExecFile specifies file descriptor for executable file using fexecve
ExecFile uintptr
// Process Limitations
Limit Limit
}
// Limit defines the process running resource limits
type Limit struct {
Time time.Duration // Time limit
Memory runner.Size // Memory limit
Proc uint64 // Process count limit
Stack runner.Size // Stack limit
}
// Usage defines the peak process resource usage
type Usage struct {
Time time.Duration
Memory runner.Size
}
// Process reference to the running process group
type Process interface {
Done() <-chan struct{} // Done returns a channel for wait process to exit
Result() runner.Result // Result is avaliable after done is closed
Usage() Usage // Usage retrieves the process usage during the run time
}
// Environment defines the interface to access container execution environment // Environment defines the interface to access container execution environment
type Environment interface { type Environment interface {
container.Environment Execve(context.Context, ExecveParam) (Process, error)
WorkDir() *os.File // WorkDir returns opened work directory, should not close after
// WorkDir returns opened work directory, should not close after // Open open file at work dir with given relative path and flags
WorkDir() *os.File Open(path string, flags int, perm os.FileMode) (*os.File, error)
// OpenAtWorkDir open file at work dir with given relative path and flags
OpenAtWorkDir(path string, flags int, perm os.FileMode) (*os.File, error)
} }
// EnvironmentPool implements pool of environments // EnvironmentPool implements pool of environments
@ -24,27 +60,3 @@ type EnvironmentPool interface {
Get() (Environment, error) Get() (Environment, error)
Put(Environment) Put(Environment)
} }
// Cgroup defines interface to limit and monitor resources consumption of a process
type Cgroup interface {
SetMemoryLimit(runner.Size) error
SetProcLimit(uint64) error
CPUUsage() (time.Duration, error)
MemoryUsage() (runner.Size, error)
AddProc(int) error
Reset() error
Destroy() error
}
// CPUUsager access process cpu usage (from cgroup)
type CPUUsager interface {
CPUUsage() (time.Duration, error)
}
// CgroupPool implements pool of Cgroup
type CgroupPool interface {
Get() (Cgroup, error)
Put(Cgroup)
}

View File

@ -4,23 +4,14 @@ import (
"context" "context"
"os" "os"
"github.com/criyle/go-sandbox/container"
"github.com/criyle/go-sandbox/runner" "github.com/criyle/go-sandbox/runner"
) )
// runSingle runs Cmd inside the given environment and cgroup // runSingle runs Cmd inside the given environment and cgroup
func runSingle(m Environment, cg Cgroup, c *Cmd, fds []*os.File, ptc []pipeCollector) (result Result, err error) { func runSingle(m Environment, c *Cmd, fds []*os.File, ptc []pipeCollector) (result Result, err error) {
fdToClose := fds fdToClose := fds
defer func() { closeFiles(fdToClose) }() defer func() { closeFiles(fdToClose) }()
// setup cgroup limits
if err := cg.SetMemoryLimit(runner.Size(uint64(c.MemoryLimit) + memoryLimitExtra)); err != nil {
return result, err
}
if err := cg.SetProcLimit(c.ProcLimit); err != nil {
return result, err
}
// copyin // copyin
if len(c.CopyIn) > 0 { if len(c.CopyIn) > 0 {
if err := copyIn(m, c.CopyIn); err != nil { if err := copyIn(m, c.CopyIn); err != nil {
@ -29,18 +20,22 @@ func runSingle(m Environment, cg Cgroup, c *Cmd, fds []*os.File, ptc []pipeColle
} }
// set running parameters // set running parameters
execParam := container.ExecveParam{ execParam := ExecveParam{
Args: c.Args, Args: c.Args,
Env: c.Env, Env: c.Env,
Files: getFdArray(fds), Files: getFdArray(fds),
SyncFunc: cg.AddProc, Limit: Limit{
Time: c.TimeLimit,
Memory: c.MemoryLimit + runner.Size(memoryLimitExtra),
Proc: c.ProcLimit,
},
} }
// start the cmd (they will be canceled in other goroutines) // start the cmd (they will be canceled in other goroutines)
ctx, cancel := context.WithCancel(context.TODO()) ctx, cancel := context.WithCancel(context.TODO())
waiterCtx, waiterCancel := context.WithCancel(ctx) waiterCtx, waiterCancel := context.WithCancel(ctx)
rc := m.Execve(ctx, execParam) process, err := m.Execve(ctx, execParam)
// close files // close files
closeFiles(fds) closeFiles(fds)
@ -48,11 +43,21 @@ func runSingle(m Environment, cg Cgroup, c *Cmd, fds []*os.File, ptc []pipeColle
// starts waiter to periodically check cpu usage // starts waiter to periodically check cpu usage
go func() { go func() {
c.Waiter(waiterCtx, cg) c.Waiter(waiterCtx, process)
cancel() cancel()
}() }()
rt := <-rc var rt runner.Result
if err == nil {
<-process.Done()
rt = process.Result()
} else {
rt = runner.Result{
Status: runner.StatusRunnerError,
Error: err.Error(),
}
}
waiterCancel() waiterCancel()
// collect result // collect result
@ -75,22 +80,6 @@ func runSingle(m Environment, cg Cgroup, c *Cmd, fds []*os.File, ptc []pipeColle
} }
result.Error = err.Error() result.Error = err.Error()
} }
// time
cpuUsage, err := cg.CPUUsage()
if err != nil {
result.Status = StatusInternalError
result.Error = err.Error()
} else {
result.Time = cpuUsage
}
// memory
memoryUsage, err := cg.MemoryUsage()
if err != nil {
result.Status = StatusInternalError
result.Error = err.Error()
} else {
result.Memory = memoryUsage
}
if result.Memory > c.MemoryLimit { if result.Memory > c.MemoryLimit {
result.Status = StatusMemoryLimitExceeded result.Status = StatusMemoryLimitExceeded
} }

View File

@ -7,9 +7,6 @@ import (
// Single defines the running instruction to run single // Single defines the running instruction to run single
// exec in restricted within cgroup // exec in restricted within cgroup
type Single struct { type Single struct {
// CgroupPool defines pool of cgroup used for Cmd
CgroupPool CgroupPool
// EnvironmentPool defines pool used for runner environment // EnvironmentPool defines pool used for runner environment
EnvironmentPool EnvironmentPool EnvironmentPool EnvironmentPool
@ -34,14 +31,7 @@ func (s *Single) Run() (result Result, err error) {
} }
defer s.EnvironmentPool.Put(m) defer s.EnvironmentPool.Put(m)
// prepare cgroup result, err = runSingle(m, s.Cmd, fd, pipeToCollect)
cg, err := s.CgroupPool.Get()
if err != nil {
return result, fmt.Errorf("failed to get cgroup %v", err)
}
defer s.CgroupPool.Put(cg)
result, err = runSingle(m, cg, s.Cmd, fd, pipeToCollect)
fileToClose = nil // already closed by runOne fileToClose = nil // already closed by runOne
if err != nil { if err != nil {
result.Status = StatusInternalError result.Status = StatusInternalError

View File

@ -1,8 +1,6 @@
package pool package pool
import "github.com/criyle/go-judge/pkg/envexec" var _ CgroupPool = &FakeCgroupPool{}
var _ envexec.CgroupPool = &FakeCgroupPool{}
// FakeCgroupPool implements cgroup pool but not actually do pool // FakeCgroupPool implements cgroup pool but not actually do pool
type FakeCgroupPool struct { type FakeCgroupPool struct {
@ -10,12 +8,12 @@ type FakeCgroupPool struct {
} }
// NewFakeCgroupPool creates FakeCgroupPool // NewFakeCgroupPool creates FakeCgroupPool
func NewFakeCgroupPool(builder CgroupBuilder) *FakeCgroupPool { func NewFakeCgroupPool(builder CgroupBuilder) CgroupPool {
return &FakeCgroupPool{builder: builder} return &FakeCgroupPool{builder: builder}
} }
// Get gets new cgroup // Get gets new cgroup
func (f *FakeCgroupPool) Get() (envexec.Cgroup, error) { func (f *FakeCgroupPool) Get() (Cgroup, error) {
cg, err := f.builder.Build() cg, err := f.builder.Build()
if err != nil { if err != nil {
return nil, err return nil, err
@ -24,7 +22,7 @@ func (f *FakeCgroupPool) Get() (envexec.Cgroup, error) {
} }
// Put destory the cgroup // Put destory the cgroup
func (f *FakeCgroupPool) Put(c envexec.Cgroup) { func (f *FakeCgroupPool) Put(c Cgroup) {
c.Destroy() c.Destroy()
} }

View File

@ -3,13 +3,12 @@ package pool
import ( import (
"time" "time"
"github.com/criyle/go-judge/pkg/envexec"
"github.com/criyle/go-sandbox/pkg/cgroup" "github.com/criyle/go-sandbox/pkg/cgroup"
"github.com/criyle/go-sandbox/runner" "github.com/criyle/go-sandbox/runner"
) )
var ( var (
_ envexec.Cgroup = &wCgroup{} _ Cgroup = &wCgroup{}
) )
type wCgroup cgroup.Cgroup type wCgroup cgroup.Cgroup

View File

@ -2,25 +2,45 @@ package pool
import ( import (
"sync" "sync"
"time"
"github.com/criyle/go-judge/pkg/envexec" "github.com/criyle/go-sandbox/runner"
) )
// CgroupPool implements cgroup pool // Cgroup defines interface to limit and monitor resources consumption of a process
type CgroupPool struct { type Cgroup interface {
SetMemoryLimit(runner.Size) error
SetProcLimit(uint64) error
CPUUsage() (time.Duration, error)
MemoryUsage() (runner.Size, error)
AddProc(int) error
Reset() error
Destroy() error
}
// CgroupPool implements pool of Cgroup
type CgroupPool interface {
Get() (Cgroup, error)
Put(Cgroup)
}
// CgroupListPool implements cgroup pool
type CgroupListPool struct {
builder CgroupBuilder builder CgroupBuilder
cgs []envexec.Cgroup cgs []Cgroup
mu sync.Mutex mu sync.Mutex
} }
// NewCgroupPool creates new cgroup pool // NewCgroupListPool creates new cgroup pool
func NewCgroupPool(builder CgroupBuilder) *CgroupPool { func NewCgroupListPool(builder CgroupBuilder) CgroupPool {
return &CgroupPool{builder: builder} return &CgroupListPool{builder: builder}
} }
// Get gets cgroup from pool, if pool is empty, creates new one // Get gets cgroup from pool, if pool is empty, creates new one
func (w *CgroupPool) Get() (envexec.Cgroup, error) { func (w *CgroupListPool) Get() (Cgroup, error) {
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()
@ -38,7 +58,7 @@ func (w *CgroupPool) Get() (envexec.Cgroup, error) {
} }
// Put puts cgroup into the pool // Put puts cgroup into the pool
func (w *CgroupPool) Put(c envexec.Cgroup) { func (w *CgroupListPool) Put(c Cgroup) {
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()
@ -47,7 +67,7 @@ func (w *CgroupPool) Put(c envexec.Cgroup) {
} }
// Shutdown destroy all cgroup // Shutdown destroy all cgroup
func (w *CgroupPool) Shutdown() { func (w *CgroupListPool) Shutdown() {
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()

64
pkg/pool/environment.go Normal file
View File

@ -0,0 +1,64 @@
package pool
import (
"context"
"fmt"
"os"
"syscall"
"github.com/criyle/go-judge/pkg/envexec"
"github.com/criyle/go-sandbox/container"
)
var _ envexec.Environment = &Environment{}
// Environment defines interface to access container resources
type Environment struct {
container.Environment
cgPool CgroupPool
wd *os.File // container work dir
}
// Destory destories the environment
func (c *Environment) Destory() error {
return c.Environment.Destroy()
}
// Execve execute process inside the environment
func (c *Environment) Execve(ctx context.Context, param envexec.ExecveParam) (envexec.Process, error) {
cg, err := c.cgPool.Get()
if err != nil {
return nil, fmt.Errorf("execve: failed to get cgroup %v", err)
}
cg.SetMemoryLimit(param.Limit.Memory)
cg.SetProcLimit(param.Limit.Proc)
p := container.ExecveParam{
Args: param.Args,
Env: param.Env,
Files: param.Files,
ExecFile: param.ExecFile,
SyncFunc: cg.AddProc,
}
rt := c.Environment.Execve(ctx, p)
return newProcess(rt, cg, c.cgPool), nil
}
// WorkDir returns opened work directory, should not close after
func (c *Environment) WorkDir() *os.File {
c.wd.Seek(0, 0)
return c.wd
}
// Open opens file relative to work directory
func (c *Environment) Open(path string, flags int, perm os.FileMode) (*os.File, error) {
fd, err := syscall.Openat(int(c.wd.Fd()), path, flags|syscall.O_CLOEXEC, uint32(perm))
if err != nil {
return nil, fmt.Errorf("openAtWorkDir: %v", err)
}
f := os.NewFile(uintptr(fd), path)
if f == nil {
return nil, fmt.Errorf("openAtWorkDir: failed to NewFile")
}
return f, nil
}

View File

@ -2,7 +2,6 @@ package pool
import ( import (
"fmt" "fmt"
"os"
"sync" "sync"
"syscall" "syscall"
@ -10,24 +9,20 @@ import (
"github.com/criyle/go-sandbox/container" "github.com/criyle/go-sandbox/container"
) )
// Environment defines interface to access container resources
type Environment struct {
container.Environment
wd *os.File // container work dir
}
// EnvPool implements container environment pool // EnvPool implements container environment pool
type EnvPool struct { type EnvPool struct {
builder EnvironmentBuilder builder EnvironmentBuilder
cgPool CgroupPool
env []envexec.Environment env []*Environment
mu sync.Mutex mu sync.Mutex
} }
// NewEnvPool creats new EnvPool with a builder // NewEnvPool creats new EnvPool with a builder
func NewEnvPool(builder EnvironmentBuilder) *EnvPool { func NewEnvPool(builder EnvironmentBuilder, cgPool CgroupPool) *EnvPool {
return &EnvPool{ return &EnvPool{
builder: builder, builder: builder,
cgPool: cgPool,
} }
} }
@ -53,11 +48,16 @@ func (p *EnvPool) Get() (envexec.Environment, error) {
if err != nil { if err != nil {
return nil, fmt.Errorf("container: failed to prepare work directory") return nil, fmt.Errorf("container: failed to prepare work directory")
} }
return &Environment{Environment: m, wd: wd[0]}, nil return &Environment{
Environment: m,
cgPool: p.cgPool,
wd: wd[0],
}, nil
} }
// Put puts environment to the pool with reset the environment // Put puts environment to the pool with reset the environment
func (p *EnvPool) Put(env envexec.Environment) { func (p *EnvPool) Put(e envexec.Environment) {
env, _ := e.(*Environment)
env.Reset() env.Reset()
p.mu.Lock() p.mu.Lock()
@ -67,7 +67,8 @@ func (p *EnvPool) Put(env envexec.Environment) {
} }
// Destroy destory an environment // Destroy destory an environment
func (p *EnvPool) Destroy(env container.Environment) { func (p *EnvPool) Destroy(e envexec.Environment) {
env, _ := e.(*Environment)
env.Destroy() env.Destroy()
} }
@ -90,22 +91,3 @@ func (p *EnvPool) Shutdown() {
p.Destroy(e) p.Destroy(e)
} }
} }
// WorkDir returns opened work directory, should not close after
func (c *Environment) WorkDir() *os.File {
c.wd.Seek(0, 0)
return c.wd
}
// OpenAtWorkDir opens file relative to work directory
func (c *Environment) OpenAtWorkDir(path string, flags int, perm os.FileMode) (*os.File, error) {
fd, err := syscall.Openat(int(c.wd.Fd()), path, flags|syscall.O_CLOEXEC, uint32(perm))
if err != nil {
return nil, fmt.Errorf("openAtWorkDir: %v", err)
}
f := os.NewFile(uintptr(fd), path)
if f == nil {
return nil, fmt.Errorf("openAtWorkDir: failed to NewFile")
}
return f, nil
}

52
pkg/pool/envprocess.go Normal file
View File

@ -0,0 +1,52 @@
package pool
import (
"github.com/criyle/go-judge/pkg/envexec"
"github.com/criyle/go-sandbox/runner"
)
var _ envexec.Process = &process{}
// process defines the running process
type process struct {
rt runner.Result
done chan struct{}
cg Cgroup
}
func newProcess(ch <-chan runner.Result, cg Cgroup, cgPool CgroupPool) *process {
p := &process{
done: make(chan struct{}),
cg: cg,
}
go func() {
defer close(p.done)
defer cgPool.Put(cg)
p.rt = <-ch
if t, err := cg.CPUUsage(); err == nil {
p.rt.Time = t
}
if m, err := cg.MemoryUsage(); err == nil {
p.rt.Memory = m
}
}()
return p
}
func (p *process) Done() <-chan struct{} {
return p.done
}
func (p *process) Result() runner.Result {
<-p.done
return p.rt
}
func (p *process) Usage() envexec.Usage {
t, _ := p.cg.CPUUsage()
m, _ := p.cg.MemoryUsage()
return envexec.Usage{
Time: t,
Memory: m,
}
}