From 8e8cc171231c3f0f213768b08698c6c0ed58ad2c Mon Sep 17 00:00:00 2001 From: criyle Date: Wed, 4 Mar 2020 17:27:27 -0500 Subject: [PATCH] Implements multiple command & update example --- README.md | 93 ++++++++++++++++++--- cmd/executorserver/worker.go | 153 ++++++++++++++++++++++++++--------- 2 files changed, 196 insertions(+), 50 deletions(-) diff --git a/README.md b/README.md index d23d8db..3ee3f12 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,8 @@ interface Result { Example Request & Response: +Single: + ```json { "cmd": [{ @@ -146,21 +148,90 @@ Example Request & Response: ``` ```json -{ - "status": "Accepted", - "time": 324879037, - "memory": 32378880, - "files": { - "stderr": "", - "stdout": "" - }, - "fileIds": { - "a": "Yj6uHXVnTrBtUNeq", - "a.cc": "uYJTBTdtKWXaP4xE" +[ + { + "status": "Accepted", + "time": 303225231, + "memory": 32243712, + "files": { + "stderr": "", + "stdout": "" + }, + "fileIds": { + "a": "5LWIZAA45JHX4Y4Z", + "a.cc": "NOHPGGDTYQUFRSLJ" + } } +] +``` + +Multiple: + +```json +{ + "cmd": [{ + "args": ["/bin/cat", "1"], + "env": ["PATH=/usr/bin:/bin"], + "files": [{ + "content": "" + }, null, { + "name": "stderr", + "max": 10240 + }], + "cpuLimit": 1, + "memoryLimit": 1048576, + "procLimit": 50, + "copyIn": { + "1": { "content": "TEST 1" } + }, + "copyOut": ["stderr"] + }, + { + "args": ["/bin/cat"], + "env": ["PATH=/usr/bin:/bin"], + "files": [null, { + "name": "stdout", + "max": 10240 + }, { + "name": "stderr", + "max": 10240 + }], + "cpuLimit": 1, + "memoryLimit": 1048576, + "procLimit": 50, + "copyOut": ["stdout", "stderr"] + }], + "pipeMapping": [{ + "in" : {"index": 0, "fd": 1 }, + "out" : {"index": 1, "fd" : 0 } + }] } ``` +```json +[ + { + "status": "Accepted", + "time": 1545123, + "memory": 253952, + "files": { + "stderr": "" + }, + "fileIds": {} + }, + { + "status": "Accepted", + "time": 1501463, + "memory": 253952, + "files": { + "stderr": "", + "stdout": "TEST 1" + }, + "fileIds": {} + } +] +``` + ### Workflow ``` text diff --git a/cmd/executorserver/worker.go b/cmd/executorserver/worker.go index b3df4ee..27e4eae 100644 --- a/cmd/executorserver/worker.go +++ b/cmd/executorserver/worker.go @@ -15,7 +15,7 @@ const maxWaiting = 512 type workRequest struct { *request - resultCh chan<- response + resultCh chan<- []response } var ( @@ -55,52 +55,19 @@ func workerLoop() { func workDoCmd(req workRequest) { if len(req.Cmd) == 1 { - req.resultCh <- workDoSingle(req.Cmd[0]) + req.resultCh <- []response{workDoSingle(req.Cmd[0])} } else { - req.resultCh <- response{ - Error: "not implemented yet TAT", - } + req.resultCh <- workDoGroup(req.Cmd, req.PipeMapping) } } func workDoSingle(rc cmd) (res response) { - files, pipeFileName, err := prepareCmdFiles(rc.Files) + c, copyOutSet, err := prepareCmd(rc) if err != nil { res.Status = status(envexec.StatusInternalError) res.Error = err.Error() return } - copyIn, err := prepareCopyIn(rc.CopyIn) - - copyOutSet := make(map[string]bool) - copyOut := make([]string, 0, len(rc.CopyOut)+len(rc.CopyOutCached)) - for _, fn := range rc.CopyOut { - if !pipeFileName[fn] { - copyOut = append(copyOut, fn) - } - copyOutSet[fn] = true - } - for _, fn := range rc.CopyOutCached { - if !pipeFileName[fn] { - copyOut = append(copyOut, fn) - } - } - - w := &waiter{ - timeLimit: time.Duration(rc.CPULimit * float64(time.Second)), - realTimeLimit: time.Duration(rc.RealCPULimit * float64(time.Second)), - } - - c := &envexec.Cmd{ - Args: rc.Args, - Env: rc.Env, - Files: files, - MemoryLimit: runner.Size(rc.MemoryLimit), - ProcLimit: rc.ProcLimit, - CopyIn: copyIn, - CopyOut: copyOut, - Waiter: w.Wait, - } s := &envexec.Single{ CgroupPool: cgroupPool, EnvironmentPool: envPool, @@ -141,6 +108,114 @@ func workDoSingle(rc cmd) (res response) { return } +func workDoGroup(rc []cmd, pm []pipeMap) (rts []response) { + p := preparePipeMapping(pm) + cs := make([]*envexec.Cmd, 0, len(rc)) + copyOutSets := make([]map[string]bool, 0, len(rc)) + for _, cc := range rc { + c, os, err := prepareCmd(cc) + if err != nil { + rts = []response{{Status: status(envexec.StatusInternalError), Error: err.Error()}} + return + } + cs = append(cs, c) + copyOutSets = append(copyOutSets, os) + } + g := envexec.Group{ + CgroupPool: cgroupPool, + EnvironmentPool: envPool, + + Cmd: cs, + Pipes: p, + } + results, err := g.Run() + if err != nil { + rts = []response{{Status: status(envexec.StatusInternalError), Error: err.Error()}} + return + } + rts = make([]response, 0, len(results)) + for i, result := range results { + var res response + res.Status = status(result.Status) + res.Error = result.Error + res.Time = uint64(result.Time) + res.Memory = uint64(result.Memory) + res.Files = make(map[string]string) + res.FileIDs = make(map[string]string) + + for name, fi := range result.Files { + b, err := fi.Content() + if err != nil { + res.Status = status(envexec.StatusFileError) + res.Error = err.Error() + return + } + if copyOutSets[i][name] { + res.Files[name] = string(b) + } else { + id, err := fs.Add(name, b) + if err != nil { + res.Status = status(envexec.StatusFileError) + res.Error = err.Error() + return + } + res.FileIDs[name] = id + } + } + rts = append(rts, res) + } + return +} + +func prepareCmd(rc cmd) (*envexec.Cmd, map[string]bool, error) { + files, pipeFileName, err := prepareCmdFiles(rc.Files) + if err != nil { + return nil, nil, err + } + copyIn, err := prepareCopyIn(rc.CopyIn) + + copyOutSet := make(map[string]bool) + copyOut := make([]string, 0, len(rc.CopyOut)+len(rc.CopyOutCached)) + for _, fn := range rc.CopyOut { + if !pipeFileName[fn] { + copyOut = append(copyOut, fn) + } + copyOutSet[fn] = true + } + for _, fn := range rc.CopyOutCached { + if !pipeFileName[fn] { + copyOut = append(copyOut, fn) + } + } + + w := &waiter{ + timeLimit: time.Duration(rc.CPULimit * float64(time.Second)), + realTimeLimit: time.Duration(rc.RealCPULimit * float64(time.Second)), + } + + return &envexec.Cmd{ + Args: rc.Args, + Env: rc.Env, + Files: files, + MemoryLimit: runner.Size(rc.MemoryLimit), + ProcLimit: rc.ProcLimit, + CopyIn: copyIn, + CopyOut: copyOut, + Waiter: w.Wait, + }, copyOutSet, nil +} + +func preparePipeMapping(pm []pipeMap) []*envexec.Pipe { + rt := make([]*envexec.Pipe, 0, len(pm)) + for _, p := range pm { + rt = append(rt, &envexec.Pipe{ + In: envexec.PipeIndex{Index: p.In.Index, Fd: p.In.Fd}, + Out: envexec.PipeIndex{Index: p.Out.Index, Fd: p.Out.Fd}, + }) + } + return rt +} + func prepareCopyIn(cf map[string]cmdFile) (map[string]file.File, error) { rt := make(map[string]file.File) for name, f := range cf { @@ -194,8 +269,8 @@ func prepareCmdFile(f *cmdFile) (interface{}, error) { } } -func submitRequest(req *request) <-chan response { - ch := make(chan response, 1) +func submitRequest(req *request) <-chan []response { + ch := make(chan []response, 1) workCh <- workRequest{ request: req, resultCh: ch,