TBR lib: Change gosh.Cmd.StdinPipe to have unlimited-size.
It's more convenient for users if StdinPipe has an unlimited
size; otherwise they need to worry about deadlock. You might
think we could just use our buffered pipe and be done with it,
but it's pretty tricky. In particular os/exec will create its
own os.Pipe and a copier goroutine if we don't pass in a *os.File
for exec.Cmd.Stdin, and exec.Cmd.Wait will wait for that
goroutine to finish.
Note that we can't call exec.Cmd.StdinPipe; that ensures that
Close is only called once, but doesn't ensure Write and Close
calls are synchronous. This makes a race possible: the user may
call Write on the StdinPipe, concurrently with the exit-waiter
goroutine calling Close on the StdinPipe when the process exits.
This race has shown up on jenkins, and is also easily
reproduced (under heavy machine load) with the new test.
To make this all work, we create our own os.Pipe that we set for
exec.Cmd.Stdin, and also create a buffered pipe, along with our
own copier goroutine.
MultiPart: 1/3
Change-Id: I3a7ed86164d1c4427ba55a88dac559f34fdd23b5
diff --git a/gosh/.api b/gosh/.api
index d805d75..c48f046 100644
--- a/gosh/.api
+++ b/gosh/.api
@@ -1,6 +1,5 @@
pkg gosh, func InitChildMain()
pkg gosh, func InitMain()
-pkg gosh, func NewBufferedPipe() io.ReadWriteCloser
pkg gosh, func NewShell(Opts) *Shell
pkg gosh, func NopWriteCloser(io.Writer) io.WriteCloser
pkg gosh, func RegisterFunc(string, interface{}) *Func
@@ -12,12 +11,13 @@
pkg gosh, method (*Cmd) CombinedOutput() string
pkg gosh, method (*Cmd) Pid() int
pkg gosh, method (*Cmd) Run()
+pkg gosh, method (*Cmd) SetStdinReader(io.Reader)
pkg gosh, method (*Cmd) Signal(os.Signal)
pkg gosh, method (*Cmd) Start()
-pkg gosh, method (*Cmd) StderrPipe() io.Reader
+pkg gosh, method (*Cmd) StderrPipe() io.ReadCloser
pkg gosh, method (*Cmd) StdinPipe() io.WriteCloser
pkg gosh, method (*Cmd) Stdout() string
-pkg gosh, method (*Cmd) StdoutPipe() io.Reader
+pkg gosh, method (*Cmd) StdoutPipe() io.ReadCloser
pkg gosh, method (*Cmd) StdoutStderr() (string, string)
pkg gosh, method (*Cmd) Terminate(os.Signal)
pkg gosh, method (*Cmd) Wait()
@@ -43,7 +43,6 @@
pkg gosh, type Cmd struct, OutputDir string
pkg gosh, type Cmd struct, Path string
pkg gosh, type Cmd struct, PropagateOutput bool
-pkg gosh, type Cmd struct, Stdin string
pkg gosh, type Cmd struct, Vars map[string]string
pkg gosh, type Func struct
pkg gosh, type Opts struct
diff --git a/gosh/buffered_pipe.go b/gosh/buffered_pipe.go
index b7cd44b..d9e05fd 100644
--- a/gosh/buffered_pipe.go
+++ b/gosh/buffered_pipe.go
@@ -16,10 +16,10 @@
closed bool
}
-// NewBufferedPipe returns a new thread-safe pipe backed by an unbounded
+// newBufferedPipe returns a new thread-safe pipe backed by an unbounded
// in-memory buffer. Writes on the pipe never block; reads on the pipe block
// until data is available.
-func NewBufferedPipe() io.ReadWriteCloser {
+func newBufferedPipe() io.ReadWriteCloser {
return &bufferedPipe{cond: sync.NewCond(&sync.Mutex{})}
}
diff --git a/gosh/buffered_pipe_test.go b/gosh/buffered_pipe_test.go
index 593f77b..1dcca26 100644
--- a/gosh/buffered_pipe_test.go
+++ b/gosh/buffered_pipe_test.go
@@ -2,17 +2,26 @@
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
-package gosh_test
+package gosh
import (
+ "io/ioutil"
"testing"
-
- "v.io/x/lib/gosh"
)
func TestReadAfterClose(t *testing.T) {
- p := gosh.NewBufferedPipe()
- p.Write([]byte("foo"))
- p.Close()
- eq(t, toString(t, p), "foo")
+ p := newBufferedPipe()
+ if _, err := p.Write([]byte("foo")); err != nil {
+ t.Errorf("write failed: %v", err)
+ }
+ if err := p.Close(); err != nil {
+ t.Errorf("close failed: %v", err)
+ }
+ b, err := ioutil.ReadAll(p)
+ if err != nil {
+ t.Errorf("read failed: %v", err)
+ }
+ if got, want := string(b), "foo"; got != want {
+ t.Errorf("got %s, want %s", got, want)
+ }
}
diff --git a/gosh/cmd.go b/gosh/cmd.go
index e748670..e956500 100644
--- a/gosh/cmd.go
+++ b/gosh/cmd.go
@@ -12,14 +12,15 @@
"os"
"os/exec"
"path/filepath"
- "strings"
"sync"
+ "syscall"
"time"
)
var (
errAlreadyCalledStart = errors.New("gosh: already called Cmd.Start")
errAlreadyCalledWait = errors.New("gosh: already called Cmd.Wait")
+ errAlreadySetStdin = errors.New("gosh: already set stdin")
errCloseStdout = errors.New("gosh: use NopWriteCloser(os.Stdout) to prevent stdout from being closed")
errCloseStderr = errors.New("gosh: use NopWriteCloser(os.Stderr) to prevent stderr from being closed")
errDidNotCallStart = errors.New("gosh: did not call Cmd.Start")
@@ -55,22 +56,21 @@
// ExitErrorIsOk specifies whether an *exec.ExitError should be reported via
// Shell.HandleError.
ExitErrorIsOk bool
- // Stdin is a string to write to the child's stdin.
- Stdin string
// Internal state.
- sh *Shell
- c *exec.Cmd
- stdinWriteCloser io.WriteCloser // from exec.Cmd.StdinPipe
- calledStart bool
- calledWait bool
- cond *sync.Cond
- waitChan chan error
- started bool // protected by sh.cleanupMu
- exited bool // protected by cond.L
- stdoutWriters []io.Writer
- stderrWriters []io.Writer
- closers []io.Closer
- recvVars map[string]string // protected by cond.L
+ sh *Shell
+ c *exec.Cmd
+ calledStart bool
+ calledWait bool
+ cond *sync.Cond
+ waitChan chan error
+ stdinDoneChan chan error
+ started bool // protected by sh.cleanupMu
+ exited bool // protected by cond.L
+ stdoutWriters []io.Writer
+ stderrWriters []io.Writer
+ afterStartClosers []io.Closer
+ afterWaitClosers []io.Closer
+ recvVars map[string]string // protected by cond.L
}
// Clone returns a new Cmd with a copy of this Cmd's configuration.
@@ -81,12 +81,11 @@
return res
}
-// StdinPipe returns a thread-safe WriteCloser backed by a buffered pipe for the
-// command's stdin. The returned pipe will be closed when the process exits, but
-// may also be closed earlier by the caller, e.g. if the command does not exit
-// until its stdin is closed. Must be called before Start. It is safe to call
-// StdinPipe multiple times; calls after the first return the pipe created by
-// the first call.
+// StdinPipe returns a WriteCloser backed by an unlimited-size pipe for the
+// command's stdin. The pipe will be closed when the process exits, but may also
+// be closed earlier by the caller, e.g. if the command does not exit until its
+// stdin is closed. Must be called before Start. Only one call may be made to
+// StdinPipe or SetStdinReader; subsequent calls will fail.
func (c *Cmd) StdinPipe() io.WriteCloser {
c.sh.Ok()
res, err := c.stdinPipe()
@@ -94,26 +93,34 @@
return res
}
-// StdoutPipe returns a Reader backed by a buffered pipe for the command's
-// stdout. Must be called before Start. May be called more than once; each
-// invocation creates a new pipe.
-func (c *Cmd) StdoutPipe() io.Reader {
+// StdoutPipe returns a ReadCloser backed by an unlimited-size pipe for the
+// command's stdout. Must be called before Start. May be called more than once;
+// each invocation creates a new pipe.
+func (c *Cmd) StdoutPipe() io.ReadCloser {
c.sh.Ok()
res, err := c.stdoutPipe()
c.handleError(err)
return res
}
-// StderrPipe returns a Reader backed by a buffered pipe for the command's
-// stderr. Must be called before Start. May be called more than once; each
-// invocation creates a new pipe.
-func (c *Cmd) StderrPipe() io.Reader {
+// StderrPipe returns a ReadCloser backed by an unlimited-size pipe for the
+// command's stderr. Must be called before Start. May be called more than once;
+// each invocation creates a new pipe.
+func (c *Cmd) StderrPipe() io.ReadCloser {
c.sh.Ok()
res, err := c.stderrPipe()
c.handleError(err)
return res
}
+// SetStdinReader configures this Cmd to read the child's stdin from the given
+// Reader. Must be called before Start. Only one call may be made to StdinPipe
+// or SetStdinReader; subsequent calls will fail.
+func (c *Cmd) SetStdinReader(r io.Reader) {
+ c.sh.Ok()
+ c.handleError(c.setStdinReader(r))
+}
+
// AddStdoutWriter configures this Cmd to tee the child's stdout to the given
// WriteCloser, which will be closed when the process exits.
//
@@ -271,18 +278,6 @@
}
}
-func (c *Cmd) closeClosers() {
- // If the same WriteCloser was passed to both AddStdoutWriter and
- // AddStderrWriter, we should only close it once.
- cm := map[io.Closer]bool{}
- for _, c := range c.closers {
- if !cm[c] {
- cm[c] = true
- c.Close() // best-effort; ignore returned error
- }
- }
-}
-
func (c *Cmd) isRunning() bool {
if !c.started {
return false
@@ -339,18 +334,6 @@
return len(p), nil
}
-type lockedWriter struct {
- mu *sync.Mutex
- w io.Writer
-}
-
-func (w lockedWriter) Write(p []byte) (int, error) {
- w.mu.Lock()
- n, err := w.w.Write(p)
- w.mu.Unlock()
- return n, err
-}
-
func (c *Cmd) makeStdoutStderr() (io.Writer, io.Writer, error) {
c.stderrWriters = append(c.stderrWriters, &recvWriter{c: c})
if c.PropagateOutput {
@@ -366,14 +349,14 @@
return nil, nil, err
default:
c.stdoutWriters = append(c.stdoutWriters, file)
- c.closers = append(c.closers, file)
+ c.afterWaitClosers = append(c.afterWaitClosers, file)
}
switch file, err := os.OpenFile(name+".stderr", flags, 0600); {
case err != nil:
return nil, nil, err
default:
c.stderrWriters = append(c.stderrWriters, file)
- c.closers = append(c.closers, file)
+ c.afterWaitClosers = append(c.afterWaitClosers, file)
}
}
switch hasOut, hasErr := len(c.stdoutWriters) > 0, len(c.stderrWriters) > 0; {
@@ -382,8 +365,8 @@
// writers that capture both will see the same ordering, and don't need to
// worry about concurrent writes.
sharedMu := &sync.Mutex{}
- stdout := lockedWriter{sharedMu, io.MultiWriter(c.stdoutWriters...)}
- stderr := lockedWriter{sharedMu, io.MultiWriter(c.stderrWriters...)}
+ stdout := &sharedLockWriter{sharedMu, io.MultiWriter(c.stdoutWriters...)}
+ stderr := &sharedLockWriter{sharedMu, io.MultiWriter(c.stderrWriters...)}
return stdout, stderr, nil
case hasOut:
return io.MultiWriter(c.stdoutWriters...), nil, nil
@@ -393,6 +376,18 @@
return nil, nil, nil
}
+type sharedLockWriter struct {
+ mu *sync.Mutex
+ w io.Writer
+}
+
+func (w *sharedLockWriter) Write(p []byte) (int, error) {
+ w.mu.Lock()
+ n, err := w.w.Write(p)
+ w.mu.Unlock()
+ return n, err
+}
+
func (c *Cmd) clone() (*Cmd, error) {
args := make([]string, len(c.Args))
copy(args, c.Args)
@@ -405,39 +400,83 @@
res.PropagateOutput = c.PropagateOutput
res.OutputDir = c.OutputDir
res.ExitErrorIsOk = c.ExitErrorIsOk
- res.Stdin = c.Stdin
return res, nil
}
func (c *Cmd) stdinPipe() (io.WriteCloser, error) {
- if c.calledStart {
+ switch {
+ case c.calledStart:
return nil, errAlreadyCalledStart
+ case c.c.Stdin != nil:
+ return nil, errAlreadySetStdin
}
- if c.stdinWriteCloser != nil {
- return c.stdinWriteCloser, nil
+ // We want to provide an unlimited-size pipe to the user. If we set
+ // c.c.Stdin directly to the newBufferedPipe, the os/exec package will
+ // create an os.Pipe for us, along with a goroutine to copy data over. And
+ // exec.Cmd.Wait will wait for this goroutine to exit before returning, even
+ // if the process has already exited. That means the user will be forced to
+ // call Close on the returned WriteCloser, which is annoying.
+ //
+ // Instead, we set c.c.Stdin to our own os.Pipe, so that os/exec won't create
+ // the pipe nor the goroutine. We chain our newBufferedPipe in front of this,
+ // with our own copier goroutine. This gives the user a pipe that never blocks
+ // on Write, and which they don't need to Close if the process exits.
+ pr, pw, err := os.Pipe()
+ if err != nil {
+ return nil, err
}
- var err error
- c.stdinWriteCloser, err = c.c.StdinPipe()
- return c.stdinWriteCloser, err
+ c.c.Stdin = pr
+ c.afterStartClosers = append(c.afterStartClosers, pr)
+ bp := newBufferedPipe()
+ c.afterWaitClosers = append(c.afterWaitClosers, bp)
+ c.stdinDoneChan = make(chan error, 1)
+ go c.stdinPipeCopier(pw, bp) // pw is closed by stdinPipeCopier
+ return bp, nil
}
-func (c *Cmd) stdoutPipe() (io.Reader, error) {
+func (c *Cmd) stdinPipeCopier(dst io.WriteCloser, src io.Reader) {
+ var firstErr error
+ _, err := io.Copy(dst, src)
+ // Ignore EPIPE errors copying to stdin, indicating the process exited. This
+ // mirrors logic in os/exec/exec_posix.go. Also see:
+ // https://github.com/golang/go/issues/9173
+ if pe, ok := err.(*os.PathError); !ok || pe.Op != "write" || pe.Path != "|1" || pe.Err != syscall.EPIPE {
+ firstErr = err
+ }
+ if err := dst.Close(); err != nil && firstErr == nil {
+ firstErr = err
+ }
+ c.stdinDoneChan <- firstErr
+}
+
+func (c *Cmd) setStdinReader(r io.Reader) error {
+ switch {
+ case c.calledStart:
+ return errAlreadyCalledStart
+ case c.c.Stdin != nil:
+ return errAlreadySetStdin
+ }
+ c.c.Stdin = r
+ return nil
+}
+
+func (c *Cmd) stdoutPipe() (io.ReadCloser, error) {
if c.calledStart {
return nil, errAlreadyCalledStart
}
- p := NewBufferedPipe()
+ p := newBufferedPipe()
c.stdoutWriters = append(c.stdoutWriters, p)
- c.closers = append(c.closers, p)
+ c.afterWaitClosers = append(c.afterWaitClosers, p)
return p, nil
}
-func (c *Cmd) stderrPipe() (io.Reader, error) {
+func (c *Cmd) stderrPipe() (io.ReadCloser, error) {
if c.calledStart {
return nil, errAlreadyCalledStart
}
- p := NewBufferedPipe()
+ p := newBufferedPipe()
c.stderrWriters = append(c.stderrWriters, p)
- c.closers = append(c.closers, p)
+ c.afterWaitClosers = append(c.afterWaitClosers, p)
return p, nil
}
@@ -451,7 +490,7 @@
return errCloseStderr
}
c.stdoutWriters = append(c.stdoutWriters, wc)
- c.closers = append(c.closers, wc)
+ c.afterWaitClosers = append(c.afterWaitClosers, wc)
return nil
}
@@ -465,7 +504,7 @@
return errCloseStderr
}
c.stderrWriters = append(c.stderrWriters, wc)
- c.closers = append(c.closers, wc)
+ c.afterWaitClosers = append(c.afterWaitClosers, wc)
return nil
}
@@ -474,8 +513,9 @@
func (c *Cmd) start() error {
defer func() {
+ closeClosers(c.afterStartClosers)
if !c.started {
- c.closeClosers()
+ closeClosers(c.afterWaitClosers)
}
}()
if c.calledStart {
@@ -504,12 +544,6 @@
}
c.c.Env = mapToSlice(vars)
c.c.Args = c.Args
- if c.Stdin != "" {
- if c.stdinWriteCloser != nil {
- return errors.New("gosh: cannot both set Stdin and call StdinPipe")
- }
- c.c.Stdin = strings.NewReader(c.Stdin)
- }
var err error
if c.c.Stdout, c.c.Stderr, err = c.makeStdoutStderr(); err != nil {
return err
@@ -534,11 +568,29 @@
c.exited = true
c.cond.Signal()
c.cond.L.Unlock()
- c.closeClosers()
+ closeClosers(c.afterWaitClosers)
+ if c.stdinDoneChan != nil {
+ // Wait for the stdinPipeCopier goroutine to finish.
+ if err := <-c.stdinDoneChan; waitErr == nil {
+ waitErr = err
+ }
+ }
c.waitChan <- waitErr
}()
}
+func closeClosers(closers []io.Closer) {
+ // If the same WriteCloser was passed to both AddStdoutWriter and
+ // AddStderrWriter, we should only close it once.
+ cm := map[io.Closer]bool{}
+ for _, closer := range closers {
+ if !cm[closer] {
+ cm[closer] = true
+ closer.Close() // best-effort; ignore returned error
+ }
+ }
+}
+
// TODO(sadovsky): Maybe add optional timeouts for Cmd.{awaitVars,wait}.
func (c *Cmd) awaitVars(keys ...string) (map[string]string, error) {
diff --git a/gosh/shell_test.go b/gosh/shell_test.go
index c497498..d8d6930 100644
--- a/gosh/shell_test.go
+++ b/gosh/shell_test.go
@@ -23,6 +23,7 @@
"path/filepath"
"reflect"
"runtime/debug"
+ "strings"
"testing"
"time"
@@ -369,38 +370,67 @@
sh := gosh.NewShell(gosh.Opts{Fatalf: makeFatalf(t), Logf: t.Logf})
defer sh.Cleanup()
+ // The "cat" command exits after the reader returns EOF.
c := sh.FuncCmd(catFunc)
- c.Stdin = "foo\n"
- // We set c.Stdin and did not call c.StdinPipe(), so stdin should close and
- // cat should exit immediately.
+ c.SetStdinReader(strings.NewReader("foo\n"))
eq(t, c.Stdout(), "foo\n")
+ // The "cat" command exits after the reader returns EOF, so we must explicitly
+ // close the stdin pipe.
c = sh.FuncCmd(catFunc)
- c.StdinPipe().Write([]byte("foo\n"))
- // The "cat" command only exits when stdin is closed, so we must explicitly
- // close the stdin pipe. Note, it's safe to call c.StdinPipe multiple times.
- c.StdinPipe().Close()
+ stdin := c.StdinPipe()
+ stdin.Write([]byte("foo\n"))
+ stdin.Close()
eq(t, c.Stdout(), "foo\n")
+ // The "read" command exits when it sees a newline, so it is not necessary to
+ // explicitly close the stdin pipe.
c = sh.FuncCmd(readFunc)
- c.StdinPipe().Write([]byte("foo\n"))
- // The "read" command exits when it sees a newline, so Cmd.Wait (and thus
- // Cmd.Run) should return immediately; it should not be necessary to close the
- // stdin pipe.
+ stdin = c.StdinPipe()
+ stdin.Write([]byte("foo\n"))
c.Run()
- c = sh.FuncCmd(catFunc)
// No stdin, so cat should exit immediately.
+ c = sh.FuncCmd(catFunc)
eq(t, c.Stdout(), "")
- // It's an error (detected at command start time) to both set c.Stdin and call
- // c.StdinPipe. Note, this indirectly tests that Shell.Cleanup works even if
- // some Cmd.Start failed.
+ // It's an error to call both StdinPipe and SetStdinReader.
c = sh.FuncCmd(catFunc)
- c.Stdin = "foo"
- c.StdinPipe().Write([]byte("bar"))
- c.StdinPipe().Close()
- setsErr(t, sh, func() { c.Start() })
+ c.StdinPipe()
+ setsErr(t, sh, func() { c.StdinPipe() })
+
+ c = sh.FuncCmd(catFunc)
+ c.StdinPipe()
+ setsErr(t, sh, func() { c.SetStdinReader(strings.NewReader("")) })
+
+ c = sh.FuncCmd(catFunc)
+ c.SetStdinReader(strings.NewReader(""))
+ setsErr(t, sh, func() { c.StdinPipe() })
+
+ c = sh.FuncCmd(catFunc)
+ c.SetStdinReader(strings.NewReader(""))
+ setsErr(t, sh, func() { c.SetStdinReader(strings.NewReader("")) })
+}
+
+func TestStdinPipeWriteUntilExit(t *testing.T) {
+ sh := gosh.NewShell(gosh.Opts{Fatalf: makeFatalf(t), Logf: t.Logf})
+ defer sh.Cleanup()
+
+ // Ensure that Write calls on stdin fail after the process exits. Note that we
+ // write to the command's stdin concurrently with the command's exit waiter
+ // goroutine closing stdin. Use "go test -race" catch races.
+ //
+ // Set a non-zero exit code, so that os.Exit exits immediately. See the
+ // implementation of https://golang.org/pkg/os/#Exit for details.
+ c := sh.FuncCmd(exitFunc, 1)
+ c.ExitErrorIsOk = true
+ stdin := c.StdinPipe()
+ c.Start()
+ for {
+ if _, err := stdin.Write([]byte("a")); err != nil {
+ return
+ }
+ }
}
var writeFunc = gosh.RegisterFunc("writeFunc", func(stdout, stderr bool) error {
@@ -565,30 +595,6 @@
eq(t, wc.count, 1)
}
-// Tests piping from one Cmd's stdout/stderr to another's stdin. It should be
-// possible to wait on just the last Cmd.
-func TestPiping(t *testing.T) {
- sh := gosh.NewShell(gosh.Opts{Fatalf: makeFatalf(t), Logf: t.Logf})
- defer sh.Cleanup()
-
- echo := sh.FuncCmd(echoFunc)
- echo.Args = append(echo.Args, "foo")
- cat := sh.FuncCmd(catFunc)
- echo.AddStdoutWriter(cat.StdinPipe())
- echo.Start()
- eq(t, cat.Stdout(), "foo\n")
-
- // This time, pipe both stdout and stderr to cat's stdin.
- c := sh.FuncCmd(writeFunc, true, true)
- cat = sh.FuncCmd(catFunc)
- c.AddStdoutWriter(cat.StdinPipe())
- c.AddStderrWriter(cat.StdinPipe())
- c.Start()
- // Note, we can't assume any particular ordering of stdout and stderr, so we
- // simply check the length of the combined output.
- eq(t, len(cat.Stdout()), 4)
-}
-
func TestSignal(t *testing.T) {
sh := gosh.NewShell(gosh.Opts{Fatalf: makeFatalf(t), Logf: t.Logf})
defer sh.Cleanup()
@@ -653,7 +659,7 @@
d200 := 200 * time.Millisecond
c0 := sh.FuncCmd(sleepFunc, d0, 0) // not started
- c1 := sh.FuncCmd(sleepFunc, d0, 0) // failed to start
+ c1 := sh.Cmd("/#invalid#/!binary!") // failed to start
c2 := sh.FuncCmd(sleepFunc, d200, 0) // running and will succeed
c3 := sh.FuncCmd(sleepFunc, d200, 1) // running and will fail
c4 := sh.FuncCmd(sleepFunc, d0, 0) // succeeded
@@ -665,10 +671,9 @@
c6.ExitErrorIsOk = true
c7.ExitErrorIsOk = true
- // Configure the "failed to start" command.
- c1.StdinPipe()
- c1.Stdin = "foo"
- setsErr(t, sh, func() { c1.Start() })
+ // Make sure c1 fails to start. This indirectly tests that Shell.Cleanup works
+ // even if Cmd.Start failed.
+ setsErr(t, sh, c1.Start)
// Start commands, then wait for them to exit.
for _, c := range []*gosh.Cmd{c2, c3, c4, c5, c6, c7} {
