go/src/v.io/x/playground/compilerd/jobqueue/jobqueue.go - release.projects.playground - Git at Google

 // 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...)
 }
	// 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...)
	}