blob: c87baaf9a1e06fce5606543124707b5d08552a7f [file] [log] [blame]
// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package jobqueue implements a job queue for jobs.
//
// Usage:
//
// dispatcher := NewDispatcher(numWorkers, maxWaitingJobs)
//
// job := NewJob(...)
// resultChan := dispatcher.Enqueue(job)
// r :=<- resultChan // r will contain the result of running the job.
//
// dispatcher.Stop() // Waits for any in-progress jobs to finish, cancels any
// // remaining jobs.
//
// Internally, the dispatcher has a channel of workers that represents a worker
// queue, and a channel of jobs that represents a job queue. The dispatcher
// reads a worker off the worker queue, and then reads a job off the job
// queue, and runs that job on that worker. When the job finishes, the worker
// is pushed back on to the worker queue.
//
// TODO(nlacasse): There are many types and functions exported in this file
// which are only exported because they are used by the compile test, in
// particular Job, Dispatcher, and Result types, and their constructors and
// methods. Consider refactoring the compiler so those tests and the logic
// they test become part of this package.
package jobqueue
import (
"bytes"
"fmt"
"io/ioutil"
"os"
"os/exec"
"sync"
"time"
"v.io/x/playground/lib"
"v.io/x/playground/lib/event"
"v.io/x/playground/lib/log"
)
var (
// A channel which returns unique playground ids.
uniq = make(chan string)
)
func init() {
val := time.Now().UnixNano()
go func() {
for {
uniq <- fmt.Sprintf("playground_%d", val)
val++
}
}()
}
type Job struct {
id string
body []byte
res *event.ResponseEventSink
resultChan chan Result
maxSize int
maxTime time.Duration
useDocker bool
dockerMem int
mu sync.Mutex
cancelled bool
}
func NewJob(body []byte, res *event.ResponseEventSink, maxSize int, maxTime time.Duration, useDocker bool, dockerMem int) *Job {
return &Job{
id: <-uniq,
body: body,
res: res,
maxSize: maxSize,
maxTime: maxTime,
useDocker: useDocker,
dockerMem: dockerMem,
// resultChan has capacity 1 so that writing to the channel won't block
// if nobody ever reads the result.
resultChan: make(chan Result, 1),
}
}
// Body is a getter for Job.body.
func (j *Job) Body() []byte {
return j.body
}
// Cancel will prevent the job from being run, if it has not already been
// started by a worker.
func (j *Job) Cancel() {
j.mu.Lock()
defer j.mu.Unlock()
log.Debugf("Cancelling job %v.\n", j.id)
j.cancelled = true
}
// Dispatcher is an interface type so it can be mocked during tests.
type Dispatcher interface {
Enqueue(j *Job) (chan Result, error)
Stop()
}
// dispatcherImpl implements Dispatcher interface.
type dispatcherImpl struct {
jobQueue chan *Job
// A message sent on the stopped channel causes the dispatcher to stop
// assigning new jobs to workers.
stopped chan bool
// wg represents currently running workers. It is used during Stop to make
// sure that all workers have finished running their active jobs.
wg sync.WaitGroup
}
var _ = Dispatcher((*dispatcherImpl)(nil))
func NewDispatcher(workers int, jobQueueCap int) Dispatcher {
log.Debugf("Creating new dispatcher with %v workers and %v queue capacity.\n", workers, jobQueueCap)
d := &dispatcherImpl{
jobQueue: make(chan *Job, jobQueueCap),
stopped: make(chan bool),
}
d.start(workers)
return d
}
// start starts a given number of workers, then reads from the jobQueue and
// assigns jobs to free workers.
func (d *dispatcherImpl) start(num int) {
log.Debugf("Dispatcher starting.\n")
// Workers are published on the workerQueue when they are free.
workerQueue := make(chan *worker, num)
for i := 0; i < num; i++ {
worker := newWorker(i)
workerQueue <- worker
}
d.wg.Add(1)
go func() {
Loop:
for {
// Wait for the next available worker.
select {
case <-d.stopped:
break Loop
case worker := <-workerQueue:
// Read the next job from the job queue.
select {
case <-d.stopped:
break Loop
case job := <-d.jobQueue:
job.mu.Lock()
cancelled := job.cancelled
job.mu.Unlock()
if cancelled {
log.Debugf("Dispatcher encountered cancelled job %v, rejecting.\n", job.id)
job.resultChan <- Result{
Success: false,
Events: nil,
}
workerQueue <- worker
} else {
log.Debugf("Dispatching job %v to worker %v.\n", job.id, worker.id)
d.wg.Add(1)
go func() {
job.resultChan <- worker.run(job)
log.Debugf("Job %v finished on worker %v.\n", job.id, worker.id)
d.wg.Done()
workerQueue <- worker
}()
}
}
}
}
log.Debugf("Dispatcher stopped.\n")
// Dispatcher stopped, treat all remaining jobs as cancelled.
for {
select {
case job := <-d.jobQueue:
log.Debugf("Dispatcher is stopped, rejecting job %v.\n", job.id)
job.resultChan <- Result{
Success: false,
Events: nil,
}
default:
log.Debugf("Dispatcher job queue drained.\n")
d.wg.Done()
return
}
}
}()
}
// Stop stops the dispatcher from assigning any new jobs to workers. Jobs that
// are currently running are allowed to continue. Other jobs are treated as
// cancelled. Stop blocks until all jobs have finished.
// TODO(nlacasse): Consider letting the dispatcher run all currently queued
// jobs, rather than rejecting them. Or, put logic in the client to retry
// cancelled jobs.
func (d *dispatcherImpl) Stop() {
log.Debugf("Stopping dispatcher.\n")
d.stopped <- true
// Wait for workers to finish their current jobs.
d.wg.Wait()
}
// Enqueue queues a job to be run be the next available worker. It returns a
// channel on which the job's results will be published.
func (d *dispatcherImpl) Enqueue(j *Job) (chan Result, error) {
select {
case d.jobQueue <- j:
return j.resultChan, nil
default:
return nil, fmt.Errorf("Error queuing job. Job queue full.")
}
}
type Result struct {
Success bool
Events []event.Event
}
type worker struct {
id int
}
func newWorker(id int) *worker {
return &worker{
id: id,
}
}
// run compiles and runs a job, caches the result, and returns the result on
// the job's result channel.
func (w *worker) run(j *Job) Result {
event.Debug(j.res, "Preparing to run program")
memoryFlag := fmt.Sprintf("%dm", j.dockerMem)
var cmd *exec.Cmd
if j.useDocker {
// TODO(nlacasse,ivanpi): Limit the CPU resources used by this docker
// builder instance. The docker "cpu-shares" flag can only limit on
// docker process relative to another, so it's not useful for limiting
// the cpu resources of all build instances. The docker "cpuset" flag
// can pin the instance to a specific processor, so that might be of
// use.
cmd = docker("run", "-i",
"--name", j.id,
// Disable external networking.
"--net", "none",
// Limit instance memory.
"--memory", memoryFlag,
// Limit instance memory+swap combined.
// Setting to the same value as memory effectively disables swap.
"--memory-swap", memoryFlag,
"playground")
} else {
// Run builder directly, without Docker. This should only happen during
// development and in tests, never in production.
cmd = exec.Command("builder")
// Run the builder in a temp dir, so the bundle files and binaries do
// not clutter up the current working dir. This also allows parallel
// bundler runs, since otherwise the files from different runs stomp on
// each other.
tmpDir, err := ioutil.TempDir("", "pg-builder-")
if err != nil {
log.Panicf("Error creating temp dir for builder: %v\n", err)
} else {
cmd.Dir = tmpDir
}
}
cmdKill := lib.DoOnce(func() {
event.Debug(j.res, "Killing program")
// The docker client can get in a state where stopping/killing/rm-ing
// the container will not kill the client. The opposite should work
// correctly (killing the docker client stops the container).
// If not, the docker rm call below will.
// Note, this wouldn't be sufficient if docker was called through sudo
// since sudo doesn't pass sigkill to child processes.
cmd.Process.Kill()
})
cmd.Stdin = bytes.NewReader(j.body)
// Builder will return all normal output as JSON Events on stdout, and will
// return unexpected errors on stderr.
// TODO(sadovsky): Security issue: what happens if the program output is huge?
// We can restrict memory use of the Docker container, but these buffers are
// outside Docker.
// TODO(ivanpi): Revisit above comment.
sizedOut := false
erroredOut := false
userLimitCallback := func() {
sizedOut = true
cmdKill()
}
systemLimitCallback := func() {
erroredOut = true
cmdKill()
}
userErrorCallback := func(err error) {
// A relay error can result from unparseable JSON caused by a builder bug
// or a malicious exploit inside Docker. Panicking could lead to a DoS.
log.Errorln(j.id, "builder stdout relay error:", err)
erroredOut = true
cmdKill()
}
outRelay, outStop := event.LimitedEventRelay(j.res, j.maxSize, userLimitCallback, userErrorCallback)
// Builder stdout should already contain a JSON Event stream.
cmd.Stdout = outRelay
// Any stderr is unexpected, most likely a bug (panic) in builder, but could
// also result from a malicious exploit inside Docker.
// It is quietly logged as long as it doesn't exceed maxSize.
errBuffer := new(bytes.Buffer)
cmd.Stderr = lib.NewLimitedWriter(errBuffer, j.maxSize, systemLimitCallback)
event.Debug(j.res, "Running program")
timeout := time.After(j.maxTime)
// User code execution is time limited in builder.
// This flag signals only unexpected timeouts. maxTime should be sufficient
// for end-to-end request processing by builder for worst-case user input.
// TODO(ivanpi): builder doesn't currently time compilation, so builder
// worst-case execution time is not clearly bounded.
timedOut := false
exit := make(chan error)
go func() { exit <- cmd.Run() }()
select {
case err := <-exit:
if err != nil && !sizedOut {
erroredOut = true
}
case <-timeout:
timedOut = true
cmdKill()
<-exit
}
// Close and wait for the output relay.
outStop()
event.Debug(j.res, "Program exited")
// Return the appropriate error message to the client.
if timedOut {
j.res.Write(event.New("", "stderr", "Internal timeout, please retry."))
} else if erroredOut {
j.res.Write(event.New("", "stderr", "Internal error, please retry."))
} else if sizedOut {
j.res.Write(event.New("", "stderr", "Program output too large, killed."))
}
// Log builder internal errors, if any.
// TODO(ivanpi): Prevent caching? Report to client if debug requested?
if errBuffer.Len() > 0 {
log.Warnln(j.id, "builder stderr:", errBuffer.String())
}
event.Debug(j.res, "Response finished")
// TODO(nlacasse): This "docker rm" can be slow (several seconds), and seems
// to block other Docker commands, thereby slowing down other concurrent
// requests. We should figure out how to make it not block other Docker
// commands. Setting GOMAXPROCS may or may not help.
// See: https://github.com/docker/docker/issues/6480
if j.useDocker {
go func() {
docker("rm", "-f", j.id).Run()
}()
} else {
// Clean up after the builder process.
os.RemoveAll(cmd.Dir)
}
// If we timed out or errored out, do not cache anything.
// TODO(sadovsky): This policy is helpful for development, but may not be wise
// for production. Revisit.
if !timedOut && !erroredOut {
return Result{
Success: true,
Events: j.res.PopWrittenEvents(),
}
} else {
return Result{
Success: false,
Events: nil,
}
}
}
func docker(args ...string) *exec.Cmd {
return exec.Command("docker", args...)
}