Implements multiple command & update example

This commit is contained in:
criyle 2020-03-04 17:27:27 -05:00
parent fa072e373f
commit 8e8cc17123
2 changed files with 196 additions and 50 deletions

View File

@ -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

View File

@ -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,