From 9f1f7de77073b113b13b900a3a262cea6dbf3277 Mon Sep 17 00:00:00 2001 From: criyle Date: Sun, 29 Sep 2019 15:19:36 -0700 Subject: [PATCH] update design --- README.md | 24 ++++++++++++++++-------- client/interface.go | 11 ++++++++--- file/interface.go | 10 ++++++++-- file/localfile/localfile.go | 11 +++++++---- file/memfile/memfile.go | 6 +++--- go.mod | 2 +- go.sum | 8 ++++---- judger/judger.go | 2 +- judger/loop.go | 16 ++++++++-------- runner/pool.go | 14 +++++++------- runner/run.go | 14 ++++++++------ runner/runner.go | 35 +++++++++++++++++++++++------------ taskqueue/channel/channel.go | 18 +++++++++--------- taskqueue/interface.go | 24 +++++++++++++++++++----- types/run_task.go | 10 +++++----- 15 files changed, 127 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index cd39dd7..472d42c 100644 --- a/README.md +++ b/README.md @@ -6,22 +6,30 @@ The goal to to reimplement [syzoj/judge-v3](https://github.com/syzoj/judge-v3) i ## Planned Design +Container Root Filesystem: + ++ necessary lib / exec / compiler / header readonly bind mounted from current file system: /lib /lib64 /bin /usr ++ work directory tmpfs mount: /w (work dir), /tmp (compiler temp files) ++ additional compiler scripts / exec readonly bind mounted: /c ++ additional header readonly bind mounted: /i + Brokers and interfaces: -+ client: receive pushed judge tasks from website (web-socket / socket.io / RabbitMQ) -+ data: interface to download, cache and access test files from website by id -+ taskqueue: send to and receive from a queue to run task (GO channel / (RabbitMQ, Redis)) and also including upload / download executable files from compile task -+ file: general file interface (local / memory) -+ language: language configuration for runner ++ client: receive pushed judge tasks from website (websocket / socket.io / RabbitMQ) ++ data: interface to download, cache, lock and access test data files from website by data_id ++ taskqueue: message queue to send run task and receive result (GO channel / (RabbitMQ, Redis)) ++ file: general file interface (disk / memory) ++ language: programming language compile & execute configurations Workers: -+ judger: execute judge tasks and distribute as run task to queue -+ runner: receive run task and execute in sandbox (compile / standard / interactive / answer submit) ++ judger: execute judge logics (compile / standard / interactive / answer submit) and distribute as run task to queue ++ runner: receive run task and execute in sandbox (dumb runner) Models: -+ JudgeTask: judge task pushed from website (type, data) ++ JudgeTask: judge task pushed from website (type, source, data) ++ JudgeResult: judge task result send back to website + JudgeSetting: problem setting (from yaml) and JudgeCase + RunTask: run task parameters send to run_queue + RunResult: run task result sent back from queue diff --git a/client/interface.go b/client/interface.go index 935b748..fda2590 100644 --- a/client/interface.go +++ b/client/interface.go @@ -4,14 +4,19 @@ import "github.com/criyle/go-judge/types" // Task contains a single task received from web type Task interface { + // Param get the judge task Param() *types.JudgeTask - Progress(p types.JudgeProgress) - Finish(p types.JudgeResult) + + // Progress emits current progress to website + Progress(*types.JudgeProgress) + + // Finish emits the final result to website + Finish(*types.JudgeResult) } // Client should connect to web service and receive works from web // it should sent received work through go channel (have background goroutine(s)) type Client interface { - // C should return channel to receive works + // C return channel to receive works C() <-chan Task } diff --git a/file/interface.go b/file/interface.go index 9fbae5b..b29b9aa 100644 --- a/file/interface.go +++ b/file/interface.go @@ -5,7 +5,13 @@ import "os" // File defines file name with its content // file could on file system or memory type File interface { + // Name get the file name Name() string - Content() ([]byte, error) // get content of the file - Open() (*os.File, error) // get readonly fd of the file + + // Content reads the file content + Content() ([]byte, error) + + // Open opens the file in readonly mode + // caller should close afterwards + Open() (*os.File, error) } diff --git a/file/localfile/localfile.go b/file/localfile/localfile.go index 00cd26b..5174f3a 100644 --- a/file/localfile/localfile.go +++ b/file/localfile/localfile.go @@ -3,23 +3,26 @@ package localfile import ( "io/ioutil" "os" + + "github.com/criyle/go-judge/file" ) // File stores a path to represent a local file type File struct { - path string + name, path string } -// New creats a wrapper to path -func New(path string) *File { +// New creates a wrapper to file system by path +func New(name, path string) file.File { return &File{ + name: name, path: path, } } // Name get the path func (f *File) Name() string { - return f.path + return f.name } // Content reads file content diff --git a/file/memfile/memfile.go b/file/memfile/memfile.go index ce7b7c0..80e662a 100644 --- a/file/memfile/memfile.go +++ b/file/memfile/memfile.go @@ -4,6 +4,7 @@ import ( "bytes" "os" + "github.com/criyle/go-judge/file" "github.com/criyle/go-sandbox/pkg/memfd" ) @@ -13,9 +14,8 @@ type File struct { content []byte } -// New create a file interface, content should not be modified after -// NewMemFile -func New(name string, content []byte) *File { +// New create a file interface from content and content should not be modified afterwards +func New(name string, content []byte) file.File { return &File{ name: name, content: content, diff --git a/go.mod b/go.mod index afec747..1b6867e 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,4 @@ module github.com/criyle/go-judge go 1.12 -require github.com/criyle/go-sandbox v0.0.0-20190915221016-9898c301d566 +require github.com/criyle/go-sandbox v0.0.0-20190929004305-4001c7c7671f diff --git a/go.sum b/go.sum index d693357..badf642 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -github.com/criyle/go-sandbox v0.0.0-20190915221016-9898c301d566 h1:JB33qewgHSIHojNHn9tnFu1HaBwRkh8zBDP0LAu6Ars= -github.com/criyle/go-sandbox v0.0.0-20190915221016-9898c301d566/go.mod h1:V5llzm3jeoB8pq2XEwX/tSVK3RQxjPl+vTSBDFGUO4A= +github.com/criyle/go-sandbox v0.0.0-20190929004305-4001c7c7671f h1:2VJ5GhYvEEXOwIeRF2SNepXp5TQ98ZjELGIriGWQIps= +github.com/criyle/go-sandbox v0.0.0-20190929004305-4001c7c7671f/go.mod h1:xAafqqJsAlGuL8gMgnrbNtYLmCtAWzOZQ4kM9y6/zdk= github.com/seccomp/libseccomp-golang v0.9.1/go.mod h1:GbW5+tmTXfcxTToHLXlScSlAvWlF4P2Ca7zGrPiEpWo= -golang.org/x/sys v0.0.0-20190913121621-c3b328c6e5a7 h1:wYqz/tQaWUgGKyx+B/rssSE6wkIKdY5Ee6ryOmzarIg= -golang.org/x/sys v0.0.0-20190913121621-c3b328c6e5a7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190927073244-c990c680b611 h1:q9u40nxWT5zRClI/uU9dHCiYGottAg6Nzz4YUQyHxdA= +golang.org/x/sys v0.0.0-20190927073244-c990c680b611/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= diff --git a/judger/judger.go b/judger/judger.go index f82b44c..9c27a39 100644 --- a/judger/judger.go +++ b/judger/judger.go @@ -9,6 +9,6 @@ import ( // Judger receives task from client and translate to task for runner type Judger struct { client.Client - taskqueue.Queue + taskqueue.Sender problem.Builder } diff --git a/judger/loop.go b/judger/loop.go index 536b6e4..3f9450b 100644 --- a/judger/loop.go +++ b/judger/loop.go @@ -30,11 +30,11 @@ loop: } } -func (j *Judger) run(done <-chan struct{}, t client.Task) types.JudgeResult { +func (j *Judger) run(done <-chan struct{}, t client.Task) *types.JudgeResult { var result types.JudgeResult - errResult := func(err error) types.JudgeResult { + errResult := func(err error) *types.JudgeResult { result.Error = err.Error() - return result + return &result } p := t.Param() @@ -45,7 +45,7 @@ func (j *Judger) run(done <-chan struct{}, t client.Task) types.JudgeResult { // compile compileRet := make(chan types.RunTaskResult) - err = j.Enqueue(types.RunTask{ + err = j.Send(types.RunTask{ Type: "compile", Language: p.Language, Code: p.Code, @@ -78,7 +78,7 @@ func (j *Judger) run(done <-chan struct{}, t client.Task) types.JudgeResult { for range pconf.Subtasks { result.SubTasks = append(result.SubTasks, <-subTaskResult) } - return result + return &result } type problemJudger struct { @@ -94,12 +94,12 @@ func (pj *problemJudger) runSubtask(done <-chan struct{}, exec []file.File, s *t var result types.JudgeSubTaskResult caseResult := make(chan types.RunTaskResult, len(s.Cases)) for _, c := range s.Cases { - pj.Enqueue(types.RunTask{ + pj.Send(types.RunTask{ Type: pj.ProblemConfig.Type, Language: pj.Language, TimeLimit: pj.TileLimit, MemoryLimit: pj.MemoryLimit, - Executables: exec, + ExtraFiles: exec, InputFile: c.Input, AnswerFile: c.Answer, }, caseResult) @@ -121,7 +121,7 @@ func (pj *problemJudger) runSubtask(done <-chan struct{}, exec []file.File, s *t result.Score += rt.ScoringRate // report prograss atomic.AddInt32(&pj.count, 1) - pj.Progress(types.JudgeProgress{ + pj.Progress(&types.JudgeProgress{ Message: fmt.Sprintf("%d/%d", pj.count, pj.total), }) } diff --git a/runner/pool.go b/runner/pool.go index 4f0cefa..d30a1bc 100644 --- a/runner/pool.go +++ b/runner/pool.go @@ -7,17 +7,17 @@ import ( ) type pool struct { - queue chan *daemon.Master - count int32 - root string + builder *daemon.Builder + queue chan *daemon.Master + count int32 } const maxPoolSize = 64 -func newPool(root string) *pool { +func newPool(builder *daemon.Builder) *pool { return &pool{ - queue: make(chan *daemon.Master, maxPoolSize), - root: root, + queue: make(chan *daemon.Master, maxPoolSize), + builder: builder, } } @@ -28,7 +28,7 @@ func (p *pool) Get() (*daemon.Master, error) { default: } atomic.AddInt32(&p.count, 1) - return daemon.New(p.root) + return p.builder.Build() } func (p *pool) Put(master *daemon.Master) { diff --git a/runner/run.go b/runner/run.go index 634080a..055c875 100644 --- a/runner/run.go +++ b/runner/run.go @@ -21,7 +21,9 @@ const cgroupPrefix = "go-judge" const minCPUPercent = 40 // 40% const checkIntervalMS = 50 -var env = []string{"PATH=/usr/local/bin:/usr/bin:/bin"} +var env = []string{ + "PATH=/usr/local/bin:/usr/bin:/bin", +} func (r *Runner) run(done <-chan struct{}, task *types.RunTask) *types.RunTaskResult { t := language.TypeExec @@ -80,7 +82,7 @@ func (r *Runner) run(done <-chan struct{}, task *types.RunTask) *types.RunTaskRe SyncFunc: cg.AddProc, } - // copyin source code for compile or executables for exec + // copyin source code for compile or exec files for exec if t == language.TypeCompile { source := memfile.New("source", []byte(task.Code)) sourceFile, err := source.Open() @@ -92,7 +94,7 @@ func (r *Runner) run(done <-chan struct{}, task *types.RunTask) *types.RunTaskRe return errResult(err.Error()) } } else { - for _, f := range task.Executables { + for _, f := range task.ExecFiles { execFile, err := f.Open() if err != nil { return errResult(err.Error()) @@ -131,7 +133,7 @@ func (r *Runner) run(done <-chan struct{}, task *types.RunTask) *types.RunTaskRe var lastCPUUsage uint64 var totalTime time.Duration var rt stypes.Result - var rtreceived bool + var rtReceived bool lastCheckTime := time.Now() // wait task finish @@ -156,7 +158,7 @@ loop: lastCPUUsage = cpuUsage case rt = <-rc: // returned - rtreceived = true + rtReceived = true break loop case <-outputPipe.Done: // outputlimit exceeded @@ -169,7 +171,7 @@ loop: // get result if did not received cancelC.cancel() - if !rtreceived { + if !rtReceived { rt = <-rc } diff --git a/runner/runner.go b/runner/runner.go index 0b1d52a..5e9661f 100644 --- a/runner/runner.go +++ b/runner/runner.go @@ -5,38 +5,49 @@ import ( "github.com/criyle/go-judge/language" "github.com/criyle/go-judge/taskqueue" + "github.com/criyle/go-sandbox/daemon" ) // Runner is the task runner type Runner struct { - Queue taskqueue.Queue + // Queue is the message queue to receive run tasks + Queue taskqueue.Receiver + + // Builder builds the sandbox runner + Builder *daemon.Builder + Language language.Language - Root string - pool *pool - once sync.Once + + // pool of sandbox to use + pool *pool + once sync.Once } func (r *Runner) init() { - r.pool = newPool(r.Root) + r.pool = newPool(r.Builder) } // Loop status a runner in a forever loop, waiting for task and execute // call it in new goroutine func (r *Runner) Loop(done <-chan struct{}) error { r.once.Do(r.init) - c := r.Queue.C() + c := r.Queue.ReceiveC() loop: for { select { case <-done: break loop + case task := <-c: - select { - case <-done: - break loop - default: - task.Finish(r.run(done, task.Task())) - } + task.Done(r.run(done, task.Task())) + } + + // check if cancel is signaled + select { + case <-done: + break loop + + default: } } return nil diff --git a/taskqueue/channel/channel.go b/taskqueue/channel/channel.go index 219c566..2d7da06 100644 --- a/taskqueue/channel/channel.go +++ b/taskqueue/channel/channel.go @@ -7,20 +7,20 @@ import ( const buffSize = 512 -// Queue implements taskqueue by go channel +// Queue implements taskqueue by buffered go channel type Queue struct { queue chan taskqueue.Task } -// New craetes new Queue with buffed go channel -func New() *Queue { +// New creates new Queue with buffed go channel +func New() taskqueue.Queue { return &Queue{ queue: make(chan taskqueue.Task, buffSize), } } -// Enqueue puts task into run queue -func (q *Queue) Enqueue(t types.RunTask, r chan<- types.RunTaskResult) error { +// Send puts task into run queue +func (q *Queue) Send(t types.RunTask, r chan<- types.RunTaskResult) error { q.queue <- Task{ task: t, result: r, @@ -28,8 +28,8 @@ func (q *Queue) Enqueue(t types.RunTask, r chan<- types.RunTaskResult) error { return nil } -// C returns the underlying channel -func (q *Queue) C() <-chan taskqueue.Task { +// ReceiveC returns the underlying channel +func (q *Queue) ReceiveC() <-chan taskqueue.Task { return q.queue } @@ -44,7 +44,7 @@ func (t Task) Task() *types.RunTask { return &t.task } -// Finish returns the run task result -func (t Task) Finish(r *types.RunTaskResult) { +// Done returns the run task result +func (t Task) Done(r *types.RunTaskResult) { t.result <- *r } diff --git a/taskqueue/interface.go b/taskqueue/interface.go index 29a951e..ba853e0 100644 --- a/taskqueue/interface.go +++ b/taskqueue/interface.go @@ -2,15 +2,29 @@ package taskqueue import "github.com/criyle/go-judge/types" -// Queue provides asynchronous message queue to run execution tasks +// Sender interface is used to send run task into message queue +type Sender interface { + // Send used to initial a RPC call and receive result to channel (should have enough space) + Send(types.RunTask, chan<- types.RunTaskResult) error +} + +// Receiver interface is used to receive run task from message queue +type Receiver interface { + // ReceiveC get the channel to receive tasks + ReceiveC() <-chan Task +} + +// Queue provides asynchronous message queue for run tasks type Queue interface { - // Enqueue used to initial a RPC call and receive result to channel - Enqueue(types.RunTask, chan<- types.RunTaskResult) error - C() <-chan Task + Sender + Receiver } // Task represent a single task to run type Task interface { + // Task gets the run task parameter Task() *types.RunTask - Finish(*types.RunTaskResult) + + // Done returns the run task result for the run task (should be called only once at end) + Done(*types.RunTaskResult) } diff --git a/types/run_task.go b/types/run_task.go index 3017fa3..17fda58 100644 --- a/types/run_task.go +++ b/types/run_task.go @@ -15,17 +15,17 @@ type RunTask struct { // Used for run tasks TimeLimit uint64 // ms MemoryLimit uint64 // kb - Executables []file.File + ExecFiles []file.File InputFile file.File AnswerFile file.File // Used for standard run task - SpjLanguage string - SpjExecutables []file.File + SpjLanguage string + SpjExecFiles []file.File // Used for interaction run task - InteractorLanguage string - InteractorExecutables []file.File + InteractorLanguage string + InteractorExecFiles []file.File // Used for answer submission run task UserAnswer file.File