estargz: parallelize MinChunkSize builds

Builds with MinChunkSize > 0 run on a single core, no matter how large
the layer is.

They are serial because the writer maintains one invariant -- every
gzip stream except the last holds at least MinChunkSize compressed
bytes -- and does so by placing each stream boundary based on the
compressed size of everything written before it.

The build can be parallelized while preserving the invariant as long
as:

- each worker's slice of the tar holds at least MinChunkSize * 1032
  uncompressed bytes (1032 is DEFLATE's maximum compression ratio), so
  every slice fills at least one full stream;

- the trailing stream of a slice, which usually ends below the minimum,
  is folded into the stream before it. The writer now withholds a full
  stream's terminator until the next stream also reaches the minimum,
  and folds the tail back by replaying its buffered raw bytes. Only the
  short tail is ever recompressed.

The invariant also comes out stronger: the trailing stream of the whole
blob, previously allowed to end short, is folded as well. A stream now
ends below MinChunkSize if and only if the data itself is smaller (or a
prefetch landmark forces a boundary).

Note that with this change MinChunkSize layer digests differ from
previous releases: large layers build in parallel (the layout depends
on GOMAXPROCS, like every other eStargz build) and trailing short
streams are folded away. eStargz makes no cross-version byte stability
promise.

Signed-off-by: Simone Primarosa <simone.primarosa@gmail.com>

Author: simonepri

cmd/ctr-remote/commands/convert.go

@@ -82,7 +82,7 @@ When '--all-platforms' is given all images in a manifest list must be available.
 		},
 		&cli.IntFlag{
 			Name:  "estargz-parallelism",
-			Usage: "Number of workers used to build the layer. Pinning it makes builds reproducible across machines regardless of CPU count. 0 (default) uses GOMAXPROCS; 1 forces a sequential build. Has no effect with --estargz-min-chunk-size.",
+			Usage: "Number of workers used to build the layer. Pinning it makes builds reproducible across machines regardless of CPU count. 0 (default) uses GOMAXPROCS; 1 forces a sequential build.",
 			Value: 0,
 		},
 		&cli.BoolFlag{

cmd/ctr-remote/commands/optimize.go

@@ -115,7 +115,7 @@ var OptimizeCommand = &cli.Command{
 		},
 		&cli.IntFlag{
 			Name:  "estargz-parallelism",
-			Usage: "Number of workers used to build the layer. Pinning it makes builds reproducible across machines regardless of CPU count. 0 (default) uses GOMAXPROCS; 1 forces a sequential build. Has no effect with --estargz-min-chunk-size (not applied to zstd:chunked)",
+			Usage: "Number of workers used to build the layer. Pinning it makes builds reproducible across machines regardless of CPU count. 0 (default) uses GOMAXPROCS; 1 forces a sequential build (not applied to zstd:chunked)",
 			Value: 0,
 		},
 		&cli.StringFlag{

docs/smaller-estargz.md

@@ -4,9 +4,9 @@ The following flags of `ctr-remote i convert` and `ctr-remote i optimize` allow
 
 - `--estargz-external-toc`: Separate TOC JSON into another image (called "TOC image"). The result eStargz doesn't contain TOC so we can expect a smaller size than normal eStargz.
 
-- `--estargz-min-chunk-size`: The minimal number of bytes of data must be written in one gzip stream. If it's > 0, multiple files and chunks can be written into one gzip stream. Smaller number of gzip header and smaller size of the result blob can be expected. `--estargz-min-chunk-size=0` produces normal eStargz.
+- `--estargz-min-chunk-size`: The minimal number of bytes of data must be written in one gzip stream. If it's > 0, multiple files and chunks can be written into one gzip stream. Smaller number of gzip header and smaller size of the result blob can be expected. `--estargz-min-chunk-size=0` produces normal eStargz. A trailing gzip stream that cannot reach the minimum is folded into the preceding stream instead, so a stream falls below `--estargz-min-chunk-size` only when the layer itself is smaller.
 
-- `--estargz-parallelism`: The number of workers used to build each layer. The tar is split into this many slices that are compressed in parallel, so the value also fixes the chunk boundaries: pinning it makes builds reproducible across machines regardless of their CPU count. `0` (the default) uses `GOMAXPROCS`; `1` forces a fully sequential build. This has no effect with `--estargz-min-chunk-size`, which is built with a single worker.
+- `--estargz-parallelism`: The number of workers used to build each layer. The tar is split into this many slices that are compressed in parallel, so the value also fixes the chunk boundaries: pinning it makes builds reproducible across machines regardless of their CPU count. `0` (the default) uses `GOMAXPROCS`; `1` forces a fully sequential build. With `--estargz-min-chunk-size`, fewer workers may be used when the layer is too small to give each one at least one full gzip stream.
 
 ## `--estargz-external-toc` usage
 

estargz/build.go

@@ -123,6 +123,10 @@ func WithContext(ctx context.Context) Option {
 // By increasing this number, one gzip stream can contain multiple files
 // and it hopefully leads to smaller result blob.
 // NOTE: This adds a TOC property that old reader doesn't understand.
+// Builds run in parallel across the configured workers (see WithParallelism);
+// a trailing stream that cannot reach minChunkSize is folded into its
+// predecessor, so a stream falls below minChunkSize only when the input itself
+// is smaller.
 func WithMinChunkSize(minChunkSize int) Option {
 	return func(o *options) error {
 		o.minChunkSize = minChunkSize
@@ -135,8 +139,8 @@ func WithMinChunkSize(minChunkSize int) Option {
 // concurrently, so the value also fixes the resulting chunk boundaries: pinning
 // it makes builds reproducible across machines regardless of their CPU count.
 // Zero (the default) selects runtime.GOMAXPROCS(0); a value of 1 forces a fully
-// sequential build. This has no effect together with WithMinChunkSize, which is
-// always built with a single worker.
+// sequential build. With WithMinChunkSize, fewer workers may be used when the
+// layer is too small to give each one at least one full gzip stream.
 func WithParallelism(workers int) Option {
 	return func(o *options) error {
 		o.parallelism = workers
@@ -246,11 +250,16 @@ func Build(tarBlob *io.SectionReader, opt ...Option) (_ *Blob, rErr error) {
 		workers = runtime.GOMAXPROCS(0)
 	}
 	var tarParts [][]*entry
-	if opts.minChunkSize > 0 {
-		// Each entry needs to know the size of the current gzip stream so they
-		// cannot be processed in parallel.
+	switch {
+	case workers <= 1:
 		tarParts = [][]*entry{entries}
-	} else {
+	case opts.minChunkSize > 0:
+		// Give each worker enough data to fill at least one full stream even at
+		// gzip's maximum compression ratio, so that folding a short trailing
+		// stream (see cutGz) never crosses worker boundaries. This coarsens
+		// parallelism, but layers small enough to stay sequential compress fast.
+		tarParts = divideEntriesByMinSize(entries, workers, int64(opts.minChunkSize)*maxGzipCompressionRatio)
+	default:
 		tarParts = divideEntries(entries, workers)
 	}
 	writers := make([]*Writer, len(tarParts))
@@ -393,6 +402,50 @@ func tocAndFooter(compressor Compressor, toc *JTOC, offset int64) (io.Reader, di
 	return buf, tocDigest, nil
 }
 
+// maxGzipCompressionRatio is the largest ratio DEFLATE can achieve: a 258-byte
+// match coded as a roughly two-bit symbol pair, i.e. 258*8/2 = 1032. A slice
+// of MinChunkSize*1032 bytes thus compresses to at least MinChunkSize.
+const maxGzipCompressionRatio = 1032
+
+// divideEntriesByMinSize packs entries into at most maxParts consecutive
+// groups of at least minPartSize uncompressed bytes each (total/maxParts when
+// that is larger). A trailing remainder below minPartSize is folded into the
+// last group; data that cannot fill even one group is returned as one.
+func divideEntriesByMinSize(entries []*entry, maxParts int, minPartSize int64) (set [][]*entry) {
+	var total int64
+	for _, e := range entries {
+		total += e.header.Size
+	}
+	target := total / int64(maxParts)
+	if target < minPartSize {
+		target = minPartSize
+	}
+	var (
+		cur     []*entry
+		curSize int64
+	)
+	for _, e := range entries {
+		cur = append(cur, e)
+		curSize += e.header.Size
+		// the last group takes the remainder, so the count never exceeds maxParts
+		if curSize >= target && len(set) < maxParts-1 {
+			set = append(set, cur)
+			cur, curSize = nil, 0
+		}
+	}
+	switch {
+	case len(cur) == 0:
+	case len(set) > 0 && curSize < minPartSize:
+		set[len(set)-1] = append(set[len(set)-1], cur...)
+	default:
+		set = append(set, cur)
+	}
+	if len(set) == 0 {
+		set = [][]*entry{entries}
+	}
+	return
+}
+
 // divideEntries divides passed entries to the parts at least the number specified by the
 // argument.
 func divideEntries(entries []*entry, minPartsNum int) (set [][]*entry) {