Merge lp:~wgrant/launchpad/bug-728836 into lp:launchpad

[William Grant]

[r=lifeless][bug=728836] Optimise PublishingSet.copyBinariesTo to O(1) queries.

This branch improves the scalability of Archive:+copy-packages and Archive.syncSource by optimising PublishingSet.copyBinariesTo to a constant query count. The queries produced by this method can be seen clearly in OOPS-1858O745: it makes up all of the 2nd and 3rd most repeated queries, and much of the 5th (the 1st and 4th have already been addressed by another branch). Each of those cases is now reduced to a single query per source.

[Robert Collins (lifeless)]

Couple of bits that can be improved.

302 + if len(needed) == 0:
'if not needed:' would be clearer.

the 8 line generator expression is getting tricky to read.

review: Approve

=== modified file 'lib/lp/registry/interfaces/distroseries.py'
--- lib/lp/registry/interfaces/distroseries.py	2011-02-24 15:30:54 +0000
+++ lib/lp/registry/interfaces/distroseries.py	2011-03-07 07:05:26 +0000
@@ -767,7 +767,7 @@
         """
 
     def newArch(architecturetag, processorfamily, official, owner,
-                supports_virtualized=False):
+                supports_virtualized=False, enabled=True):
         """Create a new port or DistroArchSeries for this DistroSeries."""
 
     def copyTranslationsFromParent(ztm):
 
=== modified file 'lib/lp/registry/model/distroseries.py'
--- lib/lp/registry/model/distroseries.py	2011-02-22 08:20:21 +0000
+++ lib/lp/registry/model/distroseries.py	2011-03-07 07:05:26 +0000
@@ -1459,12 +1459,12 @@
         return DecoratedResultSet(package_caches, result_to_dsbp)
 
     def newArch(self, architecturetag, processorfamily, official, owner,
-                supports_virtualized=False):
+                supports_virtualized=False, enabled=True):
         """See `IDistroSeries`."""
         distroarchseries = DistroArchSeries(
             architecturetag=architecturetag, processorfamily=processorfamily,
             official=official, distroseries=self, owner=owner,
-            supports_virtualized=supports_virtualized)
+            supports_virtualized=supports_virtualized, enabled=enabled)
         return distroarchseries
 
     def newMilestone(self, name, dateexpected=None, summary=None,
 
=== modified file 'lib/lp/soyuz/interfaces/binarypackagerelease.py'
--- lib/lp/soyuz/interfaces/binarypackagerelease.py	2011-02-23 20:26:53 +0000
+++ lib/lp/soyuz/interfaces/binarypackagerelease.py	2011-03-07 07:05:26 +0000
@@ -44,6 +44,7 @@
 class IBinaryPackageRelease(Interface):
     id = Int(title=_('ID'), required=True)
     binarypackagename = Int(required=True)
+    binarypackagenameID = Int(required=True)
     version = TextLine(required=True, constraint=valid_debian_version)
     summary = Text(required=True)
     description = Text(required=True)
 
=== modified file 'lib/lp/soyuz/interfaces/publishing.py'
--- lib/lp/soyuz/interfaces/publishing.py	2011-03-03 00:43:44 +0000
+++ lib/lp/soyuz/interfaces/publishing.py	2011-03-07 07:05:26 +0000
@@ -678,6 +678,7 @@
             required=False, readonly=False,
             ),
         exported_as="distro_arch_series")
+    distroseries = Attribute("The distroseries being published into")
     component = Int(
             title=_('The component being published into'),
             required=False, readonly=False,
@@ -878,7 +879,25 @@
             publishing histories.
         """
 
-    def publishBinary(archive, binarypackagerelease, distroarchseries,
+    def publishBinaries(archive, distroseries, pocket, binaries):
+        """Efficiently publish multiple BinaryPackageReleases in an Archive.
+
+        Creates `IBinaryPackagePublishingHistory` records for each binary,
+        handling architecture-independent and debug packages, avoiding
+        creation of duplicate publications, and leaving disabled
+        architectures alone.
+
+        :param archive: The target `IArchive`.
+        :param distroseries: The target `IDistroSeries`.
+        :param pocket: The target `PackagePublishingPocket`.
+        :param binaries: A dict mapping `BinaryPackageReleases` to their
+            desired overrides as (`Component`, `Section`,
+            `PackagePublishingPriority`) tuples.
+
+        :return: A list of new `IBinaryPackagePublishingHistory` records.
+        """
+
+    def publishBinary(archive, binarypackagerelease, distroseries,
                       component, section, priority, pocket):
         """Publish a `BinaryPackageRelease` in an archive.
 
@@ -890,9 +909,7 @@
 
         :param archive: The target `IArchive`.
         :param binarypackagerelease: The `IBinaryPackageRelease` to copy.
-        :param distroarchseries: An `IDistroArchSeries`. If the binary is
-            architecture-independent, it will be published to all enabled
-            architectures in this series.
+        :param distroseries: An `IDistroSeries`.
         :param component: The target `IComponent`.
         :param section: The target `ISection`.
         :param priority: The target `PackagePublishingPriority`.
 
=== modified file 'lib/lp/soyuz/model/archive.py'
--- lib/lp/soyuz/model/archive.py	2011-03-03 03:13:13 +0000
+++ lib/lp/soyuz/model/archive.py	2011-03-07 07:05:26 +0000
@@ -386,7 +386,7 @@
             query, clauseTables=clauseTables, orderBy=orderBy)
         return dependencies
 
-    @property
+    @cachedproperty
     def debug_archive(self):
         """See `IArchive`."""
         if self.purpose == ArchivePurpose.PRIMARY:
@@ -1933,6 +1933,11 @@
         if enabled == False:
             new_archive.disable()
 
+        if purpose == ArchivePurpose.DEBUG:
+            if distribution.main_archive is not None:
+                del get_property_cache(
+                    distribution.main_archive).debug_archive
+
         # Private teams cannot have public PPAs.
         if owner.visibility == PersonVisibility.PRIVATE:
             new_archive.buildd_secret = create_unique_token_for_table(
 
=== modified file 'lib/lp/soyuz/model/publishing.py'
--- lib/lp/soyuz/model/publishing.py	2011-03-03 08:01:34 +0000
+++ lib/lp/soyuz/model/publishing.py	2011-03-07 07:05:26 +0000
@@ -29,8 +29,10 @@
     StringCol,
     )
 from storm.expr import (
+    And,
     Desc,
     LeftJoin,
+    Or,
     Sum,
     )
 from storm.store import Store
@@ -63,7 +65,6 @@
     IStoreSelector,
     MAIN_STORE,
     )
-from lp.app.errors import NotFoundError
 from lp.buildmaster.enums import BuildStatus
 from lp.buildmaster.model.buildfarmjob import BuildFarmJob
 from lp.buildmaster.model.packagebuild import PackageBuild
@@ -121,7 +122,7 @@
         'pool', poolify(source_name, component_name))
 
 
-def maybe_override_component(archive, distroseries, component):
+def get_component(archive, distroseries, component):
     """Override the component to fit in the archive, if possible.
 
     If the archive has a default component, and it forbids use of the
@@ -136,6 +137,22 @@
     return component
 
 
+def get_archive(archive, bpr):
+    """Get the archive in which this binary should be published.
+
+    Debug packages live in a DEBUG archive instead of a PRIMARY archive.
+    This helper implements that override.
+    """
+    if bpr.binpackageformat == BinaryPackageFormat.DDEB:
+        debug_archive = archive.debug_archive
+        if debug_archive is None:
+            raise QueueInconsistentStateError(
+                "Could not find the corresponding DEBUG archive "
+                "for %s" % (archive.displayname))
+        archive = debug_archive
+    return archive
+
+
 class FilePublishingBase:
     """Base class to publish files in the archive."""
 
@@ -914,18 +931,23 @@
             binarypackagepublishing=self).prejoin(preJoins)
 
     @property
+    def distroseries(self):
+        """See `IBinaryPackagePublishingHistory`"""
+        return self.distroarchseries.distroseries
+
+    @property
     def binary_package_name(self):
-        """See `ISourcePackagePublishingHistory`"""
+        """See `IBinaryPackagePublishingHistory`"""
         return self.binarypackagerelease.name
 
     @property
     def binary_package_version(self):
-        """See `ISourcePackagePublishingHistory`"""
+        """See `IBinaryPackagePublishingHistory`"""
         return self.binarypackagerelease.version
 
     @property
     def priority_name(self):
-        """See `ISourcePackagePublishingHistory`"""
+        """See `IBinaryPackagePublishingHistory`"""
         return self.priority.name
 
     @property
@@ -1245,6 +1267,42 @@
         self.requestDeletion(removed_by, removal_comment)
 
 
+def expand_binary_requests(distroseries, binaries):
+    """Architecture-expand a dict of binary publication requests.
+
+    For architecture-independent binaries, a tuple will be returned for each
+    enabled architecture in the series.
+    For architecture-dependent binaries, a tuple will be returned only for the
+    architecture corresponding to the build architecture, if it exists and is
+    enabled.
+
+    :param binaries: A dict mapping `BinaryPackageReleases` to tuples of their
+        desired overrides.
+
+    :return: The binaries and the architectures in which they should be
+        published, as a sequence of (`DistroArchSeries`,
+        `BinaryPackageRelease`, (overrides)) tuples.
+    """
+
+    archs = list(distroseries.enabled_architectures)
+    arch_map = dict((arch.architecturetag, arch) for arch in archs)
+
+    expanded = []
+    for bpr, overrides in binaries.iteritems():
+        if bpr.architecturespecific:
+            # Find the DAS in this series corresponding to the original
+            # build arch tag. If it does not exist or is disabled, we should
+            # not publish.
+            target_arch = arch_map.get(
+                bpr.build.distro_arch_series.architecturetag)
+            target_archs = [target_arch] if target_arch is not None else []
+        else:
+            target_archs = archs
+        for target_arch in target_archs:
+            expanded.append((target_arch, bpr, overrides))
+    return expanded
+
+
 class PublishingSet:
     """Utilities for manipulating publications in batches."""
 
@@ -1252,61 +1310,82 @@
 
     def copyBinariesTo(self, binaries, distroseries, pocket, archive):
         """See `IPublishingSet`."""
-        secure_copies = []
-        for binary in binaries:
-            # This will go wrong if nominatedarchindep gets deleted in a
-            # future series -- it will attempt to retrieve i386 from the
-            # new series, fail, and skip the publication instead of
-            # publishing the remaining archs.
-            try:
-                build = binary.binarypackagerelease.build
-                target_architecture = distroseries[
-                    build.distro_arch_series.architecturetag]
-            except NotFoundError:
-                continue
-            if not target_architecture.enabled:
-                continue
-            secure_copies.extend(
-                getUtility(IPublishingSet).publishBinary(
-                    archive, binary.binarypackagerelease, target_architecture,
-                    binary.component, binary.section, binary.priority,
-                    pocket))
-        return secure_copies
-
-    def publishBinary(self, archive, binarypackagerelease, distroarchseries,
+        return self.publishBinaries(
+            archive, distroseries, pocket,
+            dict(
+                (bpph.binarypackagerelease, (bpph.component, bpph.section,
+                 bpph.priority)) for bpph in binaries))
+
+    def publishBinaries(self, archive, distroseries, pocket,
+                        binaries):
+        """See `IPublishingSet`."""
+        # Expand the dict of binaries into a list of tuples including the
+        # architecture.
+        expanded = expand_binary_requests(distroseries, binaries)
+
+        # Find existing publications.
+        # We should really be able to just compare BPR.id, but
+        # CopyChecker doesn't seem to ensure that there are no
+        # conflicting binaries from other sources.
+        def make_package_condition(archive, das, bpr):
+            return And(
+                BinaryPackagePublishingHistory.archiveID ==
+                    get_archive(archive, bpr).id,
+                BinaryPackagePublishingHistory.distroarchseriesID == das.id,
+                BinaryPackageRelease.binarypackagenameID ==
+                    bpr.binarypackagenameID,
+                BinaryPackageRelease.version == bpr.version,
+                )
+        candidates = (
+            make_package_condition(archive, das, bpr)
+            for das, bpr, overrides in expanded)
+        already_published = IMasterStore(BinaryPackagePublishingHistory).find(
+            (BinaryPackagePublishingHistory.distroarchseriesID,
+             BinaryPackageRelease.binarypackagenameID,
+             BinaryPackageRelease.version),
+            BinaryPackagePublishingHistory.pocket == pocket,
+            BinaryPackagePublishingHistory.status.is_in(
+                active_publishing_status),
+            BinaryPackageRelease.id ==
+                BinaryPackagePublishingHistory.binarypackagereleaseID,
+            Or(*candidates)).config(distinct=True)
+        already_published = frozenset(already_published)
+
+        needed = [
+            (das, bpr, overrides) for (das, bpr, overrides) in
+            expanded if (das.id, bpr.binarypackagenameID, bpr.version)
+            not in already_published]
+        if not needed:
+            return []
+
+        insert_head = """
+            INSERT INTO BinaryPackagePublishingHistory
+            (archive, distroarchseries, pocket, binarypackagerelease,
+             component, section, priority, status, datecreated)
+            VALUES
+            """
+        insert_pubs = ", ".join(
+            "(%s)" % ", ".join(sqlvalues(
+                get_archive(archive, bpr).id, das.id, pocket, bpr.id,
+                get_component(archive, das.distroseries, component).id,
+                section.id, priority, PackagePublishingStatus.PENDING,
+                UTC_NOW))
+            for (das, bpr, (component, section, priority)) in needed)
+        insert_tail = " RETURNING BinaryPackagePublishingHistory.id"
+        new_ids = IMasterStore(BinaryPackagePublishingHistory).execute(
+            insert_head + insert_pubs + insert_tail)
+
+        publications = IMasterStore(BinaryPackagePublishingHistory).find(
+            BinaryPackagePublishingHistory,
+            BinaryPackagePublishingHistory.id.is_in(id[0] for id in new_ids))
+        return list(publications)
+
+    def publishBinary(self, archive, binarypackagerelease, distroseries,
                       component, section, priority, pocket):
         """See `IPublishingSet`."""
-        if not binarypackagerelease.architecturespecific:
-            target_archs = distroarchseries.distroseries.enabled_architectures
-        else:
-            target_archs = [distroarchseries]
-
-        # DDEBs targeted to the PRIMARY archive are published in the
-        # corresponding DEBUG archive.
-        if binarypackagerelease.binpackageformat == BinaryPackageFormat.DDEB:
-            debug_archive = archive.debug_archive
-            if debug_archive is None:
-                raise QueueInconsistentStateError(
-                    "Could not find the corresponding DEBUG archive "
-                    "for %s" % (archive.displayname))
-            archive = debug_archive
-
-        published_binaries = []
-        for target_arch in target_archs:
-            # We only publish the binary if it doesn't already exist in
-            # the destination. Note that this means we don't support
-            # override changes on their own.
-            binaries_in_destination = archive.getAllPublishedBinaries(
-                name=binarypackagerelease.name, exact_match=True,
-                version=binarypackagerelease.version,
-                status=active_publishing_status, pocket=pocket,
-                distroarchseries=target_arch)
-            if not bool(binaries_in_destination):
-                published_binaries.append(
-                    getUtility(IPublishingSet).newBinaryPublication(
-                        archive, binarypackagerelease, target_arch, component,
-                        section, priority, pocket))
-        return published_binaries
+        return self.publishBinaries(
+            archive, distroseries, pocket,
+            {binarypackagerelease: (component, section, priority)})
 
     def newBinaryPublication(self, archive, binarypackagerelease,
                              distroarchseries, component, section, priority,
@@ -1318,7 +1397,7 @@
             archive=archive,
             binarypackagerelease=binarypackagerelease,
             distroarchseries=distroarchseries,
-            component=maybe_override_component(
+            component=get_component(
                 archive, distroarchseries.distroseries, component),
             section=section,
             priority=priority,
@@ -1335,8 +1414,7 @@
             pocket=pocket,
             archive=archive,
             sourcepackagerelease=sourcepackagerelease,
-            component=maybe_override_component(
-                archive, distroseries, component),
+            component=get_component(archive, distroseries, component),
             section=section,
             status=PackagePublishingStatus.PENDING,
             datecreated=UTC_NOW,
 
=== modified file 'lib/lp/soyuz/model/queue.py'
--- lib/lp/soyuz/model/queue.py	2011-02-17 17:02:54 +0000
+++ lib/lp/soyuz/model/queue.py	2011-03-07 07:05:26 +0000
@@ -1457,12 +1457,9 @@
         """See `IPackageUploadBuild`."""
         # Determine the build's architecturetag
         build_archtag = self.build.distro_arch_series.architecturetag
-        # Determine the target arch series.
-        # This will raise NotFoundError if anything odd happens.
-        target_das = self.packageupload.distroseries[build_archtag]
+        distroseries = self.packageupload.distroseries
         debug(logger, "Publishing build to %s/%s/%s" % (
-            target_das.distroseries.distribution.name,
-            target_das.distroseries.name,
+            distroseries.distribution.name, distroseries.name,
             build_archtag))
 
         # First up, publish everything in this build into that dar.
@@ -1478,7 +1475,7 @@
                 getUtility(IPublishingSet).publishBinary(
                     archive=self.packageupload.archive,
                     binarypackagerelease=binary,
-                    distroarchseries=target_das,
+                    distroseries=distroseries,
                     component=binary.component,
                     section=binary.section,
                     priority=binary.priority,
 
=== modified file 'lib/lp/soyuz/tests/test_publishing.py'
--- lib/lp/soyuz/tests/test_publishing.py	2011-03-03 08:01:34 +0000
+++ lib/lp/soyuz/tests/test_publishing.py	2011-03-07 07:05:26 +0000
@@ -53,6 +53,7 @@
     PackagePublishingPriority,
     PackagePublishingStatus,
     )
+from lp.soyuz.interfaces.queue import QueueInconsistentStateError
 from lp.soyuz.interfaces.section import ISectionSet
 from lp.soyuz.model.processor import ProcessorFamily
 from lp.soyuz.model.publishing import (
@@ -1399,3 +1400,113 @@
                 for bpf in files:
                     bpf.libraryfile.filename
         self.assertThat(recorder, HasQueryCount(Equals(5)))
+
+
+class TestPublishBinaries(TestCaseWithFactory):
+    """Test PublishingSet.publishBinary() works."""
+
+    layer = LaunchpadZopelessLayer
+
+    def makeArgs(self, bprs, distroseries, archive=None):
+        """Create a dict of arguments for publishBinary."""
+        if archive is None:
+            archive = distroseries.main_archive
+        return {
+            'archive': archive,
+            'distroseries': distroseries,
+            'pocket': PackagePublishingPocket.BACKPORTS,
+            'binaries': dict(
+                (bpr, (self.factory.makeComponent(),
+                 self.factory.makeSection(),
+                 PackagePublishingPriority.REQUIRED)) for bpr in bprs),
+            }
+
+    def test_architecture_dependent(self):
+        # Architecture-dependent binaries get created as PENDING in the
+        # corresponding architecture of the destination series and pocket,
+        # with the given overrides.
+        arch_tag = self.factory.getUniqueString('arch-')
+        orig_das = self.factory.makeDistroArchSeries(
+            architecturetag=arch_tag)
+        target_das = self.factory.makeDistroArchSeries(
+            architecturetag=arch_tag)
+        build = self.factory.makeBinaryPackageBuild(distroarchseries=orig_das)
+        bpr = self.factory.makeBinaryPackageRelease(
+            build=build, architecturespecific=True)
+        args = self.makeArgs([bpr], target_das.distroseries)
+        [bpph] = getUtility(IPublishingSet).publishBinaries(**args)
+        overrides = args['binaries'][bpr]
+        self.assertEqual(bpr, bpph.binarypackagerelease)
+        self.assertEqual(
+            (args['archive'], target_das, args['pocket']),
+            (bpph.archive, bpph.distroarchseries, bpph.pocket))
+        self.assertEqual(
+            overrides, (bpph.component, bpph.section, bpph.priority))
+        self.assertEqual(PackagePublishingStatus.PENDING, bpph.status)
+
+    def test_architecture_independent(self):
+        # Architecture-independent binaries get published to all enabled
+        # DASes in the series.
+        bpr = self.factory.makeBinaryPackageRelease(
+            architecturespecific=False)
+        # Create 3 architectures. The binary will not be published in
+        # the disabled one.
+        target_das_a = self.factory.makeDistroArchSeries()
+        target_das_b = self.factory.makeDistroArchSeries(
+            distroseries=target_das_a.distroseries)
+        target_das_c = self.factory.makeDistroArchSeries(
+            distroseries=target_das_a.distroseries, enabled=False)
+        args = self.makeArgs([bpr], target_das_a.distroseries)
+        bpphs = getUtility(IPublishingSet).publishBinaries(
+            **args)
+        self.assertEquals(2, len(bpphs))
+        self.assertEquals(
+            set((target_das_a, target_das_b)),
+            set(bpph.distroarchseries for bpph in bpphs))
+
+    def test_does_not_duplicate(self):
+        # An attempt to copy something for a second time is ignored.
+        bpr = self.factory.makeBinaryPackageRelease()
+        target_das = self.factory.makeDistroArchSeries()
+        args = self.makeArgs([bpr], target_das.distroseries)
+        [new_bpph] = getUtility(IPublishingSet).publishBinaries(**args)
+        [] = getUtility(IPublishingSet).publishBinaries(**args)
+
+        # But changing the target (eg. to RELEASE instead of BACKPORTS)
+        # causes a new publication to be created.
+        args['pocket'] = PackagePublishingPocket.RELEASE
+        [another_bpph] = getUtility(IPublishingSet).publishBinaries(**args)
+
+    def test_ddebs_need_debug_archive(self):
+        debug = self.factory.makeBinaryPackageRelease(
+            binpackageformat=BinaryPackageFormat.DDEB)
+        args = self.makeArgs(
+            [debug], debug.build.distro_arch_series.distroseries)
+        self.assertRaises(
+            QueueInconsistentStateError,
+            getUtility(IPublishingSet).publishBinaries, **args)
+
+    def test_ddebs_go_to_debug_archive(self):
+        # Normal packages go to the given archive, but debug packages go
+        # to the corresponding debug archive.
+        das = self.factory.makeDistroArchSeries()
+        self.factory.makeArchive(
+            purpose=ArchivePurpose.DEBUG,
+            distribution=das.distroseries.distribution)
+        build = self.factory.makeBinaryPackageBuild(distroarchseries=das)
+        normal = self.factory.makeBinaryPackageRelease(build=build)
+        debug = self.factory.makeBinaryPackageRelease(
+            build=build, binpackageformat=BinaryPackageFormat.DDEB)
+        args = self.makeArgs([normal, debug], das.distroseries)
+        bpphs = getUtility(IPublishingSet).publishBinaries(**args)
+        self.assertEquals(2, len(bpphs))
+        self.assertEquals(
+            set((normal, debug)),
+            set(bpph.binarypackagerelease for bpph in bpphs))
+        self.assertEquals(
+            set((das.main_archive, das.main_archive.debug_archive)),
+            set(bpph.archive for bpph in bpphs))
+
+        # A second copy does nothing, because it checks in the debug
+        # archive too.
+        [] = getUtility(IPublishingSet).publishBinaries(**args)
 
=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py	2011-03-03 10:58:03 +0000
+++ lib/lp/testing/factory.py	2011-03-07 07:05:26 +0000
@@ -2315,7 +2315,7 @@
     def makeDistroArchSeries(self, distroseries=None,
                              architecturetag=None, processorfamily=None,
                              official=True, owner=None,
-                             supports_virtualized=False):
+                             supports_virtualized=False, enabled=True):
         """Create a new distroarchseries"""
 
         if distroseries is None:
@@ -2331,7 +2331,7 @@
             architecturetag = self.getUniqueString('arch')
         return distroseries.newArch(
             architecturetag, processorfamily, official, owner,
-            supports_virtualized)
+            supports_virtualized, enabled)
 
     def makeComponent(self, name=None):
         """Make a new `IComponent`."""