Merge pull request #1633 from barnumbirr/concurrent-index-renames

deb: rename published indexes concurrently

AUTHORS

@@ -88,3 +88,4 @@ List of contributors, in chronological order:
 * Phil Dibowitz (https://github.com/jaymzh)
 * James Munson (https://github.com/jmunson)
 * Luan Dang (https://github.com/luanmdang)
+* Martin Simon (https://github.com/barnumbirr)

deb/index_files.go

@@ -7,6 +7,7 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/pgp"
@@ -500,14 +501,41 @@ func (files *indexFiles) FinalizeAll(progress aptly.Progress, signer pgp.Signer)
 }
 
 func (files *indexFiles) RenameFiles() error {
-	var err error
+	type renamePair struct{ oldName, newName string }
 
+	pairs := make([]renamePair, 0, len(files.renameMap))
 	for oldName, newName := range files.renameMap {
-		err = files.publishedStorage.RenameFile(oldName, newName)
-		if err != nil {
-			return fmt.Errorf("unable to rename: %s", err)
-		}
+		pairs = append(pairs, renamePair{oldName, newName})
 	}
 
-	return nil
+	var (
+		wg       sync.WaitGroup
+		mu       sync.Mutex
+		firstErr error
+	)
+	// Below 1 the semaphore would admit nothing, so treat it as serial.
+	sem := make(chan struct{}, max(utils.Config.PublishRenameConcurrency, 1))
+
+	for _, pair := range pairs {
+		wg.Add(1)
+
+		go func(pair renamePair) {
+			defer wg.Done()
+
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			if err := files.publishedStorage.RenameFile(pair.oldName, pair.newName); err != nil {
+				mu.Lock()
+				if firstErr == nil {
+					firstErr = fmt.Errorf("unable to rename: %s", err)
+				}
+				mu.Unlock()
+			}
+		}(pair)
+	}
+
+	wg.Wait()
+
+	return firstErr
 }

deb/index_files_rename_test.go

@@ -0,0 +1,151 @@
+package deb
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/aptly-dev/aptly/aptly"
+	"github.com/aptly-dev/aptly/utils"
+
+	. "gopkg.in/check.v1"
+)
+
+type RenameFilesSuite struct {
+	savedRenameConcurrency int
+}
+
+var _ = Suite(&RenameFilesSuite{})
+
+func (s *RenameFilesSuite) SetUpTest(c *C) {
+	s.savedRenameConcurrency = utils.Config.PublishRenameConcurrency
+}
+
+func (s *RenameFilesSuite) TearDownTest(c *C) {
+	utils.Config.PublishRenameConcurrency = s.savedRenameConcurrency
+}
+
+// Only RenameFile is exercised by RenameFiles, so the rest of the interface is
+// embedded and left nil: a call to anything else would panic, which is the
+// loudest way to notice this fake drifting from what the code under test does.
+type renameRecordingStorage struct {
+	aptly.PublishedStorage
+
+	mu       sync.Mutex
+	renamed  map[string]string
+	failOn   string
+	inFlight int
+	maxSeen  int
+
+	// With waitFor set, every rename blocks until that many run at once, so a
+	// test proves the limit is reached rather than only respected.
+	waitFor  int
+	released bool
+	release  chan struct{}
+}
+
+func (s *renameRecordingStorage) RenameFile(oldName, newName string) error {
+	s.mu.Lock()
+	s.inFlight++
+	if s.inFlight > s.maxSeen {
+		s.maxSeen = s.inFlight
+	}
+	if s.waitFor > 0 && s.inFlight == s.waitFor && !s.released {
+		s.released = true
+		close(s.release)
+	}
+	s.mu.Unlock()
+
+	defer func() {
+		s.mu.Lock()
+		s.inFlight--
+		s.mu.Unlock()
+	}()
+
+	if s.waitFor > 0 {
+		select {
+		case <-s.release:
+		case <-time.After(5 * time.Second):
+			return fmt.Errorf("never saw %d renames in flight at once", s.waitFor)
+		}
+	}
+
+	if oldName == s.failOn {
+		return fmt.Errorf("rename refused for %s", oldName)
+	}
+
+	s.mu.Lock()
+	s.renamed[oldName] = newName
+	s.mu.Unlock()
+	return nil
+}
+
+func newRenameFiles(storage aptly.PublishedStorage, n int) (*indexFiles, *renameRecordingStorage) {
+	rec, ok := storage.(*renameRecordingStorage)
+	if !ok {
+		rec = &renameRecordingStorage{renamed: make(map[string]string), release: make(chan struct{})}
+	}
+	files := &indexFiles{
+		publishedStorage: rec,
+		renameMap:        make(map[string]string, n),
+	}
+	for i := 0; i < n; i++ {
+		files.renameMap[fmt.Sprintf("dists/x/Packages.tmp.%d", i)] = fmt.Sprintf("dists/x/Packages.%d", i)
+	}
+	return files, rec
+}
+
+func (s *RenameFilesSuite) TestRenameFilesRenamesEveryEntry(c *C) {
+	files, rec := newRenameFiles(nil, 20)
+
+	c.Assert(files.RenameFiles(), IsNil)
+
+	c.Check(len(rec.renamed), Equals, 20)
+	for i := 0; i < 20; i++ {
+		c.Check(rec.renamed[fmt.Sprintf("dists/x/Packages.tmp.%d", i)], Equals,
+			fmt.Sprintf("dists/x/Packages.%d", i))
+	}
+}
+
+func (s *RenameFilesSuite) TestRenameFilesRunsUpToTheConfiguredLimit(c *C) {
+	// Above the default of 4, so a limit that ignored the config would time out.
+	utils.Config.PublishRenameConcurrency = 6
+	files, rec := newRenameFiles(nil, 50)
+	rec.waitFor = 6
+
+	c.Assert(files.RenameFiles(), IsNil)
+
+	c.Check(rec.maxSeen, Equals, 6)
+	c.Check(len(rec.renamed), Equals, 50)
+}
+
+func (s *RenameFilesSuite) TestRenameFilesIsSerialAtALimitOfOneOrLess(c *C) {
+	for _, limit := range []int{1, 0, -3} {
+		utils.Config.PublishRenameConcurrency = limit
+		files, rec := newRenameFiles(nil, 10)
+
+		c.Assert(files.RenameFiles(), IsNil, Commentf("limit %d", limit))
+
+		c.Check(rec.maxSeen, Equals, 1, Commentf("limit %d", limit))
+		c.Check(len(rec.renamed), Equals, 10, Commentf("limit %d", limit))
+	}
+}
+
+func (s *RenameFilesSuite) TestRenameFilesReportsAFailure(c *C) {
+	files, rec := newRenameFiles(nil, 10)
+	rec.failOn = "dists/x/Packages.tmp.4"
+
+	err := files.RenameFiles()
+
+	c.Assert(err, NotNil)
+	c.Check(err, ErrorMatches, "unable to rename: rename refused for dists/x/Packages.tmp.4")
+	// The failure is reported, and it does not abandon the rest: every other
+	// entry is still renamed rather than left staged.
+	c.Check(len(rec.renamed), Equals, 9)
+}
+
+func (s *RenameFilesSuite) TestRenameFilesWithNothingToDo(c *C) {
+	files, _ := newRenameFiles(nil, 0)
+
+	c.Assert(files.RenameFiles(), IsNil)
+}

debian/aptly.conf

@@ -171,6 +171,10 @@ skip_contents_publishing: false
 # Do not create bz2 files
 skip_bz2_publishing: false
 
+# Number of index files renamed in parallel when a publish finishes
+# (set to 1 to rename them one at a time)
+publish_rename_concurrency: 4
+
 
 # Storage
 ##########

man/aptly.1

@@ -201,6 +201,10 @@ The legacy json configuration is still supported (and also supports comments):
   // Do not create bz2 files
   "skipBz2Publishing": false,
 
+  // Number of index files renamed in parallel when a publish finishes
+  // (set to 1 to rename them one at a time)
+  "publishRenameConcurrency": 4,
+
 
 // Storage
 ///////////

man/aptly.1.ronn.tmpl

@@ -190,6 +190,10 @@ The legacy json configuration is still supported (and also supports comments):
       // Do not create bz2 files
       "skipBz2Publishing": false,
 
+      // Number of index files renamed in parallel when a publish finishes
+      // (set to 1 to rename them one at a time)
+      "publishRenameConcurrency": 4,
+
 
     // Storage
     ///////////

system/t02_config/ConfigShowTest_gold

@@ -40,6 +40,7 @@
     "gpgKeys": [],
     "skipContentsPublishing": false,
     "skipBz2Publishing": false,
+    "publishRenameConcurrency": 4,
     "FileSystemPublishEndpoints": {},
     "JFrogPublishEndpoints": null,
     "S3PublishEndpoints": {},

system/t02_config/ConfigShowYAMLTest_gold

@@ -37,6 +37,7 @@ gpg_disable_verify: false
 gpg_keys: []
 skip_contents_publishing: false
 skip_bz2_publishing: false
+publish_rename_concurrency: 4
 filesystem_publish_endpoints: {}
 jfrog_publish_endpoints: {}
 s3_publish_endpoints: {}

system/t02_config/CreateConfigTest_gold

@@ -171,6 +171,10 @@ skip_contents_publishing: false
 # Do not create bz2 files
 skip_bz2_publishing: false
 
+# Number of index files renamed in parallel when a publish finishes
+# (set to 1 to rename them one at a time)
+publish_rename_concurrency: 4
+
 
 # Storage
 ##########

utils/config.go

@@ -62,8 +62,9 @@ type ConfigStructure struct { // nolint: maligned
 	GpgKeys          []string `json:"gpgKeys"                       yaml:"gpg_keys"`
 
 	// Publishing
-	SkipContentsPublishing bool `json:"skipContentsPublishing"        yaml:"skip_contents_publishing"`
-	SkipBz2Publishing      bool `json:"skipBz2Publishing"             yaml:"skip_bz2_publishing"`
+	SkipContentsPublishing   bool `json:"skipContentsPublishing"        yaml:"skip_contents_publishing"`
+	SkipBz2Publishing        bool `json:"skipBz2Publishing"             yaml:"skip_bz2_publishing"`
+	PublishRenameConcurrency int  `json:"publishRenameConcurrency"      yaml:"publish_rename_concurrency"`
 
 	// Storage
 	FileSystemPublishRoots map[string]FileSystemPublishRoot `json:"FileSystemPublishEndpoints"    yaml:"filesystem_publish_endpoints"`
@@ -254,6 +255,7 @@ var Config = ConfigStructure{
 	RootDir:                filepath.Join(os.Getenv("HOME"), ".aptly"),
 	AcquireByHashAlgorithms: []string{"MD5Sum", "SHA1", "SHA256", "SHA512"},
 	DownloadConcurrency:    4,
+	PublishRenameConcurrency: 4,
 	DownloadLimit:          0,
 	Downloader:             "default",
 	DatabaseOpenAttempts:   -1,

utils/config_test.go

@@ -27,6 +27,16 @@ func (s *ConfigSuite) TestLoadConfig(c *C) {
 	c.Check(s.config.GetRootDir(), Equals, "/opt/aptly/")
 	c.Check(s.config.DownloadConcurrency, Equals, 33)
 	c.Check(s.config.DatabaseOpenAttempts, Equals, 33)
+	c.Check(s.config.PublishRenameConcurrency, Equals, 3)
+}
+
+func (s *ConfigSuite) TestLoadConfigKeepsTheDefaultRenameConcurrency(c *C) {
+	configname := filepath.Join(c.MkDir(), "aptly.json3")
+	c.Assert(os.WriteFile(configname, []byte(`{"rootDir": "/opt/aptly/"}`), 0644), IsNil)
+
+	s.config = Config
+	c.Assert(LoadConfig(configname, &s.config), IsNil)
+	c.Check(s.config.PublishRenameConcurrency, Equals, 4)
 }
 
 func (s *ConfigSuite) TestSaveConfig(c *C) {
@@ -114,6 +124,7 @@ func (s *ConfigSuite) TestSaveConfig(c *C) {
   "gpgKeys": null,
   "skipContentsPublishing": false,
   "skipBz2Publishing": false,
+  "publishRenameConcurrency": 0,
   "FileSystemPublishEndpoints": {
     "test": {
       "rootDir": "/opt/aptly-publish",
@@ -213,6 +224,7 @@ func (s *ConfigSuite) TestLoadYAMLConfig(c *C) {
   c.Check(s.config.GetRootDir(), Equals, "/opt/aptly/")
   c.Check(s.config.DownloadConcurrency, Equals, 40)
   c.Check(s.config.DatabaseOpenAttempts, Equals, 10)
+  c.Check(s.config.PublishRenameConcurrency, Equals, 2)
 }
 
 func (s *ConfigSuite) TestLoadYAMLErrorConfig(c *C) {
@@ -311,6 +323,7 @@ func (s *ConfigSuite) TestSaveYAML2Config(c *C) {
     "gpg_keys: []\n"+
     "skip_contents_publishing: false\n"+
     "skip_bz2_publishing: false\n"+
+    "publish_rename_concurrency: 0\n"+
     "filesystem_publish_endpoints: {}\n"+
     "jfrog_publish_endpoints: {}\n"+
     "s3_publish_endpoints: {}\n"+
@@ -334,7 +347,7 @@ func (s *ConfigSuite) TestLoadEmptyConfig(c *C) {
 	c.Assert(err.Error(), Equals, "invalid yaml (EOF) or json (EOF)")
 }
 
-const configFile = `{"rootDir": "/opt/aptly/", "downloadConcurrency": 33, "databaseOpenAttempts": 33}`
+const configFile = `{"rootDir": "/opt/aptly/", "downloadConcurrency": 33, "databaseOpenAttempts": 33, "publishRenameConcurrency": 3}`
 const configFileYAML = `root_dir: /opt/aptly/
 log_level: error
 log_format: json
@@ -372,6 +385,7 @@ gpg_disable_verify: true
 gpg_keys: []
 skip_contents_publishing: true
 skip_bz2_publishing: true
+publish_rename_concurrency: 2
 filesystem_publish_endpoints:
     test1:
         root_dir: /opt/srv/aptly_public