diff --git a/cmd/executorserver/cinterface.go b/cmd/executorserver/cinterface.go index 85e0dd7..6505c04 100644 --- a/cmd/executorserver/cinterface.go +++ b/cmd/executorserver/cinterface.go @@ -87,8 +87,8 @@ func Init(i *C.char) C.int { return -1 } - envPool = pool.NewEnvPool(b) - cgroupPool = pool.NewFakeCgroupPool(cgb) + cgroupPool := pool.NewFakeCgroupPool(cgb) + envPool = pool.NewEnvPool(b, cgroupPool) startWorkers() return 0 diff --git a/cmd/executorserver/main.go b/cmd/executorserver/main.go index 972bf53..9ed8085 100644 --- a/cmd/executorserver/main.go +++ b/cmd/executorserver/main.go @@ -27,8 +27,7 @@ var ( netShare = flag.Bool("net", false, "do not unshare net namespace with host") mountConf = flag.String("mount", "mount.yaml", "specifics mount configuration file") - envPool envexec.EnvironmentPool - cgroupPool envexec.CgroupPool + envPool envexec.EnvironmentPool fs fileStore @@ -90,8 +89,8 @@ func main() { } 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) startWorkers() diff --git a/cmd/executorserver/waiter.go b/cmd/executorserver/waiter.go index 384c9ed..53395fe 100644 --- a/cmd/executorserver/waiter.go +++ b/cmd/executorserver/waiter.go @@ -12,7 +12,7 @@ type waiter struct { 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 { w.realTimeLimit = w.timeLimit } diff --git a/cmd/executorserver/worker.go b/cmd/executorserver/worker.go index bdc38e4..78cc24a 100644 --- a/cmd/executorserver/worker.go +++ b/cmd/executorserver/worker.go @@ -72,7 +72,6 @@ func workDoSingle(rc cmd) (rt result) { return } s := &envexec.Single{ - CgroupPool: cgroupPool, EnvironmentPool: envPool, Cmd: c, } @@ -101,7 +100,6 @@ func workDoGroup(rc []cmd, pm []pipeMap) (rt result) { copyOutSets = append(copyOutSets, os) } g := envexec.Group{ - CgroupPool: cgroupPool, EnvironmentPool: envPool, Cmd: cs, diff --git a/pkg/envexec/cmd.go b/pkg/envexec/cmd.go index 54cf178..d1e1be1 100644 --- a/pkg/envexec/cmd.go +++ b/pkg/envexec/cmd.go @@ -21,7 +21,8 @@ type Cmd struct { // PipeCollector: a pipe write end will be passed to runner and collected as a copyout file Files []interface{} - // cgroup limits + // resource limits + TimeLimit time.Duration MemoryLimit runner.Size // in bytes ProcLimit uint64 @@ -37,7 +38,7 @@ type Cmd struct { // Waiter is called after cmd starts and it should return // once time limit exceeded. // 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 diff --git a/pkg/envexec/file_collect.go b/pkg/envexec/file_collect.go index 3bbadb2..e9a41dc 100644 --- a/pkg/envexec/file_collect.go +++ b/pkg/envexec/file_collect.go @@ -19,7 +19,7 @@ func copyOutAndCollect(m Environment, c *Cmd, ptc []pipeCollector) (map[string]f for _, n := range c.CopyOut { n := n 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 { return err } diff --git a/pkg/envexec/file_copyin.go b/pkg/envexec/file_copyin.go index 8a89269..ef24f44 100644 --- a/pkg/envexec/file_copyin.go +++ b/pkg/envexec/file_copyin.go @@ -14,7 +14,7 @@ func copyIn(m Environment, copyIn map[string]file.File) error { for n, f := range copyIn { n, f := n, f 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 { return err } diff --git a/pkg/envexec/group.go b/pkg/envexec/group.go index 5e6ddd1..3c3c7c2 100644 --- a/pkg/envexec/group.go +++ b/pkg/envexec/group.go @@ -9,9 +9,6 @@ import ( // Group defines the running instruction to run multiple // exec in parallel restricted within cgroup type Group struct { - // CgroupPool defines pool of cgroup used for Cmd - CgroupPool CgroupPool - // EnvironmentPool defines pool used for runner environment EnvironmentPool EnvironmentPool @@ -55,24 +52,13 @@ func (r *Group) Run() ([]Result, error) { 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 var g errgroup.Group result := make([]Result, len(r.Cmd)) for i, c := range r.Cmd { i, c := i, c 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 if err != nil { result[i].Status = StatusInternalError diff --git a/pkg/envexec/interface.go b/pkg/envexec/interface.go index afc5482..3c30d00 100644 --- a/pkg/envexec/interface.go +++ b/pkg/envexec/interface.go @@ -1,22 +1,58 @@ package envexec import ( + "context" "os" "time" - "github.com/criyle/go-sandbox/container" "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 type Environment interface { - container.Environment - - // WorkDir returns opened work directory, should not close after - WorkDir() *os.File - - // OpenAtWorkDir open file at work dir with given relative path and flags - OpenAtWorkDir(path string, flags int, perm os.FileMode) (*os.File, error) + Execve(context.Context, ExecveParam) (Process, error) + WorkDir() *os.File // WorkDir returns opened work directory, should not close after + // Open open file at work dir with given relative path and flags + Open(path string, flags int, perm os.FileMode) (*os.File, error) } // EnvironmentPool implements pool of environments @@ -24,27 +60,3 @@ type EnvironmentPool interface { Get() (Environment, error) 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) -} diff --git a/pkg/envexec/run_single.go b/pkg/envexec/run_single.go index 93f45fd..a9f0ca5 100644 --- a/pkg/envexec/run_single.go +++ b/pkg/envexec/run_single.go @@ -4,23 +4,14 @@ import ( "context" "os" - "github.com/criyle/go-sandbox/container" "github.com/criyle/go-sandbox/runner" ) // 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 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 if len(c.CopyIn) > 0 { 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 - execParam := container.ExecveParam{ - Args: c.Args, - Env: c.Env, - Files: getFdArray(fds), - SyncFunc: cg.AddProc, + execParam := ExecveParam{ + Args: c.Args, + Env: c.Env, + Files: getFdArray(fds), + Limit: Limit{ + Time: c.TimeLimit, + Memory: c.MemoryLimit + runner.Size(memoryLimitExtra), + Proc: c.ProcLimit, + }, } // start the cmd (they will be canceled in other goroutines) ctx, cancel := context.WithCancel(context.TODO()) waiterCtx, waiterCancel := context.WithCancel(ctx) - rc := m.Execve(ctx, execParam) + process, err := m.Execve(ctx, execParam) // close files 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 go func() { - c.Waiter(waiterCtx, cg) + c.Waiter(waiterCtx, process) 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() // collect result @@ -75,22 +80,6 @@ func runSingle(m Environment, cg Cgroup, c *Cmd, fds []*os.File, ptc []pipeColle } 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 { result.Status = StatusMemoryLimitExceeded } diff --git a/pkg/envexec/single.go b/pkg/envexec/single.go index ddf6413..d4a631c 100644 --- a/pkg/envexec/single.go +++ b/pkg/envexec/single.go @@ -7,9 +7,6 @@ import ( // Single defines the running instruction to run single // exec in restricted within cgroup type Single struct { - // CgroupPool defines pool of cgroup used for Cmd - CgroupPool CgroupPool - // EnvironmentPool defines pool used for runner environment EnvironmentPool EnvironmentPool @@ -34,14 +31,7 @@ func (s *Single) Run() (result Result, err error) { } defer s.EnvironmentPool.Put(m) - // prepare cgroup - 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) + result, err = runSingle(m, s.Cmd, fd, pipeToCollect) fileToClose = nil // already closed by runOne if err != nil { result.Status = StatusInternalError diff --git a/pkg/pool/cgroup_fake_pool.go b/pkg/pool/cgroup_fake_pool.go index 4a07d4f..b5f2982 100644 --- a/pkg/pool/cgroup_fake_pool.go +++ b/pkg/pool/cgroup_fake_pool.go @@ -1,8 +1,6 @@ package pool -import "github.com/criyle/go-judge/pkg/envexec" - -var _ envexec.CgroupPool = &FakeCgroupPool{} +var _ CgroupPool = &FakeCgroupPool{} // FakeCgroupPool implements cgroup pool but not actually do pool type FakeCgroupPool struct { @@ -10,12 +8,12 @@ type FakeCgroupPool struct { } // NewFakeCgroupPool creates FakeCgroupPool -func NewFakeCgroupPool(builder CgroupBuilder) *FakeCgroupPool { +func NewFakeCgroupPool(builder CgroupBuilder) CgroupPool { return &FakeCgroupPool{builder: builder} } // Get gets new cgroup -func (f *FakeCgroupPool) Get() (envexec.Cgroup, error) { +func (f *FakeCgroupPool) Get() (Cgroup, error) { cg, err := f.builder.Build() if err != nil { return nil, err @@ -24,7 +22,7 @@ func (f *FakeCgroupPool) Get() (envexec.Cgroup, error) { } // Put destory the cgroup -func (f *FakeCgroupPool) Put(c envexec.Cgroup) { +func (f *FakeCgroupPool) Put(c Cgroup) { c.Destroy() } diff --git a/pkg/pool/cgroup_wrapper.go b/pkg/pool/cgroup_wrapper.go index b8abafa..066acb6 100644 --- a/pkg/pool/cgroup_wrapper.go +++ b/pkg/pool/cgroup_wrapper.go @@ -3,13 +3,12 @@ package pool import ( "time" - "github.com/criyle/go-judge/pkg/envexec" "github.com/criyle/go-sandbox/pkg/cgroup" "github.com/criyle/go-sandbox/runner" ) var ( - _ envexec.Cgroup = &wCgroup{} + _ Cgroup = &wCgroup{} ) type wCgroup cgroup.Cgroup diff --git a/pkg/pool/cgrouppool.go b/pkg/pool/cgrouppool.go index 7c051bd..065e731 100644 --- a/pkg/pool/cgrouppool.go +++ b/pkg/pool/cgrouppool.go @@ -2,25 +2,45 @@ package pool import ( "sync" + "time" - "github.com/criyle/go-judge/pkg/envexec" + "github.com/criyle/go-sandbox/runner" ) -// CgroupPool implements cgroup pool -type CgroupPool struct { +// 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 +} + +// CgroupPool implements pool of Cgroup +type CgroupPool interface { + Get() (Cgroup, error) + Put(Cgroup) +} + +// CgroupListPool implements cgroup pool +type CgroupListPool struct { builder CgroupBuilder - cgs []envexec.Cgroup + cgs []Cgroup mu sync.Mutex } -// NewCgroupPool creates new cgroup pool -func NewCgroupPool(builder CgroupBuilder) *CgroupPool { - return &CgroupPool{builder: builder} +// NewCgroupListPool creates new cgroup pool +func NewCgroupListPool(builder CgroupBuilder) CgroupPool { + return &CgroupListPool{builder: builder} } // 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() defer w.mu.Unlock() @@ -38,7 +58,7 @@ func (w *CgroupPool) Get() (envexec.Cgroup, error) { } // Put puts cgroup into the pool -func (w *CgroupPool) Put(c envexec.Cgroup) { +func (w *CgroupListPool) Put(c Cgroup) { w.mu.Lock() defer w.mu.Unlock() @@ -47,7 +67,7 @@ func (w *CgroupPool) Put(c envexec.Cgroup) { } // Shutdown destroy all cgroup -func (w *CgroupPool) Shutdown() { +func (w *CgroupListPool) Shutdown() { w.mu.Lock() defer w.mu.Unlock() diff --git a/pkg/pool/environment.go b/pkg/pool/environment.go new file mode 100644 index 0000000..a767714 --- /dev/null +++ b/pkg/pool/environment.go @@ -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 +} diff --git a/pkg/pool/envpool.go b/pkg/pool/envpool.go index f40b561..b984104 100644 --- a/pkg/pool/envpool.go +++ b/pkg/pool/envpool.go @@ -2,7 +2,6 @@ package pool import ( "fmt" - "os" "sync" "syscall" @@ -10,24 +9,20 @@ import ( "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 type EnvPool struct { builder EnvironmentBuilder + cgPool CgroupPool - env []envexec.Environment + env []*Environment mu sync.Mutex } // NewEnvPool creats new EnvPool with a builder -func NewEnvPool(builder EnvironmentBuilder) *EnvPool { +func NewEnvPool(builder EnvironmentBuilder, cgPool CgroupPool) *EnvPool { return &EnvPool{ builder: builder, + cgPool: cgPool, } } @@ -53,11 +48,16 @@ func (p *EnvPool) Get() (envexec.Environment, error) { if err != nil { 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 -func (p *EnvPool) Put(env envexec.Environment) { +func (p *EnvPool) Put(e envexec.Environment) { + env, _ := e.(*Environment) env.Reset() p.mu.Lock() @@ -67,7 +67,8 @@ func (p *EnvPool) Put(env envexec.Environment) { } // Destroy destory an environment -func (p *EnvPool) Destroy(env container.Environment) { +func (p *EnvPool) Destroy(e envexec.Environment) { + env, _ := e.(*Environment) env.Destroy() } @@ -90,22 +91,3 @@ func (p *EnvPool) Shutdown() { 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 -} diff --git a/pkg/pool/envprocess.go b/pkg/pool/envprocess.go new file mode 100644 index 0000000..ee02eea --- /dev/null +++ b/pkg/pool/envprocess.go @@ -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, + } +}