diff --git a/integration-cli/docker_cli_import_test.go b/integration-cli/docker_cli_import_test.go index 70f6f05cca..14a2c2c08f 100644 --- a/integration-cli/docker_cli_import_test.go +++ b/integration-cli/docker_cli_import_test.go @@ -1,7 +1,6 @@ package main import ( - "bufio" "compress/gzip" "context" "os" @@ -66,7 +65,7 @@ func (s *DockerCLIImportSuite) TestImportFile(c *testing.T) { icmd.RunCmd(icmd.Cmd{ Command: []string{dockerBinary, "export", "test-import"}, - Stdout: bufio.NewWriter(temporaryFile), + Stdout: temporaryFile, }).Assert(c, icmd.Success) out := cli.DockerCmd(c, "import", temporaryFile.Name()).Combined() @@ -110,7 +109,7 @@ func (s *DockerCLIImportSuite) TestImportFileWithMessage(c *testing.T) { icmd.RunCmd(icmd.Cmd{ Command: []string{dockerBinary, "export", "test-import"}, - Stdout: bufio.NewWriter(temporaryFile), + Stdout: temporaryFile, }).Assert(c, icmd.Success) message := "Testing commit message" @@ -144,7 +143,7 @@ func (s *DockerCLIImportSuite) TestImportWithQuotedChanges(c *testing.T) { assert.Assert(c, err == nil, "failed to create temporary file") defer os.Remove(temporaryFile.Name()) - cli.Docker(cli.Args("export", "test-import"), cli.WithStdout(bufio.NewWriter(temporaryFile))).Assert(c, icmd.Success) + cli.Docker(cli.Args("export", "test-import"), cli.WithStdout(temporaryFile)).Assert(c, icmd.Success) result := cli.DockerCmd(c, "import", "-c", `ENTRYPOINT ["/bin/sh", "-c"]`, temporaryFile.Name()) imgRef := strings.TrimSpace(result.Stdout()) diff --git a/pkg/archive/archive.go b/pkg/archive/archive.go index 365e8f18ed..b7eae21328 100644 --- a/pkg/archive/archive.go +++ b/pkg/archive/archive.go @@ -19,13 +19,13 @@ import ( "runtime/debug" "strconv" "strings" + "sync" "sync/atomic" "syscall" "time" "github.com/containerd/log" "github.com/docker/docker/pkg/idtools" - "github.com/docker/docker/pkg/pools" "github.com/klauspost/compress/zstd" "github.com/moby/patternmatcher" "github.com/moby/sys/sequential" @@ -230,13 +230,51 @@ func (r *readCloserWrapper) Close() error { return nil } - return r.closer() + if r.closer != nil { + return r.closer() + } + return nil +} + +var ( + bufioReader32KPool = &sync.Pool{ + New: func() interface{} { return bufio.NewReaderSize(nil, 32*1024) }, + } +) + +type bufferedReader struct { + buf *bufio.Reader +} + +func newBufferedReader(r io.Reader) *bufferedReader { + buf := bufioReader32KPool.Get().(*bufio.Reader) + buf.Reset(r) + return &bufferedReader{buf} +} + +func (r *bufferedReader) Read(p []byte) (n int, err error) { + if r.buf == nil { + return 0, io.EOF + } + n, err = r.buf.Read(p) + if err == io.EOF { + r.buf.Reset(nil) + bufioReader32KPool.Put(r.buf) + r.buf = nil + } + return +} + +func (r *bufferedReader) Peek(n int) ([]byte, error) { + if r.buf == nil { + return nil, io.EOF + } + return r.buf.Peek(n) } // DecompressStream decompresses the archive and returns a ReaderCloser with the decompressed archive. func DecompressStream(archive io.Reader) (io.ReadCloser, error) { - p := pools.BufioReader32KPool - buf := p.Get(archive) + buf := newBufferedReader(archive) bs, err := buf.Peek(10) if err != nil && err != io.EOF { // Note: we'll ignore any io.EOF error because there are some odd @@ -248,26 +286,12 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) { return nil, err } - wrapReader := func(r io.Reader, cancel context.CancelFunc) io.ReadCloser { - return &readCloserWrapper{ - Reader: r, - closer: func() error { - if cancel != nil { - cancel() - } - if readCloser, ok := r.(io.ReadCloser); ok { - readCloser.Close() - } - p.Put(buf) - return nil - }, - } - } - compression := DetectCompression(bs) switch compression { case Uncompressed: - return wrapReader(buf, nil), nil + return &readCloserWrapper{ + Reader: buf, + }, nil case Gzip: ctx, cancel := context.WithCancel(context.Background()) @@ -276,10 +300,18 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) { cancel() return nil, err } - return wrapReader(gzReader, cancel), nil + return &readCloserWrapper{ + Reader: gzReader, + closer: func() error { + cancel() + return gzReader.Close() + }, + }, nil case Bzip2: bz2Reader := bzip2.NewReader(buf) - return wrapReader(bz2Reader, nil), nil + return &readCloserWrapper{ + Reader: bz2Reader, + }, nil case Xz: ctx, cancel := context.WithCancel(context.Background()) @@ -288,30 +320,44 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) { cancel() return nil, err } - return wrapReader(xzReader, cancel), nil + + return &readCloserWrapper{ + Reader: xzReader, + closer: func() error { + cancel() + return xzReader.Close() + }, + }, nil case Zstd: zstdReader, err := zstd.NewReader(buf) if err != nil { return nil, err } - return wrapReader(zstdReader, nil), nil + return &readCloserWrapper{ + Reader: zstdReader, + closer: func() error { + zstdReader.Close() + return nil + }, + }, nil default: return nil, fmt.Errorf("Unsupported compression format %s", (&compression).Extension()) } } +type nopWriteCloser struct { + io.Writer +} + +func (nopWriteCloser) Close() error { return nil } + // CompressStream compresses the dest with specified compression algorithm. func CompressStream(dest io.Writer, compression Compression) (io.WriteCloser, error) { - p := pools.BufioWriter32KPool - buf := p.Get(dest) switch compression { case Uncompressed: - writeBufWrapper := p.NewWriteCloserWrapper(buf, buf) - return writeBufWrapper, nil + return nopWriteCloser{dest}, nil case Gzip: - gzWriter := gzip.NewWriter(dest) - writeBufWrapper := p.NewWriteCloserWrapper(buf, gzWriter) - return writeBufWrapper, nil + return gzip.NewWriter(dest), nil case Bzip2, Xz: // archive/bzip2 does not support writing, and there is no xz support at all // However, this is not a problem as docker only currently generates gzipped tars @@ -382,7 +428,7 @@ func ReplaceFileTarWrapper(inputTarStream io.ReadCloser, mods map[string]TarModi pipeWriter.CloseWithError(err) return } - if _, err := pools.Copy(tarWriter, tarReader); err != nil { + if _, err := copyWithBuffer(tarWriter, tarReader); err != nil { pipeWriter.CloseWithError(err) return } @@ -529,7 +575,6 @@ type tarWhiteoutConverter interface { type tarAppender struct { TarWriter *tar.Writer - Buffer *bufio.Writer // for hardlink mapping SeenFiles map[uint64]string @@ -547,7 +592,6 @@ func newTarAppender(idMapping idtools.IdentityMapping, writer io.Writer, chownOp return &tarAppender{ SeenFiles: make(map[uint64]string), TarWriter: tar.NewWriter(writer), - Buffer: pools.BufioWriter32KPool.Get(nil), IdentityMapping: idMapping, ChownOpts: chownOpts, } @@ -665,17 +709,11 @@ func (ta *tarAppender) addTarFile(path, name string) error { return err } - ta.Buffer.Reset(ta.TarWriter) - defer ta.Buffer.Reset(nil) - _, err = pools.Copy(ta.Buffer, file) + _, err = copyWithBuffer(ta.TarWriter, file) file.Close() if err != nil { return err } - err = ta.Buffer.Flush() - if err != nil { - return err - } } return nil @@ -718,7 +756,7 @@ func createTarFile(path, extractDir string, hdr *tar.Header, reader io.Reader, o if err != nil { return err } - if _, err := pools.Copy(file, reader); err != nil { + if _, err := copyWithBuffer(file, reader); err != nil { file.Close() return err } @@ -929,9 +967,6 @@ func (t *Tarballer) Do() { } }() - // this buffer is needed for the duration of this piped stream - defer pools.BufioWriter32KPool.Put(ta.Buffer) - // In general we log errors here but ignore them because // during e.g. a diff operation the container can continue // mutating the filesystem and we can see transient errors @@ -1087,8 +1122,6 @@ func (t *Tarballer) Do() { // Unpack unpacks the decompressedArchive to dest with options. func Unpack(decompressedArchive io.Reader, dest string, options *TarOptions) error { tr := tar.NewReader(decompressedArchive) - trBuf := pools.BufioReader32KPool.Get(nil) - defer pools.BufioReader32KPool.Put(trBuf) var dirs []*tar.Header whiteoutConverter := getWhiteoutConverter(options.WhiteoutFormat) @@ -1165,7 +1198,6 @@ loop: } } } - trBuf.Reset(tr) if err := remapIDs(options.IDMap, hdr); err != nil { return err @@ -1181,7 +1213,7 @@ loop: } } - if err := createTarFile(path, dest, hdr, trBuf, options); err != nil { + if err := createTarFile(path, dest, hdr, tr, options); err != nil { return err } @@ -1384,7 +1416,7 @@ func (archiver *Archiver) CopyFileWithTar(src, dst string) (err error) { if err := tw.WriteHeader(hdr); err != nil { return err } - if _, err := pools.Copy(tw, srcF); err != nil { + if _, err := copyWithBuffer(tw, srcF); err != nil { return err } return nil diff --git a/pkg/archive/archive_test.go b/pkg/archive/archive_test.go index 4b93851853..078077eb4e 100644 --- a/pkg/archive/archive_test.go +++ b/pkg/archive/archive_test.go @@ -693,7 +693,7 @@ func tarUntar(t *testing.T, origin string, options *TarOptions) ([]Change, error defer archive.Close() buf := make([]byte, 10) - if _, err := archive.Read(buf); err != nil { + if _, err := io.ReadFull(archive, buf); err != nil { return nil, err } wrap := io.MultiReader(bytes.NewReader(buf), archive) diff --git a/pkg/archive/changes.go b/pkg/archive/changes.go index 423605fbb7..79c810a681 100644 --- a/pkg/archive/changes.go +++ b/pkg/archive/changes.go @@ -15,7 +15,6 @@ import ( "github.com/containerd/log" "github.com/docker/docker/pkg/idtools" - "github.com/docker/docker/pkg/pools" ) // ChangeType represents the change type. @@ -389,9 +388,6 @@ func ExportChanges(dir string, changes []Change, idMap idtools.IdentityMapping) go func() { ta := newTarAppender(idMap, writer, nil) - // this buffer is needed for the duration of this piped stream - defer pools.BufioWriter32KPool.Put(ta.Buffer) - sort.Sort(changesByPath(changes)) // In general we log errors here but ignore them because diff --git a/pkg/archive/copy.go b/pkg/archive/copy.go index ee4020c2a2..cddf18ecdb 100644 --- a/pkg/archive/copy.go +++ b/pkg/archive/copy.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "strings" + "sync" "github.com/containerd/log" ) @@ -20,6 +21,17 @@ var ( ErrInvalidCopySource = errors.New("invalid copy source content") ) +var copyPool = sync.Pool{ + New: func() interface{} { s := make([]byte, 32*1024); return &s }, +} + +func copyWithBuffer(dst io.Writer, src io.Reader) (written int64, err error) { + buf := copyPool.Get().(*[]byte) + written, err = io.CopyBuffer(dst, src, *buf) + copyPool.Put(buf) + return +} + // PreserveTrailingDotOrSeparator returns the given cleaned path (after // processing using any utility functions from the path or filepath stdlib // packages) and appends a trailing `/.` or `/` if its corresponding original diff --git a/pkg/archive/diff.go b/pkg/archive/diff.go index 6a05643ab6..d5a394cdc9 100644 --- a/pkg/archive/diff.go +++ b/pkg/archive/diff.go @@ -11,7 +11,6 @@ import ( "strings" "github.com/containerd/log" - "github.com/docker/docker/pkg/pools" ) // UnpackLayer unpack `layer` to a `dest`. The stream `layer` can be @@ -19,8 +18,6 @@ import ( // Returns the size in bytes of the contents of the layer. func UnpackLayer(dest string, layer io.Reader, options *TarOptions) (size int64, err error) { tr := tar.NewReader(layer) - trBuf := pools.BufioReader32KPool.Get(tr) - defer pools.BufioReader32KPool.Put(trBuf) var dirs []*tar.Header unpackedPaths := make(map[string]struct{}) @@ -159,8 +156,7 @@ func UnpackLayer(dest string, layer io.Reader, options *TarOptions) (size int64, } } - trBuf.Reset(tr) - srcData := io.Reader(trBuf) + srcData := io.Reader(tr) srcHdr := hdr // Hard links into /.wh..wh.plnk don't work, as we don't extract that directory, so diff --git a/pkg/chrootarchive/archive_unix_test.go b/pkg/chrootarchive/archive_unix_test.go index 8a4f1938d3..c7a0b5f81b 100644 --- a/pkg/chrootarchive/archive_unix_test.go +++ b/pkg/chrootarchive/archive_unix_test.go @@ -67,6 +67,8 @@ func TestUntarWithMaliciousSymlinks(t *testing.T) { assert.NilError(t, err) assert.Equal(t, string(hostData), "I am a host file") + io.Copy(io.Discard, tee) + // Now test by chrooting to an attacker controlled path // This should succeed as is and overwrite a "host" file // Note that this would be a mis-use of this function.