Merge pull request #44 from imeoer/converter-tar-pack

converter: improve pack performance by fifo

.github/workflows/ci.yml

@@ -7,7 +7,7 @@ on:
     branches: [main]
 
 env:
-  NYDUS_VERSION: v2.0.0-rc.2
+  NYDUS_VERSION: v2.0.0-rc.5
 
 jobs:
   build:
@@ -57,8 +57,8 @@ jobs:
       - name: Build
         run: |
           # Download nydus components
-          wget https://github.com/dragonflyoss/image-service/releases/download/${{ env.NYDUS_VERSION }}/nydus-static-${{ env.NYDUS_VERSION }}-x86_64.tgz
-          tar xzvf nydus-static-${{ env.NYDUS_VERSION }}-x86_64.tgz
+          wget https://github.com/dragonflyoss/image-service/releases/download/${{ env.NYDUS_VERSION }}/nydus-static-${{ env.NYDUS_VERSION }}-linux-amd64.tgz
+          tar xzvf nydus-static-${{ env.NYDUS_VERSION }}-linux-amd64.tgz
           mkdir -p /usr/bin
           sudo mv nydus-static/nydus-image nydus-static/nydusd-fusedev nydus-static/nydusify /usr/bin/
 

.github/workflows/image.yml

@@ -70,8 +70,8 @@ jobs:
           tar xzf crictl-${{ env.CRICTL_VERSION }}-linux-amd64.tar.gz -C /usr/local/bin/
           # install nydus-overlayfs
           NYDUS_VER=v$(curl -s "https://api.github.com/repos/dragonflyoss/image-service/releases/latest" | jq -r .tag_name | sed 's/^v//')
-          wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-x86_64.tgz
-          tar xzf nydus-static-$NYDUS_VER-x86_64.tgz
+          wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-linux-amd64.tgz
+          tar xzf nydus-static-$NYDUS_VER-linux-amd64.tgz
           sudo cp nydus-static/nydus-overlayfs /usr/local/sbin/
           # install containerd
           #CONTAINERD_VER=$(curl -s "https://api.github.com/repos/containerd/containerd/releases/latest" | jq -r .tag_name | sed 's/^v//')

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/containerd/cgroups v1.0.3 // indirect
+	github.com/containerd/fifo v1.0.0 // indirect
 	github.com/containerd/ttrpc v1.1.0 // indirect
 	github.com/containerd/typeurl v1.0.2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect

go.sum

@@ -190,6 +190,7 @@ github.com/containerd/fifo v0.0.0-20190226154929-a9fb20d87448/go.mod h1:ODA38xgv
 github.com/containerd/fifo v0.0.0-20200410184934-f15a3290365b/go.mod h1:jPQ2IAeZRCYxpS/Cm1495vGFww6ecHmMk1YJH2Q5ln0=
 github.com/containerd/fifo v0.0.0-20201026212402-0724c46b320c/go.mod h1:jPQ2IAeZRCYxpS/Cm1495vGFww6ecHmMk1YJH2Q5ln0=
 github.com/containerd/fifo v0.0.0-20210316144830-115abcc95a1d/go.mod h1:ocF/ME1SX5b1AOlWi9r677YJmCPSwwWnQ9O123vzpE4=
+github.com/containerd/fifo v1.0.0 h1:6PirWBr9/L7GDamKr+XM0IeUFXu5mf3M/BPpH9gaLBU=
 github.com/containerd/fifo v1.0.0/go.mod h1:ocF/ME1SX5b1AOlWi9r677YJmCPSwwWnQ9O123vzpE4=
 github.com/containerd/go-cni v1.0.1/go.mod h1:+vUpYxKvAF72G9i1WoDOiPGRtQpqsNW/ZHtSlv++smU=
 github.com/containerd/go-cni v1.0.2/go.mod h1:nrNABBHzu0ZwCug9Ije8hL2xBCYh/pjfMb1aZGrrohk=

misc/example/README.md

@@ -27,8 +27,8 @@ wget https://github.com/kubernetes-sigs/cri-tools/releases/download/$CRICTL_VERS
 tar xzf crictl-$CRICTL_VERSION-linux-amd64.tar.gz -C /usr/local/bin/
 # install nydus-overlayfs
 NYDUS_VER=v$(curl -s "https://api.github.com/repos/dragonflyoss/image-service/releases/latest" | jq -r .tag_name | sed 's/^v//')
-wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-x86_64.tgz
-tar xzf nydus-static-$NYDUS_VER-x86_64.tgz
+wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-linux-amd64.tgz
+tar xzf nydus-static-$NYDUS_VER-linux-amd64.tgz
 sudo cp nydus-static/nydus-overlayfs /usr/local/sbin/
 ```
 

misc/snapshotter/Dockerfile

@@ -2,8 +2,8 @@ FROM ubuntu:20.04 AS sourcer
 
 RUN apt update; apt install --no-install-recommends -y curl wget ca-certificates
 RUN export NYDUS_VERSION=$(curl --silent "https://api.github.com/repos/dragonflyoss/image-service/releases/latest" | grep -Po '"tag_name": "\K.*?(?=")'); \
-    wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VERSION/nydus-static-$NYDUS_VERSION-x86_64.tgz; \
-    tar xzf nydus-static-$NYDUS_VERSION-x86_64.tgz
+    wget https://github.com/dragonflyoss/image-service/releases/download/$NYDUS_VERSION/nydus-static-$NYDUS_VERSION-linux-amd64.tgz; \
+    tar xzf nydus-static-$NYDUS_VERSION-linux-amd64.tgz
 RUN mv nydus-static/* /; mv nydusd-fusedev nydusd
 
 FROM ubuntu:20.04

pkg/converter/convert.go

@@ -14,19 +14,20 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"syscall"
 
 	"github.com/containerd/containerd/archive"
 	"github.com/containerd/containerd/archive/compression"
+	"github.com/containerd/containerd/content"
+	"github.com/containerd/fifo"
 	"github.com/opencontainers/go-digest"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/containerd/nydus-snapshotter/pkg/converter/tool"
 )
 
-const dirNameInTar = "image"
-const blobNameInTar = "image/image.blob"
-const bootstrapNameInTar = "image/image.boot"
+const bootstrapNameInTar = "image.boot"
 
 const envNydusBuilder = "NYDUS_BUILDER"
 const envNydusWorkdir = "NYDUS_WORKDIR"
@@ -35,7 +36,7 @@ type Layer struct {
 	// Digest represents the hash of whole tar blob.
 	Digest digest.Digest
 	// ReaderAt holds the reader of whole tar blob.
-	ReaderAt io.ReaderAt
+	ReaderAt content.ReaderAt
 }
 
 type ConvertOption struct {
@@ -88,105 +89,63 @@ func unpackOciTar(ctx context.Context, dst string, reader io.Reader) error {
 	return nil
 }
 
-// Unpack the bootstrap from Nydus formatted tar stream (blob + bootstrap).
-func unpackBootstrapFromNydusTar(ctx context.Context, ra io.ReaderAt) (io.ReadCloser, error) {
-	pr, pw := io.Pipe()
-
-	go func() {
-		found := false
-		reader := newSeekReader(ra)
-		tr := tar.NewReader(reader)
-		for {
-			hdr, err := tr.Next()
-			if err != nil {
-				if err == io.EOF {
-					break
-				} else {
-					pw.CloseWithError(errors.Wrap(err, "seek tar"))
-					return
-				}
-			}
-			if hdr.Name == bootstrapNameInTar {
-				if _, err := io.Copy(pw, newCtxReader(ctx, tr)); err != nil {
-					pw.CloseWithError(errors.Wrap(err, "copy from tar"))
-					return
-				}
-				found = true
-				break
-			}
+// Unpack the bootstrap from nydus formatted tar stream (blob + bootstrap).
+// The nydus formatted tar stream is a tar-like structure that arranges the
+// data as follows:
+//
+// `blob_data | blob_tar_header | bootstrap_data | boostrap_tar_header`
+func unpackBootstrapFromNydusTar(ctx context.Context, ra content.ReaderAt, target io.Writer) error {
+	cur := ra.Size()
+	reader := newSeekReader(ra)
+
+	const headerSize = 512
+
+	// Seek from tail to head of nydus formatted tar stream to find nydus
+	// bootstrap data.
+	for {
+		if headerSize > cur {
+			return fmt.Errorf("invalid tar format at pos %d", cur)
 		}
 
-		if !found {
-			pw.CloseWithError(fmt.Errorf("not found %s in tar", bootstrapNameInTar))
-			return
+		// Try to seek to the part of tar header.
+		var err error
+		cur, err = reader.Seek(cur-headerSize, io.SeekCurrent)
+		if err != nil {
+			return errors.Wrapf(err, "seek to %d for tar header", cur-headerSize)
 		}
 
-		pw.Close()
-	}()
-
-	return pr, nil
-}
-
-// Write nydus artifact (blob/bootstrap) into a tar stream.
-func writeNydusTar(ctx context.Context, tw *tar.Writer, path, name string) error {
-	file, err := os.Open(path)
-	if err != nil {
-		return errors.Wrap(err, "open file for tar")
-	}
-	defer file.Close()
-
-	info, err := file.Stat()
-	if err != nil {
-		return errors.Wrap(err, "stat file for tar")
-	}
-
-	hdr := &tar.Header{
-		Name: name,
-		Mode: 0444,
-		Size: info.Size(),
-	}
-	if err := tw.WriteHeader(hdr); err != nil {
-		return errors.Wrap(err, "write file header")
-	}
-
-	if _, err := io.Copy(tw, newCtxReader(ctx, file)); err != nil {
-		return errors.Wrap(err, "copy file to tar")
-	}
-
-	return nil
-}
+		tr := tar.NewReader(reader)
+		// Parse tar header.
+		hdr, err := tr.Next()
+		if err != nil {
+			return errors.Wrap(err, "parse tar header")
+		}
 
-// Pack nydus blob and bootstrap to a tar stream
-func pack(ctx context.Context, dest io.Writer, blobPath string, bootstrapPath string) error {
-	tw := tar.NewWriter(dest)
+		if hdr.Name == bootstrapNameInTar {
+			// Try to seek to the part of tar data (bootstrap_data).
+			if hdr.Size > cur {
+				return fmt.Errorf("invalid tar format at pos %d", cur)
+			}
+			bootstrapOffset := cur - hdr.Size
+			_, err = reader.Seek(bootstrapOffset, io.SeekStart)
+			if err != nil {
+				return errors.Wrap(err, "seek to bootstrap data offset")
+			}
 
-	// Write a directory into tar stream.
-	dirHdr := &tar.Header{
-		Name:     filepath.Dir(dirNameInTar),
-		Mode:     0755,
-		Typeflag: tar.TypeDir,
-	}
-	if err := tw.WriteHeader(dirHdr); err != nil {
-		return errors.Wrap(err, "write dir header")
-	}
+			// Copy tar data (bootstrap_data) to provided target writer.
+			if _, err := io.CopyN(target, reader, hdr.Size); err != nil {
+				return errors.Wrap(err, "copy bootstrap data to reader")
+			}
 
-	// Write nydus blob into tar stream.
-	blobInfo, err := os.Stat(blobPath)
-	if err == nil && blobInfo.Size() > 0 {
-		if err := writeNydusTar(ctx, tw, blobPath, blobNameInTar); err != nil {
-			return errors.Wrap(err, "write blob")
+			return nil
 		}
-	}
-	// Write nydus bootstrap into tar stream.
-	if err := writeNydusTar(ctx, tw, bootstrapPath, bootstrapNameInTar); err != nil {
-		return errors.Wrap(err, "write bootstrap")
-	}
 
-	if err := tw.Close(); err != nil {
-		return errors.Wrap(err, "close tar writer")
+		if cur == hdr.Size {
+			break
+		}
 	}
 
-	return nil
+	return fmt.Errorf("can't find bootstrap in nydus tar")
 }
 
 // Convert converts a OCI formatted tar stream to a nydus formatted tar stream
@@ -225,6 +184,7 @@ func Convert(ctx context.Context, dest io.Writer, opt ConvertOption) (io.WriteCl
 	go func() {
 		if err := unpackOciTar(ctx, sourceDir, pr); err != nil {
 			pr.CloseWithError(errors.Wrapf(err, "unpack to %s", sourceDir))
+			close(unpackDone)
 			return
 		}
 		unpackDone <- true
@@ -240,23 +200,30 @@ func Convert(ctx context.Context, dest io.Writer, opt ConvertOption) (io.WriteCl
 		// so we need to wait for that here.
 		<-unpackDone
 
-		bootstrapPath := filepath.Join(workDir, "bootstrap")
 		blobPath := filepath.Join(workDir, "blob")
-
-		if err := tool.Convert(tool.ConvertOption{
-			BuilderPath: getBuilder(),
-
-			BootstrapPath:    bootstrapPath,
-			BlobPath:         blobPath,
-			RafsVersion:      opt.RafsVersion,
-			SourcePath:       sourceDir,
-			ChunkDictPath:    opt.ChunkDictPath,
-			PrefetchPatterns: opt.PrefetchPatterns,
-		}); err != nil {
-			return errors.Wrapf(err, "build source %s", sourceDir)
+		blobFifo, err := fifo.OpenFifo(ctx, blobPath, syscall.O_CREAT|syscall.O_RDONLY|syscall.O_NONBLOCK, 0644)
+		if err != nil {
+			return errors.Wrapf(err, "create fifo file")
 		}
+		defer blobFifo.Close()
+
+		go func() {
+			err := tool.Convert(tool.ConvertOption{
+				BuilderPath: getBuilder(),
+
+				BlobPath:         blobPath,
+				RafsVersion:      opt.RafsVersion,
+				SourcePath:       sourceDir,
+				ChunkDictPath:    opt.ChunkDictPath,
+				PrefetchPatterns: opt.PrefetchPatterns,
+			})
+			if err != nil {
+				pw.CloseWithError(errors.Wrapf(err, "convert blob for %s", sourceDir))
+				blobFifo.Close()
+			}
+		}()
 
-		if err := pack(ctx, dest, blobPath, bootstrapPath); err != nil {
+		if _, err := io.Copy(dest, blobFifo); err != nil {
 			return errors.Wrap(err, "pack nydus tar")
 		}
 
@@ -289,15 +256,9 @@ func Merge(ctx context.Context, layers []Layer, dest io.Writer, opt MergeOption)
 				}
 				defer bootstrap.Close()
 
-				bootstrapReader, err := unpackBootstrapFromNydusTar(ctx, layer.ReaderAt)
-				if err != nil {
+				if err := unpackBootstrapFromNydusTar(ctx, layer.ReaderAt, bootstrap); err != nil {
 					return errors.Wrap(err, "unpack nydus tar")
 				}
-				defer bootstrapReader.Close()
-
-				if _, err := io.Copy(bootstrap, newCtxReader(ctx, bootstrapReader)); err != nil {
-					return errors.Wrap(err, "copy bootstrap from tar")
-				}
 
 				return nil
 			}
@@ -324,7 +285,7 @@ func Merge(ctx context.Context, layers []Layer, dest io.Writer, opt MergeOption)
 	var rc io.ReadCloser
 
 	if opt.WithTar {
-		rc, err = packToTar(targetBootstrapPath, bootstrapNameInTar, false)
+		rc, err = packToTar(targetBootstrapPath, fmt.Sprintf("image/%s", bootstrapNameInTar), false)
 		if err != nil {
 			return errors.Wrap(err, "pack bootstrap to tar")
 		}

pkg/converter/tool/builder.go

@@ -47,8 +47,6 @@ func Convert(option ConvertOption) error {
 		"warn",
 		"--prefetch-policy",
 		"fs",
-		"--bootstrap",
-		option.BootstrapPath,
 		"--blob",
 		option.BlobPath,
 		"--source-type",
@@ -57,9 +55,7 @@ func Convert(option ConvertOption) error {
 		"none",
 		"--fs-version",
 		option.RafsVersion,
-		"--blob-offset",
-		// Add blob offset for chunk info with size_of(tar_header) * 2.
-		"1024",
+		"--inline-bootstrap",
 	}
 	if option.RafsVersion == "6" {
 		// FIXME: these options should be handled automatically in builder (nydus-image).