Merge pull request #16932 from c0b/fix-race-16924

Fix race #16924 [panic: runtime error: slice bounds out of range] docker daemon crash of racing

archive/archive.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/docker/pkg/fileutils"
 	"github.com/docker/docker/pkg/idtools"
+	"github.com/docker/docker/pkg/ioutils"
 	"github.com/docker/docker/pkg/pools"
 	"github.com/docker/docker/pkg/promise"
 	"github.com/docker/docker/pkg/system"
@@ -116,10 +117,10 @@ func DetectCompression(source []byte) Compression {
 	return Uncompressed
 }
 
-func xzDecompress(archive io.Reader) (io.ReadCloser, error) {
+func xzDecompress(archive io.Reader) (io.ReadCloser, <-chan struct{}, error) {
 	args := []string{"xz", "-d", "-c", "-q"}
 
-	return CmdStream(exec.Command(args[0], args[1:]...), archive)
+	return cmdStream(exec.Command(args[0], args[1:]...), archive)
 }
 
 // DecompressStream decompress the archive and returns a ReaderCloser with the decompressed archive.
@@ -148,12 +149,15 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) {
 		readBufWrapper := p.NewReadCloserWrapper(buf, bz2Reader)
 		return readBufWrapper, nil
 	case Xz:
-		xzReader, err := xzDecompress(buf)
+		xzReader, chdone, err := xzDecompress(buf)
 		if err != nil {
 			return nil, err
 		}
 		readBufWrapper := p.NewReadCloserWrapper(buf, xzReader)
-		return readBufWrapper, nil
+		return ioutils.NewReadCloserWrapper(readBufWrapper, func() error {
+			<-chdone
+			return readBufWrapper.Close()
+		}), nil
 	default:
 		return nil, fmt.Errorf("Unsupported compression format %s", (&compression).Extension())
 	}
@@ -925,57 +929,33 @@ func CopyFileWithTar(src, dst string) (err error) {
 	return defaultArchiver.CopyFileWithTar(src, dst)
 }
 
-// CmdStream executes a command, and returns its stdout as a stream.
+// cmdStream executes a command, and returns its stdout as a stream.
 // If the command fails to run or doesn't complete successfully, an error
 // will be returned, including anything written on stderr.
-func CmdStream(cmd *exec.Cmd, input io.Reader) (io.ReadCloser, error) {
+func cmdStream(cmd *exec.Cmd, input io.Reader) (io.ReadCloser, <-chan struct{}, error) {
-	if input != nil {
+	chdone := make(chan struct{})
-		stdin, err := cmd.StdinPipe()
+	cmd.Stdin = input
-		if err != nil {
-			return nil, err
-		}
-		// Write stdin if any
-		go func() {
-			io.Copy(stdin, input)
-			stdin.Close()
-		}()
-	}
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, err
-	}
-	stderr, err := cmd.StderrPipe()
-	if err != nil {
-		return nil, err
-	}
 	pipeR, pipeW := io.Pipe()
-	errChan := make(chan []byte)
+	cmd.Stdout = pipeW
-	// Collect stderr, we will use it in case of an error
+	var errBuf bytes.Buffer
-	go func() {
+	cmd.Stderr = &errBuf
-		errText, e := ioutil.ReadAll(stderr)
+
-		if e != nil {
+	// Run the command and return the pipe
-			errText = []byte("(...couldn't fetch stderr: " + e.Error() + ")")
+	if err := cmd.Start(); err != nil {
+		return nil, nil, err
 	}
-		errChan <- errText
+
-	}()
 	// Copy stdout to the returned pipe
 	go func() {
-		_, err := io.Copy(pipeW, stdout)
-		if err != nil {
-			pipeW.CloseWithError(err)
-		}
-		errText := <-errChan
 		if err := cmd.Wait(); err != nil {
-			pipeW.CloseWithError(fmt.Errorf("%s: %s", err, errText))
+			pipeW.CloseWithError(fmt.Errorf("%s: %s", err, errBuf.String()))
 		} else {
 			pipeW.Close()
 		}
+		close(chdone)
 	}()
-	// Run the command and return the pipe
+
-	if err := cmd.Start(); err != nil {
+	return pipeR, chdone, nil
-		return nil, err
-	}
-	return pipeR, nil
 }
 
 // NewTempArchive reads the content of src into a temporary file, and returns the contents

archive/archive_test.go

@@ -160,7 +160,7 @@ func TestExtensionXz(t *testing.T) {
 
 func TestCmdStreamLargeStderr(t *testing.T) {
 	cmd := exec.Command("/bin/sh", "-c", "dd if=/dev/zero bs=1k count=1000 of=/dev/stderr; echo hello")
-	out, err := CmdStream(cmd, nil)
+	out, _, err := cmdStream(cmd, nil)
 	if err != nil {
 		t.Fatalf("Failed to start command: %s", err)
 	}
@@ -181,7 +181,7 @@ func TestCmdStreamLargeStderr(t *testing.T) {
 
 func TestCmdStreamBad(t *testing.T) {
 	badCmd := exec.Command("/bin/sh", "-c", "echo hello; echo >&2 error couldn\\'t reverse the phase pulser; exit 1")
-	out, err := CmdStream(badCmd, nil)
+	out, _, err := cmdStream(badCmd, nil)
 	if err != nil {
 		t.Fatalf("Failed to start command: %s", err)
 	}
@@ -196,7 +196,7 @@ func TestCmdStreamBad(t *testing.T) {
 
 func TestCmdStreamGood(t *testing.T) {
 	cmd := exec.Command("/bin/sh", "-c", "echo hello; exit 0")
-	out, err := CmdStream(cmd, nil)
+	out, _, err := cmdStream(cmd, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

chrootarchive/archive_unix.go

@@ -8,6 +8,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"runtime"
 	"syscall"
@@ -79,6 +80,11 @@ func invokeUnpack(decompressedArchive io.Reader, dest string, options *archive.T
 	w.Close()
 
 	if err := cmd.Wait(); err != nil {
+		// when `xz -d -c -q | docker-untar ...` failed on docker-untar side,
+		// we need to exhaust `xz`'s output, otherwise the `xz` side will be
+		// pending on write pipe forever
+		io.Copy(ioutil.Discard, decompressedArchive)
+
 		return fmt.Errorf("Untar re-exec error: %v: output: %s", err, output)
 	}
 	return nil