From a93a079cb41fd6484725bcb3d775383a8a0a67f5 Mon Sep 17 00:00:00 2001 From: Derek McGowan Date: Mon, 16 Dec 2024 23:10:26 -0800 Subject: [PATCH 1/4] Remove use of pools in archive Signed-off-by: Derek McGowan --- pkg/archive/archive.go | 44 ++++++++++++------------------------- pkg/archive/archive_test.go | 2 +- pkg/archive/changes.go | 4 ---- pkg/archive/copy.go | 12 ++++++++++ pkg/archive/diff.go | 6 +---- 5 files changed, 28 insertions(+), 40 deletions(-) diff --git a/pkg/archive/archive.go b/pkg/archive/archive.go index 365e8f18ed..57b8f7d7bf 100644 --- a/pkg/archive/archive.go +++ b/pkg/archive/archive.go @@ -25,7 +25,6 @@ import ( "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" @@ -235,8 +234,7 @@ func (r *readCloserWrapper) Close() error { // 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 := bufio.NewReaderSize(archive, 32*1024) bs, err := buf.Peek(10) if err != nil && err != io.EOF { // Note: we'll ignore any io.EOF error because there are some odd @@ -258,7 +256,6 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) { if readCloser, ok := r.(io.ReadCloser); ok { readCloser.Close() } - p.Put(buf) return nil }, } @@ -300,18 +297,19 @@ func DecompressStream(archive io.Reader) (io.ReadCloser, error) { } } +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 +380,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 +527,6 @@ type tarWhiteoutConverter interface { type tarAppender struct { TarWriter *tar.Writer - Buffer *bufio.Writer // for hardlink mapping SeenFiles map[uint64]string @@ -547,7 +544,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 +661,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 +708,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 +919,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 +1074,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 +1150,6 @@ loop: } } } - trBuf.Reset(tr) if err := remapIDs(options.IDMap, hdr); err != nil { return err @@ -1181,7 +1165,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 +1368,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 From 9189a6e0abd26596550279a701b7eca7b32a5534 Mon Sep 17 00:00:00 2001 From: Derek McGowan Date: Tue, 17 Dec 2024 23:01:44 -0800 Subject: [PATCH 2/4] Fix chrootarchive test After the untar errors, the reader must complete in order to fill the buffer used by the subsequent check. Signed-off-by: Derek McGowan --- pkg/chrootarchive/archive_unix_test.go | 2 ++ 1 file changed, 2 insertions(+) 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. From 4c251b6b03f36d8391325b0cc0c304f78a95d873 Mon Sep 17 00:00:00 2001 From: Derek McGowan Date: Tue, 17 Dec 2024 23:21:59 -0800 Subject: [PATCH 3/4] Add pool for archive decompress stream Cleanup decompress logic and add a pool. The close logic should be custom defined for each compression type since they have different close interfaces. Signed-off-by: Derek McGowan --- pkg/archive/archive.go | 92 ++++++++++++++++++++++++++++++++---------- 1 file changed, 70 insertions(+), 22 deletions(-) diff --git a/pkg/archive/archive.go b/pkg/archive/archive.go index 57b8f7d7bf..b7eae21328 100644 --- a/pkg/archive/archive.go +++ b/pkg/archive/archive.go @@ -19,6 +19,7 @@ import ( "runtime/debug" "strconv" "strings" + "sync" "sync/atomic" "syscall" "time" @@ -229,12 +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) { - buf := bufio.NewReaderSize(archive, 32*1024) + 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 @@ -246,25 +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() - } - 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()) @@ -273,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()) @@ -285,13 +320,26 @@ 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()) } From be4eac753f1731b1a39b2e4d8cca43f2287cedeb Mon Sep 17 00:00:00 2001 From: Derek McGowan Date: Tue, 24 Dec 2024 22:01:05 -0800 Subject: [PATCH 4/4] Remove use of bufio in cli import tests The use of bufio for writing without flushing can lead to an incomplete writing of the tar and subsequent unexpected EOF when importing. Signed-off-by: Derek McGowan --- integration-cli/docker_cli_import_test.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) 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())