feat(snapshots): added free pool of localfs entries (#2115)

This avoids allocations of entries as large directories are traversed.

Backing up 100k small files in a single directory:

duration: current:8.1 baseline:8.6 change:-5.7 %
avg_heap_objects: current:7330688.9 baseline:7463750.6 change:-1.8 %
avg_heap_bytes: current:829248496.6 baseline:864880473.2 change:-4.1 %
avg_ram: current:159.3 baseline:159.1 change:+0.1 %
max_ram: current:283.2 baseline:285.4 change:-0.8 %
avg_cpu: current:153.1 baseline:143.6 change:+6.7 %
max_cpu: current:295.6 baseline:292.4 change:+1.1 %

Backing up Linux 5.14.8 using --parallel=4:

duration: current:6.0 baseline:6.0 change:-0.0 %
avg_heap_objects: current:5654431.1 baseline:5845078.3 change:-3.3 %
avg_heap_bytes: current:764526988.2 baseline:806833320.4 change:-5.2 %
avg_ram: current:218.2 baseline:220.1 change:-0.9 %
max_ram: current:323.5 baseline:319.5 change:+1.3 %
avg_cpu: current:144.3 baseline:145.1 change:-0.6 %
max_cpu: current:224.6 baseline:220.6 change:+1.8 %

fs/localfs/local_fs.go

@@ -71,9 +71,6 @@ func (e *filesystemEntry) LocalFilesystemPath() string {
 	return e.fullPath()
 }
 
-func (e *filesystemEntry) Close() {
-}
-
 var _ os.FileInfo = (*filesystemEntry)(nil)
 
 func newEntry(fi os.FileInfo, prefix string) filesystemEntry {
@@ -285,7 +282,7 @@ func (f *fileWithMetadata) Entry() (fs.Entry, error) {
 		return nil, errors.Wrap(err, "unable to stat() local file")
 	}
 
-	return &filesystemFile{newEntry(fi, dirPrefix(f.Name()))}, nil
+	return newFilesystemFile(newEntry(fi, dirPrefix(f.Name()))), nil
 }
 
 func (fsf *filesystemFile) Open(ctx context.Context) (fs.Reader, error) {
@@ -349,7 +346,7 @@ func Directory(path string) (fs.Directory, error) {
 	case *filesystemSymlink:
 		// it's a symbolic link, possibly to a directory, it may work or we may get a ReadDir() error.
 		// this is apparently how VSS mounted snapshots appear on Windows and attempts to os.Readlink() fail on them.
-		return &filesystemDirectory{e.filesystemEntry}, nil
+		return newFilesystemDirectory(e.filesystemEntry), nil
 
 	default:
 		return nil, errors.Errorf("not a directory: %v (was %T)", path, e)
@@ -362,28 +359,28 @@ func entryFromDirEntry(fi os.FileInfo, prefix string) fs.Entry {
 
 	switch {
 	case maskedmode == os.ModeDir && !isplaceholder:
-		return &filesystemDirectory{newEntry(fi, prefix)}
+		return newFilesystemDirectory(newEntry(fi, prefix))
 
 	case maskedmode == os.ModeDir && isplaceholder:
-		return &shallowFilesystemDirectory{newEntry(fi, prefix)}
+		return newShallowFilesystemDirectory(newEntry(fi, prefix))
 
 	case maskedmode == os.ModeSymlink && !isplaceholder:
-		return &filesystemSymlink{newEntry(fi, prefix)}
+		return newFilesystemSymlink(newEntry(fi, prefix))
 
 	case maskedmode == 0 && !isplaceholder:
-		return &filesystemFile{newEntry(fi, prefix)}
+		return newFilesystemFile(newEntry(fi, prefix))
 
 	case maskedmode == 0 && isplaceholder:
-		return &shallowFilesystemFile{newEntry(fi, prefix)}
+		return newShallowFilesystemFile(newEntry(fi, prefix))
 
 	default:
-		return &filesystemErrorEntry{newEntry(fi, prefix), fs.ErrUnknown}
+		return newFilesystemErrorEntry(newEntry(fi, prefix), fs.ErrUnknown)
 	}
 }
 
 var (
-	_ fs.Directory  = &filesystemDirectory{}
-	_ fs.File       = &filesystemFile{}
-	_ fs.Symlink    = &filesystemSymlink{}
-	_ fs.ErrorEntry = &filesystemErrorEntry{}
+	_ fs.Directory  = (*filesystemDirectory)(nil)
+	_ fs.File       = (*filesystemFile)(nil)
+	_ fs.Symlink    = (*filesystemSymlink)(nil)
+	_ fs.ErrorEntry = (*filesystemErrorEntry)(nil)
 )

fs/localfs/local_fs_pool.go

@@ -0,0 +1,122 @@
+package localfs
+
+import "github.com/kopia/kopia/internal/freepool"
+
+// nolint:gochecknoglobals
+var (
+	filesystemFilePool = freepool.New(
+		func() interface{} { return &filesystemFile{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*filesystemFile) = filesystemFile{}
+		},
+	)
+	filesystemDirectoryPool = freepool.New(
+		func() interface{} { return &filesystemDirectory{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*filesystemDirectory) = filesystemDirectory{}
+		},
+	)
+	filesystemSymlinkPool = freepool.New(
+		func() interface{} { return &filesystemSymlink{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*filesystemSymlink) = filesystemSymlink{}
+		},
+	)
+	filesystemErrorEntryPool = freepool.New(
+		func() interface{} { return &filesystemErrorEntry{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*filesystemErrorEntry) = filesystemErrorEntry{}
+		},
+	)
+	shallowFilesystemFilePool = freepool.New(
+		func() interface{} { return &shallowFilesystemFile{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*shallowFilesystemFile) = shallowFilesystemFile{}
+		},
+	)
+	shallowFilesystemDirectoryPool = freepool.New(
+		func() interface{} { return &shallowFilesystemDirectory{} },
+		func(v interface{}) {
+			// nolint:forcetypeassert
+			*v.(*shallowFilesystemDirectory) = shallowFilesystemDirectory{}
+		},
+	)
+)
+
+func newFilesystemFile(e filesystemEntry) *filesystemFile {
+	// nolint:forcetypeassert
+	fsf := filesystemFilePool.Take().(*filesystemFile)
+	fsf.filesystemEntry = e
+
+	return fsf
+}
+
+func (fsf *filesystemFile) Close() {
+	filesystemFilePool.Return(fsf)
+}
+
+func newFilesystemDirectory(e filesystemEntry) *filesystemDirectory {
+	// nolint:forcetypeassert
+	fsd := filesystemDirectoryPool.Take().(*filesystemDirectory)
+	fsd.filesystemEntry = e
+
+	return fsd
+}
+
+func (fsd *filesystemDirectory) Close() {
+	filesystemDirectoryPool.Return(fsd)
+}
+
+func newFilesystemSymlink(e filesystemEntry) *filesystemSymlink {
+	// nolint:forcetypeassert
+	fsd := filesystemSymlinkPool.Take().(*filesystemSymlink)
+	fsd.filesystemEntry = e
+
+	return fsd
+}
+
+func (fsl *filesystemSymlink) Close() {
+	filesystemSymlinkPool.Return(fsl)
+}
+
+func newFilesystemErrorEntry(e filesystemEntry, err error) *filesystemErrorEntry {
+	// nolint:forcetypeassert
+	fse := filesystemErrorEntryPool.Take().(*filesystemErrorEntry)
+	fse.filesystemEntry = e
+	fse.err = err
+
+	return fse
+}
+
+func (e *filesystemErrorEntry) Close() {
+	filesystemErrorEntryPool.Return(e)
+}
+
+func newShallowFilesystemFile(e filesystemEntry) *shallowFilesystemFile {
+	// nolint:forcetypeassert
+	fsf := shallowFilesystemFilePool.Take().(*shallowFilesystemFile)
+	fsf.filesystemEntry = e
+
+	return fsf
+}
+
+func (fsf *shallowFilesystemFile) Close() {
+	shallowFilesystemFilePool.Return(fsf)
+}
+
+func newShallowFilesystemDirectory(e filesystemEntry) *shallowFilesystemDirectory {
+	// nolint:forcetypeassert
+	fsf := shallowFilesystemDirectoryPool.Take().(*shallowFilesystemDirectory)
+	fsf.filesystemEntry = e
+
+	return fsf
+}
+
+func (fsd *shallowFilesystemDirectory) Close() {
+	shallowFilesystemDirectoryPool.Return(fsd)
+}

go.mod

@@ -106,6 +106,7 @@ require (
 	github.com/chromedp/cdproto v0.0.0-20220515234810-83d799542a04
 	github.com/chromedp/chromedp v0.8.2
 	github.com/edsrzf/mmap-go v1.1.0
+	github.com/golang-design/lockfree v0.0.1
 	github.com/hashicorp/golang-lru v0.5.4
 	go.opentelemetry.io/otel v1.7.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.7.0
@@ -117,6 +118,7 @@ require (
 	cloud.google.com/go/compute v1.7.0 // indirect
 	cloud.google.com/go/iam v0.3.0 // indirect
 	github.com/alessio/shellescape v1.4.1 // indirect
+	github.com/changkun/lockfree v0.0.1 // indirect
 	github.com/chromedp/sysutil v1.0.0 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect

go.sum

@@ -115,6 +115,8 @@ github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghf
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/changkun/lockfree v0.0.1 h1:5WefVJLglY4IHRqOQmh6Ao6wkJYaJkarshKU8VUtId4=
+github.com/changkun/lockfree v0.0.1/go.mod h1:3bKiaXn/iNzIPlSvSOMSVbRQUQtAp8qUAyBUtzU11s4=
 github.com/chmduquesne/rollinghash v4.0.0+incompatible h1:hnREQO+DXjqIw3rUTzWN7/+Dpw+N5Um8zpKV0JOEgbo=
 github.com/chmduquesne/rollinghash v4.0.0+incompatible/go.mod h1:Uc2I36RRfTAf7Dge82bi3RU0OQUmXT9iweIcPqvr8A0=
 github.com/chromedp/cdproto v0.0.0-20220515234810-83d799542a04 h1:8GLetRp0N/g2MVzUmFRBXgLJTPofYAdTyWNR2lC0EQM=
@@ -199,6 +201,8 @@ github.com/godbus/dbus/v5 v5.0.6/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
 github.com/gofrs/flock v0.8.1 h1:+gYjHKf32LDeiEEFhQaotPbLuUXjY5ZqxKgXy7n59aw=
 github.com/gofrs/flock v0.8.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
+github.com/golang-design/lockfree v0.0.1 h1:vWujpt8Z3u1ufDCMqRlfgagew+QSE++YDGnFAqsCiY8=
+github.com/golang-design/lockfree v0.0.1/go.mod h1:bWcnifzOPu9MxpFuHV0uEuJjIiDAgdj3rfCH/sUTO+M=
 github.com/golang-jwt/jwt/v4 v4.0.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
 github.com/golang-jwt/jwt/v4 v4.4.2 h1:rcc4lwaZgFMCZ5jxF9ABolDcIHdBytAFgqFPbSJQAYs=
 github.com/golang-jwt/jwt/v4 v4.4.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
@@ -343,8 +347,6 @@ github.com/klauspost/pgzip v1.2.5 h1:qnWYvvKqedOF2ulHpMG72XQol4ILEJ8k2wwRl/Km8oE
 github.com/klauspost/pgzip v1.2.5/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
-github.com/kopia/htmluibuild v0.0.0-20220626023928-f12339ad0bfb h1:F/lMaVP34B9+ydzRuba2Kv2qNnCI0x1MDfuMTvD8sis=
-github.com/kopia/htmluibuild v0.0.0-20220626023928-f12339ad0bfb/go.mod h1:eWer4rx9P8lJo2eKc+Q7AZ1dE1x1hJNdkbDFPzMu1Hw=
 github.com/kopia/htmluibuild v0.0.0-20220702012237-9d334f452b6b h1:ox9MBvtiRvcnVjYQd5XYmk9IhRGMo7lK/pbJA1Ud2EY=
 github.com/kopia/htmluibuild v0.0.0-20220702012237-9d334f452b6b/go.mod h1:eWer4rx9P8lJo2eKc+Q7AZ1dE1x1hJNdkbDFPzMu1Hw=
 github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=

internal/freepool/freepool.go

@@ -0,0 +1,39 @@
+// Package freepool manages a free pool of objects that are expensive to create.
+package freepool
+
+import (
+	"github.com/golang-design/lockfree"
+)
+
+// Pool is a small pool of recently returned objects.
+// Unlike sync.Pool, the pool is not subject to gargbage collection under memory pressure.
+type Pool struct {
+	makeNew func() interface{}
+	clean   func(v interface{})
+	stack   lockfree.Stack
+}
+
+// Take returns an item from the pool, and if not available makes a new one.
+func (p *Pool) Take() interface{} {
+	v := p.stack.Pop()
+	if v == nil {
+		return p.makeNew()
+	}
+
+	return v
+}
+
+// Return returns an item to the pool after cleaning it.
+func (p *Pool) Return(v interface{}) {
+	p.clean(v)
+
+	p.stack.Push(v)
+}
+
+// New returns a new free pool.
+func New(makeNew func() interface{}, clean func(v interface{})) *Pool {
+	return &Pool{
+		makeNew: makeNew,
+		clean:   clean,
+	}
+}

snapshot/snapshotfs/estimate.go

@@ -131,6 +131,8 @@ func estimate(ctx context.Context, relativePath string, entry fs.Entry, policyTr
 		progress.Processing(ctx, relativePath)
 
 		err := entry.IterateEntries(ctx, func(c context.Context, child fs.Entry) error {
+			defer child.Close()
+
 			if err2 := estimate(ctx, filepath.Join(relativePath, child.Name()), child, policyTree.Child(child.Name()), stats, ib, eb, ed, progress, maxExamplesPerBucket); err2 != nil {
 				return processEntryError{err2}
 			}

snapshot/snapshotfs/upload.go

@@ -280,17 +280,12 @@ func (u *Uploader) uploadFileData(ctx context.Context, parentCheckpointRegistry
 		return nil, err
 	}
 
-	fi2, err := file.Entry()
-	if err != nil {
-		return nil, errors.Wrap(err, "unable to get file entry after copying")
-	}
-
 	r, err := writer.Result()
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to get result")
 	}
 
-	de, err := newDirEntry(fi2, fname, r)
+	de, err := newDirEntry(f, fname, r)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to create dir entry")
 	}
@@ -786,6 +781,8 @@ func (u *Uploader) processSingle(
 	localDirPathOrEmpty string,
 	parentCheckpointRegistry *checkpointRegistry,
 ) error {
+	defer entry.Close()
+
 	// note this function runs in parallel and updates 'u.stats', which must be done using atomic operations.
 	t0 := timetrack.StartTimer()