diff --git a/AUTHORS b/AUTHORS index ea69b2987..19f08d686 100644 --- a/AUTHORS +++ b/AUTHORS @@ -84,3 +84,4 @@ List of contributors, in chronological order: * Zhang Xiao (https://github.com/xzhang1) * Tom Nguyen (https://github.com/lecafard) * Philip Cramer (https://github.com/PhilipCramer) +* Martin Simon (https://github.com/barnumbirr) diff --git a/deb/index_files.go b/deb/index_files.go index 88c3c7091..8c65c8aa0 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" @@ -442,15 +443,48 @@ func (files *indexFiles) FinalizeAll(progress aptly.Progress, signer pgp.Signer) return } +// 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 { - 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 + ) + 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 000000000..0758891f9 --- /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 IndexFilesSuite struct{} + +var _ = Suite(&IndexFilesSuite{}) + +// 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 *IndexFilesSuite) 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 *IndexFilesSuite) 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 *IndexFilesSuite) 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 *IndexFilesSuite) TestRenameFilesWithNothingToDo(c *C) { + files, _ := newRenameFiles(nil, 0) + + c.Assert(files.RenameFiles(), IsNil) +}