diff --git a/AUTHORS b/AUTHORS index 6f315438..267d9bc5 100644 --- a/AUTHORS +++ b/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) diff --git a/deb/index_files.go b/deb/index_files.go index 3db1f84d..0b9cedcd 100644 --- a/deb/index_files.go +++ b/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 } diff --git a/deb/index_files_rename_test.go b/deb/index_files_rename_test.go new file mode 100644 index 00000000..d61414bc --- /dev/null +++ b/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) +} diff --git a/debian/aptly.conf b/debian/aptly.conf index 03a644c3..98dd3816 100644 --- a/debian/aptly.conf +++ b/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 ########## diff --git a/man/aptly.1 b/man/aptly.1 index 2a9b9a0a..7b61fe13 100644 --- a/man/aptly.1 +++ b/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 /////////// diff --git a/man/aptly.1.ronn.tmpl b/man/aptly.1.ronn.tmpl index 86c84070..eebcfcea 100644 --- a/man/aptly.1.ronn.tmpl +++ b/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 /////////// diff --git a/system/t02_config/ConfigShowTest_gold b/system/t02_config/ConfigShowTest_gold index 62e91a60..c609a5cc 100644 --- a/system/t02_config/ConfigShowTest_gold +++ b/system/t02_config/ConfigShowTest_gold @@ -40,6 +40,7 @@ "gpgKeys": [], "skipContentsPublishing": false, "skipBz2Publishing": false, + "publishRenameConcurrency": 4, "FileSystemPublishEndpoints": {}, "JFrogPublishEndpoints": null, "S3PublishEndpoints": {}, diff --git a/system/t02_config/ConfigShowYAMLTest_gold b/system/t02_config/ConfigShowYAMLTest_gold index df4b1fd0..42d93186 100644 --- a/system/t02_config/ConfigShowYAMLTest_gold +++ b/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: {} diff --git a/system/t02_config/CreateConfigTest_gold b/system/t02_config/CreateConfigTest_gold index 03a644c3..98dd3816 100644 --- a/system/t02_config/CreateConfigTest_gold +++ b/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 ########## diff --git a/utils/config.go b/utils/config.go index 2eb5efef..d540db10 100644 --- a/utils/config.go +++ b/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, diff --git a/utils/config_test.go b/utils/config_test.go index dedd24a1..f3ddd96b 100644 --- a/utils/config_test.go +++ b/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