Improve publish performance, especially for prefixes with a large number of snapshots

api/files.go

@@ -6,8 +6,10 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/gin-gonic/gin"
+	"github.com/saracen/walker"
 )
 
 func verifyPath(path string) bool {
@@ -34,17 +36,16 @@ func verifyDir(c *gin.Context) bool {
 // GET /files
 func apiFilesListDirs(c *gin.Context) {
 	list := []string{}
+	listLock := &sync.Mutex{}
 
-	err := filepath.Walk(context.UploadPath(), func(path string, info os.FileInfo, err error) error {
-		if err != nil {
-			return err
-		}
-
+	err := walker.Walk(context.UploadPath(), func(path string, info os.FileInfo) error {
 		if path == context.UploadPath() {
 			return nil
 		}
 
 		if info.IsDir() {
+			listLock.Lock()
+			defer listLock.Unlock()
 			list = append(list, filepath.Base(path))
 			return filepath.SkipDir
 		}
@@ -121,17 +122,16 @@ func apiFilesListFiles(c *gin.Context) {
 	}
 
 	list := []string{}
+	listLock := &sync.Mutex{}
 	root := filepath.Join(context.UploadPath(), c.Params.ByName("dir"))
 
-	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
-		if err != nil {
-			return err
-		}
-
+	err := walker.Walk(root, func(path string, info os.FileInfo) error {
 		if path == root {
 			return nil
 		}
 
+		listLock.Lock()
+		defer listLock.Unlock()
 		list = append(list, filepath.Base(path))
 
 		return nil

deb/changes.go

@@ -8,11 +8,13 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"sync"
 	"text/template"
 
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/pgp"
 	"github.com/aptly-dev/aptly/utils"
+	"github.com/saracen/walker"
 )
 
 // Changes is a result of .changes file parsing
@@ -247,6 +249,8 @@ func (c *Changes) GetArchitecture() string {
 
 // CollectChangesFiles walks filesystem collecting all .changes files
 func CollectChangesFiles(locations []string, reporter aptly.ResultReporter) (changesFiles, failedFiles []string) {
+	changesFilesLock := &sync.Mutex{}
+
 	for _, location := range locations {
 		info, err2 := os.Stat(location)
 		if err2 != nil {
@@ -255,15 +259,14 @@ func CollectChangesFiles(locations []string, reporter aptly.ResultReporter) (cha
 			continue
 		}
 		if info.IsDir() {
-			err2 = filepath.Walk(location, func(path string, info os.FileInfo, err3 error) error {
-				if err3 != nil {
-					return err3
-				}
+			err2 = walker.Walk(location, func(path string, info os.FileInfo) error {
 				if info.IsDir() {
 					return nil
 				}
 
 				if strings.HasSuffix(info.Name(), ".changes") {
+					changesFilesLock.Lock()
+					defer changesFilesLock.Unlock()
 					changesFiles = append(changesFiles, path)
 				}
 

deb/import.go

@@ -5,14 +5,19 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/pgp"
 	"github.com/aptly-dev/aptly/utils"
+	"github.com/saracen/walker"
 )
 
 // CollectPackageFiles walks filesystem collecting all candidates for package files
 func CollectPackageFiles(locations []string, reporter aptly.ResultReporter) (packageFiles, otherFiles, failedFiles []string) {
+	packageFilesLock := &sync.Mutex{}
+	otherFilesLock := &sync.Mutex{}
+
 	for _, location := range locations {
 		info, err2 := os.Stat(location)
 		if err2 != nil {
@@ -21,18 +26,19 @@ func CollectPackageFiles(locations []string, reporter aptly.ResultReporter) (pac
 			continue
 		}
 		if info.IsDir() {
-			err2 = filepath.Walk(location, func(path string, info os.FileInfo, err3 error) error {
-				if err3 != nil {
-					return err3
-				}
+			err2 = walker.Walk(location, func(path string, info os.FileInfo) error {
 				if info.IsDir() {
 					return nil
 				}
 
 				if strings.HasSuffix(info.Name(), ".deb") || strings.HasSuffix(info.Name(), ".udeb") ||
 					strings.HasSuffix(info.Name(), ".dsc") || strings.HasSuffix(info.Name(), ".ddeb") {
+					packageFilesLock.Lock()
+					defer packageFilesLock.Unlock()
 					packageFiles = append(packageFiles, path)
 				} else if strings.HasSuffix(info.Name(), ".buildinfo") {
+					otherFilesLock.Lock()
+					defer otherFilesLock.Unlock()
 					otherFiles = append(otherFiles, path)
 				}
 

deb/publish.go

@@ -1141,18 +1141,10 @@ func (collection *PublishedRepoCollection) Len() int {
 	return len(collection.list)
 }
 
-// CleanupPrefixComponentFiles removes all unreferenced files in published storage under prefix/component pair
-func (collection *PublishedRepoCollection) CleanupPrefixComponentFiles(prefix string, components []string,
-	publishedStorage aptly.PublishedStorage, collectionFactory *CollectionFactory, progress aptly.Progress) error {
-
-	collection.loadList()
-
-	var err error
+func (collection *PublishedRepoCollection) listReferencedFilesByComponent(prefix string, components []string,
+	collectionFactory *CollectionFactory, progress aptly.Progress) (map[string][]string, error) {
 	referencedFiles := map[string][]string{}
-
-	if progress != nil {
-		progress.Printf("Cleaning up prefix %#v components %s...\n", prefix, strings.Join(components, ", "))
-	}
+	processedComponentRefs := map[string]*PackageRefList{}
 
 	for _, r := range collection.list {
 		if r.Prefix == prefix {
@@ -1171,16 +1163,28 @@ func (collection *PublishedRepoCollection) CleanupPrefixComponentFiles(prefix st
 				continue
 			}
 
-			err = collection.LoadComplete(r, collectionFactory)
-			if err != nil {
-				return err
+			if err := collection.LoadComplete(r, collectionFactory); err != nil {
+				return nil, err
 			}
 
 			for _, component := range components {
 				if utils.StrSliceHasItem(repoComponents, component) {
-					packageList, err := NewPackageListFromRefList(r.RefList(component), collectionFactory.PackageCollection(), progress)
+					unseenRefs := r.RefList(component)
+					processedRefs := processedComponentRefs[component]
+					if processedRefs != nil {
+						unseenRefs = unseenRefs.Subtract(processedRefs)
+					} else {
+						processedRefs = NewPackageRefList()
+					}
+
+					if unseenRefs.Len() == 0 {
+						continue
+					}
+					processedComponentRefs[component] = processedRefs.Merge(unseenRefs, false, true)
+
+					packageList, err := NewPackageListFromRefList(unseenRefs, collectionFactory.PackageCollection(), progress)
 					if err != nil {
-						return err
+						return nil, err
 					}
 
 					packageList.ForEach(func(p *Package) error {
@@ -1200,6 +1204,24 @@ func (collection *PublishedRepoCollection) CleanupPrefixComponentFiles(prefix st
 		}
 	}
 
+	return referencedFiles, nil
+}
+
+// CleanupPrefixComponentFiles removes all unreferenced files in published storage under prefix/component pair
+func (collection *PublishedRepoCollection) CleanupPrefixComponentFiles(prefix string, components []string,
+	publishedStorage aptly.PublishedStorage, collectionFactory *CollectionFactory, progress aptly.Progress) error {
+
+	collection.loadList()
+
+	if progress != nil {
+		progress.Printf("Cleaning up prefix %#v components %s...\n", prefix, strings.Join(components, ", "))
+	}
+
+	referencedFiles, err := collection.listReferencedFilesByComponent(prefix, components, collectionFactory, progress)
+	if err != nil {
+		return err
+	}
+
 	for _, component := range components {
 		sort.Strings(referencedFiles[component])
 

deb/publish_bench_test.go

@@ -0,0 +1,113 @@
+package deb
+
+import (
+	"fmt"
+	"os"
+	"sort"
+	"testing"
+
+	"github.com/aptly-dev/aptly/database/goleveldb"
+)
+
+func BenchmarkListReferencedFiles(b *testing.B) {
+	const defaultComponent = "main"
+	const repoCount = 16
+	const repoPackagesCount = 1024
+	const uniqPackagesCount = 64
+
+	tmpDir, err := os.MkdirTemp("", "aptly-bench")
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer os.RemoveAll(tmpDir)
+
+	db, err := goleveldb.NewOpenDB(tmpDir)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer db.Close()
+
+	factory := NewCollectionFactory(db)
+	packageCollection := factory.PackageCollection()
+	repoCollection := factory.LocalRepoCollection()
+	publishCollection := factory.PublishedRepoCollection()
+
+	sharedRefs := NewPackageRefList()
+	{
+		transaction, err := db.OpenTransaction()
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		for pkgIndex := 0; pkgIndex < repoPackagesCount-uniqPackagesCount; pkgIndex++ {
+			p := &Package{
+				Name:         fmt.Sprintf("pkg-shared_%d", pkgIndex),
+				Version:      "1",
+				Architecture: "amd64",
+			}
+			p.UpdateFiles(PackageFiles{PackageFile{
+				Filename: fmt.Sprintf("pkg-shared_%d.deb", pkgIndex),
+			}})
+
+			packageCollection.UpdateInTransaction(p, transaction)
+			sharedRefs.Refs = append(sharedRefs.Refs, p.Key(""))
+		}
+
+		sort.Sort(sharedRefs)
+
+		if err := transaction.Commit(); err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	for repoIndex := 0; repoIndex < repoCount; repoIndex++ {
+		refs := NewPackageRefList()
+
+		transaction, err := db.OpenTransaction()
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		for pkgIndex := 0; pkgIndex < uniqPackagesCount; pkgIndex++ {
+			p := &Package{
+				Name:         fmt.Sprintf("pkg%d_%d", repoIndex, pkgIndex),
+				Version:      "1",
+				Architecture: "amd64",
+			}
+			p.UpdateFiles(PackageFiles{PackageFile{
+				Filename: fmt.Sprintf("pkg%d_%d.deb", repoIndex, pkgIndex),
+			}})
+
+			packageCollection.UpdateInTransaction(p, transaction)
+			refs.Refs = append(refs.Refs, p.Key(""))
+		}
+
+		if err := transaction.Commit(); err != nil {
+			b.Fatal(err)
+		}
+
+		sort.Sort(refs)
+
+		repo := NewLocalRepo(fmt.Sprintf("repo%d", repoIndex), "comment")
+		repo.DefaultDistribution = fmt.Sprintf("dist%d", repoIndex)
+		repo.DefaultComponent = defaultComponent
+		repo.UpdateRefList(refs.Merge(sharedRefs, false, true))
+		repoCollection.Add(repo)
+
+		publish, err := NewPublishedRepo("", "test", "", nil, []string{defaultComponent}, []interface{}{repo}, factory)
+		if err != nil {
+			b.Fatal(err)
+		}
+		publishCollection.Add(publish)
+	}
+
+	db.CompactDB()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := publishCollection.listReferencedFilesByComponent("test", []string{defaultComponent}, factory, nil)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}