Engine: integer job status, improved stream API

* Jobs return an integer status instead of a string * Status convention mimics unix process execution: 0=success, 1=generic error, 127="no such command" * Stdout and Stderr support multiple thread-safe data receivers and ring buffer filtering
2026-09-05 05:01:28 +00:00 · 2013-11-30 00:25:46 +00:00
parent 82cecb34b5
commit a4f8a2494b
12 changed files with 290 additions and 154 deletions
@@ -9,7 +9,7 @@ import (
 	"strings"
 )

-type Handler func(*Job) string
+type Handler func(*Job) Status

 var globalHandlers map[string]Handler

@@ -99,10 +99,12 @@ func (eng *Engine) Job(name string, args ...string) *Job {
 		Eng:    eng,
 		Name:   name,
 		Args:   args,
-		Stdin:  os.Stdin,
-		Stdout: os.Stdout,
-		Stderr: os.Stderr,
+		Stdin:  NewInput(),
+		Stdout: NewOutput(),
+		Stderr: NewOutput(),
 	}
+	job.Stdout.Add(utils.NopWriteCloser(os.Stdout))
+	job.Stderr.Add(utils.NopWriteCloser(os.Stderr))
 	handler, exists := eng.handlers[name]
 	if exists {
 		job.handler = handler
@@ -38,8 +38,9 @@ func TestJob(t *testing.T) {
 		t.Fatalf("job1.handler should be empty")
 	}

-	h := func(j *Job) string {
-		return j.Name
+	h := func(j *Job) Status {
+		j.Printf("%s\n", j.Name)
+		return 42
 	}

 	eng.Register("dummy2", h)
@@ -49,7 +50,7 @@ func TestJob(t *testing.T) {
 		t.Fatalf("job2.handler shouldn't be nil")
 	}

-	if job2.handler(job2) != job2.Name {
+	if job2.handler(job2) != 42 {
 		t.Fatalf("handler dummy2 was not found in job2")
 	}
 }
@@ -1,32 +1,18 @@
 package engine

 import (
-	"fmt"
 	"github.com/dotcloud/docker/utils"
-	"io/ioutil"
-	"runtime"
-	"strings"
 	"testing"
 )

 var globalTestID string

 func newTestEngine(t *testing.T) *Engine {
-	// Use the caller function name as a prefix.
-	// This helps trace temp directories back to their test.
-	pc, _, _, _ := runtime.Caller(1)
-	callerLongName := runtime.FuncForPC(pc).Name()
-	parts := strings.Split(callerLongName, ".")
-	callerShortName := parts[len(parts)-1]
-	if globalTestID == "" {
-		globalTestID = utils.RandomString()[:4]
-	}
-	prefix := fmt.Sprintf("docker-test%s-%s-", globalTestID, callerShortName)
-	root, err := ioutil.TempDir("", prefix)
+	tmp, err := utils.TestDirectory("")
 	if err != nil {
 		t.Fatal(err)
 	}
-	eng, err := New(root)
+	eng, err := New(tmp)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1,16 +1,13 @@
 package engine

 import (
-	"bufio"
 	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
-	"io/ioutil"
-	"os"
 	"strconv"
 	"strings"
-	"sync"
+	"time"
 )

 // A job is the fundamental unit of work in the docker engine.
@@ -31,126 +28,75 @@ type Job struct {
 	Name    string
 	Args    []string
 	env     []string
-	Stdin   io.Reader
-	Stdout  io.Writer
-	Stderr  io.Writer
-	handler func(*Job) string
-	status  string
+	Stdout  *Output
+	Stderr  *Output
+	Stdin   *Input
+	handler Handler
+	status  Status
+	end     time.Time
 	onExit  []func()
 }

+type Status int
+
+const (
+	StatusOK       Status = 0
+	StatusErr      Status = 1
+	StatusNotFound Status = 127
+)
+
 // Run executes the job and blocks until the job completes.
 // If the job returns a failure status, an error is returned
 // which includes the status.
 func (job *Job) Run() error {
-	defer func() {
-		var wg sync.WaitGroup
-		for _, f := range job.onExit {
-			wg.Add(1)
-			go func(f func()) {
-				f()
-				wg.Done()
-			}(f)
-		}
-		wg.Wait()
-	}()
-	if job.Stdout != nil && job.Stdout != os.Stdout {
-		job.Stdout = io.MultiWriter(job.Stdout, os.Stdout)
-	}
-	if job.Stderr != nil && job.Stderr != os.Stderr {
-		job.Stderr = io.MultiWriter(job.Stderr, os.Stderr)
+	// FIXME: make this thread-safe
+	// FIXME: implement wait
+	if !job.end.IsZero() {
+		return fmt.Errorf("%s: job has already completed", job.Name)
 	}
+	// Log beginning and end of the job
 	job.Eng.Logf("+job %s", job.CallString())
 	defer func() {
 		job.Eng.Logf("-job %s%s", job.CallString(), job.StatusString())
 	}()
+	var errorMessage string
+	job.Stderr.AddString(&errorMessage)
 	if job.handler == nil {
-		job.status = "command not found"
+		job.Errorf("%s: command not found", job.Name)
+		job.status = 127
 	} else {
 		job.status = job.handler(job)
+		job.end = time.Now()
 	}
-	if job.status != "0" {
-		return fmt.Errorf("%s: %s", job.Name, job.status)
+	// Wait for all background tasks to complete
+	if err := job.Stdout.Close(); err != nil {
+		return err
+	}
+	if err := job.Stderr.Close(); err != nil {
+		return err
+	}
+	if job.status != 0 {
+		return fmt.Errorf("%s: %s", job.Name, errorMessage)
 	}
 	return nil
 }

-func (job *Job) StdoutParseLines(dst *[]string, limit int) {
-	job.parseLines(job.StdoutPipe(), dst, limit)
-}
-
-func (job *Job) StderrParseLines(dst *[]string, limit int) {
-	job.parseLines(job.StderrPipe(), dst, limit)
-}
-
-func (job *Job) parseLines(src io.Reader, dst *[]string, limit int) {
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		scanner := bufio.NewScanner(src)
-		for scanner.Scan() {
-			// If the limit is reached, flush the rest of the source and return
-			if limit > 0 && len(*dst) >= limit {
-				io.Copy(ioutil.Discard, src)
-				return
-			}
-			line := scanner.Text()
-			// Append the line (with delimitor removed)
-			*dst = append(*dst, line)
-		}
-	}()
-	job.onExit = append(job.onExit, wg.Wait)
-}
-
-func (job *Job) StdoutParseString(dst *string) {
-	lines := make([]string, 0, 1)
-	job.StdoutParseLines(&lines, 1)
-	job.onExit = append(job.onExit, func() {
-		if len(lines) >= 1 {
-			*dst = lines[0]
-		}
-	})
-}
-
-func (job *Job) StderrParseString(dst *string) {
-	lines := make([]string, 0, 1)
-	job.StderrParseLines(&lines, 1)
-	job.onExit = append(job.onExit, func() { *dst = lines[0] })
-}
-
-func (job *Job) StdoutPipe() io.ReadCloser {
-	r, w := io.Pipe()
-	job.Stdout = w
-	job.onExit = append(job.onExit, func() { w.Close() })
-	return r
-}
-
-func (job *Job) StderrPipe() io.ReadCloser {
-	r, w := io.Pipe()
-	job.Stderr = w
-	job.onExit = append(job.onExit, func() { w.Close() })
-	return r
-}
-
 func (job *Job) CallString() string {
 	return fmt.Sprintf("%s(%s)", job.Name, strings.Join(job.Args, ", "))
 }

 func (job *Job) StatusString() string {
-	// FIXME: if a job returns the empty string, it will be printed
-	// as not having returned.
-	// (this only affects String which is a convenience function).
-	if job.status != "" {
-		var okerr string
-		if job.status == "0" {
-			okerr = "OK"
-		} else {
-			okerr = "ERR"
-		}
-		return fmt.Sprintf(" = %s (%s)", okerr, job.status)
+	// If the job hasn't completed, status string is empty
+	if job.end.IsZero() {
+		return ""
 	}
-	return ""
+	var okerr string
+	if job.status == StatusOK {
+		okerr = "OK"
+	} else {
+		okerr = "ERR"
+	}
+	return fmt.Sprintf(" = %s (%d)", okerr, job.status)
 }

 // String returns a human-readable description of `job`
@@ -338,5 +284,8 @@ func (job *Job) Printf(format string, args ...interface{}) (n int, err error) {

 func (job *Job) Errorf(format string, args ...interface{}) (n int, err error) {
 	return fmt.Fprintf(job.Stderr, format, args...)
+}

+func (job *Job) Error(err error) (int, error) {
+	return fmt.Fprintf(job.Stderr, "%s", err)
 }
@@ -0,0 +1,166 @@
+package engine
+
+import (
+	"bufio"
+	"container/ring"
+	"fmt"
+	"io"
+	"sync"
+)
+
+type Output struct {
+	sync.Mutex
+	dests []io.Writer
+	tasks sync.WaitGroup
+}
+
+// NewOutput returns a new Output object with no destinations attached.
+// Writing to an empty Output will cause the written data to be discarded.
+func NewOutput() *Output {
+	return &Output{}
+}
+
+// Add attaches a new destination to the Output. Any data subsequently written
+// to the output will be written to the new destination in addition to all the others.
+// This method is thread-safe.
+// FIXME: Add cannot fail
+func (o *Output) Add(dst io.Writer) error {
+	o.Mutex.Lock()
+	defer o.Mutex.Unlock()
+	o.dests = append(o.dests, dst)
+	return nil
+}
+
+// AddPipe creates an in-memory pipe with io.Pipe(), adds its writing end as a destination,
+// and returns its reading end for consumption by the caller.
+// This is a rough equivalent similar to Cmd.StdoutPipe() in the standard os/exec package.
+// This method is thread-safe.
+func (o *Output) AddPipe() (io.Reader, error) {
+	r, w := io.Pipe()
+	o.Add(w)
+	return r, nil
+}
+
+// AddTail starts a new goroutine which will read all subsequent data written to the output,
+// line by line, and append the last `n` lines to `dst`.
+func (o *Output) AddTail(dst *[]string, n int) error {
+	src, err := o.AddPipe()
+	if err != nil {
+		return err
+	}
+	o.tasks.Add(1)
+	go func() {
+		defer o.tasks.Done()
+		Tail(src, n, dst)
+	}()
+	return nil
+}
+
+// AddString starts a new goroutine which will read all subsequent data written to the output,
+// line by line, and store the last line into `dst`.
+func (o *Output) AddString(dst *string) error {
+	src, err := o.AddPipe()
+	if err != nil {
+		return err
+	}
+	o.tasks.Add(1)
+	go func() {
+		defer o.tasks.Done()
+		lines := make([]string, 0, 1)
+		Tail(src, 1, &lines)
+		if len(lines) == 0 {
+			*dst = ""
+		} else {
+			*dst = lines[0]
+		}
+	}()
+	return nil
+}
+
+// Write writes the same data to all registered destinations.
+// This method is thread-safe.
+func (o *Output) Write(p []byte) (n int, err error) {
+	o.Mutex.Lock()
+	defer o.Mutex.Unlock()
+	var firstErr error
+	for _, dst := range o.dests {
+		_, err := dst.Write(p)
+		if err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	return len(p), err
+}
+
+// Close unregisters all destinations and waits for all background
+// AddTail and AddString tasks to complete.
+// The Close method of each destination is called if it exists.
+func (o *Output) Close() error {
+	o.Mutex.Lock()
+	defer o.Mutex.Unlock()
+	var firstErr error
+	for _, dst := range o.dests {
+		if closer, ok := dst.(io.WriteCloser); ok {
+			err := closer.Close()
+			if err != nil && firstErr == nil {
+				firstErr = err
+			}
+		}
+	}
+	o.tasks.Wait()
+	return firstErr
+}
+
+type Input struct {
+	src io.Reader
+	sync.Mutex
+}
+
+// NewInput returns a new Input object with no source attached.
+// Reading to an empty Input will return io.EOF.
+func NewInput() *Input {
+	return &Input{}
+}
+
+// Read reads from the input in a thread-safe way.
+func (i *Input) Read(p []byte) (n int, err error) {
+	i.Mutex.Lock()
+	defer i.Mutex.Unlock()
+	if i.src == nil {
+		return 0, io.EOF
+	}
+	return i.src.Read(p)
+}
+
+// Add attaches a new source to the input.
+// Add can only be called once per input. Subsequent calls will
+// return an error.
+func (i *Input) Add(src io.Reader) error {
+	i.Mutex.Lock()
+	defer i.Mutex.Unlock()
+	if i.src != nil {
+		return fmt.Errorf("Maximum number of sources reached: 1")
+	}
+	i.src = src
+	return nil
+}
+
+// Tail reads from `src` line per line, and returns the last `n` lines as an array.
+// A ring buffer is used to only store `n` lines at any time.
+func Tail(src io.Reader, n int, dst *[]string) {
+	scanner := bufio.NewScanner(src)
+	r := ring.New(n)
+	for scanner.Scan() {
+		if n == 0 {
+			continue
+		}
+		r.Value = scanner.Text()
+		r = r.Next()
+	}
+	r.Do(func(v interface{}) {
+		if v == nil {
+			return
+		}
+		*dst = append(*dst, v.(string))
+	})
+}