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..2b4390de 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" @@ -499,15 +500,48 @@ func (files *indexFiles) FinalizeAll(progress aptly.Progress, signer pgp.Signer) return } -func (files *indexFiles) RenameFiles() error { - var err error +// How many index renames are in flight at once. Every rename is a round trip, +// and on an S3 backend it is a CopyObject followed by a DeleteObject, so a +// publish rewriting a dozen indexes spent most of its wall clock waiting on +// them one at a time. Four matches the default used for concurrent package +// uploads. +const renameConcurrency = 4 +func (files *indexFiles) RenameFiles() 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 + ) + sem := make(chan struct{}, renameConcurrency) + + 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..754ce10b --- /dev/null +++ b/deb/index_files_rename_test.go @@ -0,0 +1,108 @@ +package deb + +import ( + "fmt" + "sync" + + "github.com/aptly-dev/aptly/aptly" + + . "gopkg.in/check.v1" +) + +type RenameFilesSuite struct{} + +var _ = Suite(&RenameFilesSuite{}) + +// 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 +} + +func (s *renameRecordingStorage) RenameFile(oldName, newName string) error { + s.mu.Lock() + s.inFlight++ + if s.inFlight > s.maxSeen { + s.maxSeen = s.inFlight + } + s.mu.Unlock() + + defer func() { + s.mu.Lock() + s.inFlight-- + s.mu.Unlock() + }() + + 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)} + } + 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) TestRenameFilesStaysWithinTheConcurrencyLimit(c *C) { + files, rec := newRenameFiles(nil, 50) + + c.Assert(files.RenameFiles(), IsNil) + + // An upper bound, so this cannot flake: fewer in flight than the limit is + // always acceptable, more never is. + c.Check(rec.maxSeen <= renameConcurrency, Equals, true, + Commentf("saw %d renames in flight, limit is %d", rec.maxSeen, renameConcurrency)) +} + +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) +}