New upstream version 1.1.1

cmd/mirror_update.go

@@ -5,7 +5,9 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 
+	"github.com/smira/aptly/aptly"
 	"github.com/smira/aptly/deb"
 	"github.com/smira/aptly/query"
 	"github.com/smira/aptly/utils"
@@ -69,7 +71,7 @@ func aptlyMirrorUpdate(cmd *commander.Command, args []string) error {
 		}
 
 		var oldLen, newLen int
-		oldLen, newLen, err = repo.ApplyFilter(context.DependencyOptions(), filterQuery)
+		oldLen, newLen, err = repo.ApplyFilter(context.DependencyOptions(), filterQuery, context.Progress())
 		if err != nil {
 			return fmt.Errorf("unable to update: %s", err)
 		}
@@ -81,8 +83,12 @@ func aptlyMirrorUpdate(cmd *commander.Command, args []string) error {
 		queue        []deb.PackageDownloadTask
 	)
 
+	skipExistingPackages := context.Flags().Lookup("skip-existing-packages").Value.Get().(bool)
+
 	context.Progress().Printf("Building download queue...\n")
-	queue, downloadSize, err = repo.BuildDownloadQueue(context.PackagePool())
+	queue, downloadSize, err = repo.BuildDownloadQueue(context.PackagePool(), context.CollectionFactory().PackageCollection(),
+		context.CollectionFactory().ChecksumCollection(), skipExistingPackages)
+
 	if err != nil {
 		return fmt.Errorf("unable to update: %s", err)
 	}
@@ -110,6 +116,14 @@ func aptlyMirrorUpdate(cmd *commander.Command, args []string) error {
 	// Catch ^C
 	sigch := make(chan os.Signal)
 	signal.Notify(sigch, os.Interrupt)
+	defer signal.Stop(sigch)
+
+	abort := make(chan struct{})
+	go func() {
+		<-sigch
+		signal.Stop(sigch)
+		close(abort)
+	}()
 
 	count := len(queue)
 	context.Progress().Printf("Download queue: %d items (%s)\n", count, utils.HumanBytes(downloadSize))
@@ -117,37 +131,82 @@ func aptlyMirrorUpdate(cmd *commander.Command, args []string) error {
 	// Download from the queue
 	context.Progress().InitBar(downloadSize, true)
 
-	// Download all package files
-	ch := make(chan error, count)
+	downloadQueue := make(chan int)
+
+	var (
+		errors  []string
+		errLock sync.Mutex
+	)
+
+	pushError := func(err error) {
+		errLock.Lock()
+		errors = append(errors, err.Error())
+		errLock.Unlock()
+	}
 
-	// In separate goroutine (to avoid blocking main), push queue to downloader
 	go func() {
-		for _, task := range queue {
-			context.Downloader().DownloadWithChecksum(repo.PackageURL(task.RepoURI).String(), task.DestinationPath, ch, task.Checksums, ignoreMismatch, maxTries)
+		for idx := range queue {
+			select {
+			case downloadQueue <- idx:
+			case <-abort:
+				return
+			}
 		}
-
-		// We don't need queue after this point
-		queue = nil
+		close(downloadQueue)
 	}()
 
-	// Wait for all downloads to finish
-	var errors []string
+	var wg sync.WaitGroup
 
-	for count > 0 {
-		select {
-		case <-sigch:
-			signal.Stop(sigch)
-			return fmt.Errorf("unable to update: interrupted")
-		case err = <-ch:
-			if err != nil {
-				errors = append(errors, err.Error())
+	for i := 0; i < context.Config().DownloadConcurrency; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case idx, ok := <-downloadQueue:
+					if !ok {
+						return
+					}
+
+					task := &queue[idx]
+
+					var e error
+
+					// provision download location
+					task.TempDownPath, e = context.PackagePool().(aptly.LocalPackagePool).GenerateTempPath(task.File.Filename)
+					if e != nil {
+						pushError(e)
+						continue
+					}
+
+					// download file...
+					e = context.Downloader().DownloadWithChecksum(
+						repo.PackageURL(task.File.DownloadURL()).String(),
+						task.TempDownPath,
+						&task.File.Checksums,
+						ignoreMismatch,
+						maxTries)
+					if e != nil {
+						pushError(e)
+						continue
+					}
+				case <-abort:
+					return
+				}
 			}
-			count--
-		}
+		}()
+	}
+
+	// Wait for all downloads to finish
+	wg.Wait()
+
+	select {
+	case <-abort:
+		return fmt.Errorf("unable to update: interrupted")
+	default:
 	}
 
 	context.Progress().ShutdownBar()
-	signal.Stop(sigch)
 
 	if len(errors) > 0 {
 		return fmt.Errorf("unable to update: download errors:\n  %s", strings.Join(errors, "\n  "))
@@ -158,7 +217,37 @@ func aptlyMirrorUpdate(cmd *commander.Command, args []string) error {
 		return fmt.Errorf("unable to update: %s", err)
 	}
 
-	repo.FinalizeDownload()
+	// Import downloaded files
+	context.Progress().InitBar(int64(len(queue)), false)
+
+	for idx := range queue {
+
+		context.Progress().AddBar(1)
+
+		task := &queue[idx]
+
+		// and import it back to the pool
+		task.File.PoolPath, err = context.PackagePool().Import(task.TempDownPath, task.File.Filename, &task.File.Checksums, true, context.CollectionFactory().ChecksumCollection())
+		if err != nil {
+			return fmt.Errorf("unable to import file: %s", err)
+		}
+
+		// update "attached" files if any
+		for _, additionalTask := range task.Additional {
+			additionalTask.File.PoolPath = task.File.PoolPath
+			additionalTask.File.Checksums = task.File.Checksums
+		}
+
+		select {
+		case <-abort:
+			return fmt.Errorf("unable to update: interrupted")
+		default:
+		}
+	}
+
+	context.Progress().ShutdownBar()
+
+	repo.FinalizeDownload(context.CollectionFactory(), context.Progress())
 	err = context.CollectionFactory().RemoteRepoCollection().Update(repo)
 	if err != nil {
 		return fmt.Errorf("unable to update: %s", err)
@@ -188,6 +277,7 @@ Example:
 	cmd.Flag.Bool("force", false, "force update mirror even if it is locked by another process")
 	cmd.Flag.Bool("ignore-checksums", false, "ignore checksum mismatches while downloading package files and metadata")
 	cmd.Flag.Bool("ignore-signatures", false, "disable verification of Release file signatures")
+	cmd.Flag.Bool("skip-existing-packages", false, "do not check file existence for packages listed in the internal database of the mirror")
 	cmd.Flag.Int64("download-limit", 0, "limit download speed (kbytes/sec)")
 	cmd.Flag.Int("max-tries", 1, "max download tries till process fails with download error")
 	cmd.Flag.Var(&keyRingsFlag{}, "keyring", "gpg keyring to use when verifying Release file (could be specified multiple times)")