From 0b6d76712fec13e4e80f515e8ab735f075117e47 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Fri, 1 Jul 2022 21:12:55 -0700 Subject: [PATCH] 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 | 27 ++++--- fs/localfs/local_fs_pool.go | 122 ++++++++++++++++++++++++++++++++ go.mod | 2 + go.sum | 6 +- internal/freepool/freepool.go | 39 ++++++++++ snapshot/snapshotfs/estimate.go | 2 + snapshot/snapshotfs/upload.go | 9 +-- 7 files changed, 184 insertions(+), 23 deletions(-) create mode 100644 fs/localfs/local_fs_pool.go create mode 100644 internal/freepool/freepool.go diff --git a/fs/localfs/local_fs.go b/fs/localfs/local_fs.go index 9c24ef8a4..c9259132a 100644 --- a/fs/localfs/local_fs.go +++ b/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) ) diff --git a/fs/localfs/local_fs_pool.go b/fs/localfs/local_fs_pool.go new file mode 100644 index 000000000..4258801ff --- /dev/null +++ b/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) +} diff --git a/go.mod b/go.mod index d444c9bc4..dc8cc84ed 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 15ba83b3f..74b030ef9 100644 --- a/go.sum +++ b/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= diff --git a/internal/freepool/freepool.go b/internal/freepool/freepool.go new file mode 100644 index 000000000..7c3948277 --- /dev/null +++ b/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, + } +} diff --git a/snapshot/snapshotfs/estimate.go b/snapshot/snapshotfs/estimate.go index 3b356faf6..37ca7f430 100644 --- a/snapshot/snapshotfs/estimate.go +++ b/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} } diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go index d1c1e7899..a3ccb38bb 100644 --- a/snapshot/snapshotfs/upload.go +++ b/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()