Merge lp:~julian-edwards/launchpad/async-file-uploads-bug-662631 into lp:launchpad/db-devel

[Julian Edwards]

Make file uploads from builders asynchronous so that it doesn't block the rest of the build manager. This is the last part of the build manager to be made asynchronous.

This is the final part of the buildd-manager changes to make it fully asynchronous. Previously it was blocking when grabbing files from the builders - it now does that asynchronously using Twisted's HTTP file transfer.

Most of the changes are quite mechanical which accounts for most of the diff unfortunately - it's where I need to make the code an inner function which is called back from the Deferred. Hopefully you're familiar with this pattern; if not you'll learn it pretty quickly :)

Other changes are to add a new getFiles() function on the BuilderSlave object. This is initially complementing getFile() by calling it repeatedly but the longer term intention is to remove getFile() so that we can call one place and get hold of everything the builder knows about all at once.

The only other major change was a large amount of work to get the translations code fixed and tests passing. In particular, they were trying to test a successful tarball retrieval but not actually doing so - and in fact one test was asserting that nothing was retrieved!

[Jonathan Lange (jml)]

I don't have time to review this properly, but after having a quick skim through it all looks good.

 <jml>bigjools: I don't think you need files_downloaded (in packagebuild.py) you could also do:

   d = slave.getFiles(filenames_to_download)
   d.addCallback(lambda x: self.storeBuildInfo(self, librarian, slave_status))
   d.addCallback(build_info_stored)
   return d

Is my only comment of any substance.

review: Abstain

[Graham Binns (gmb)]

One typo:

> 865 + # incompressed file in the librarian

Should be uncompressed.

Other than that I'm happy with this branch.

review: Approve (code)

=== modified file 'lib/lp/buildmaster/interfaces/builder.py'
--- lib/lp/buildmaster/interfaces/builder.py	2010-11-16 14:51:02 +0000
+++ lib/lp/buildmaster/interfaces/builder.py	2010-11-30 11:46:52 +0000
@@ -203,7 +203,7 @@
         :param filename: The name of the file to be given to the librarian file
             alias.
         :param private: True if the build is for a private archive.
-        :return: A librarian file alias.
+        :return: A Deferred that calls back with a librarian file alias.
         """
 
     def getBuildQueue():

=== modified file 'lib/lp/buildmaster/interfaces/buildfarmjobbehavior.py'
--- lib/lp/buildmaster/interfaces/buildfarmjobbehavior.py	2010-04-12 05:52:01 +0000
+++ lib/lp/buildmaster/interfaces/buildfarmjobbehavior.py	2010-11-30 11:46:52 +0000
@@ -72,5 +72,8 @@
         """Verify the current build job status.
 
         Perform the required actions for each state.
+
+        :param queueItem: The `BuildQueue` for the build.
+        :return: A Deferred that fires when the update is done.
         """
 

=== modified file 'lib/lp/buildmaster/interfaces/packagebuild.py'
--- lib/lp/buildmaster/interfaces/packagebuild.py	2010-10-14 20:41:13 +0000
+++ lib/lp/buildmaster/interfaces/packagebuild.py	2010-11-30 11:46:52 +0000
@@ -94,7 +94,11 @@
         """
 
     def getLogFromSlave(build):
-        """Get last buildlog from slave. """
+        """Get last buildlog from slave. 
+        
+        :return: A Deferred that fires with the librarian ID of the log
+            when the log is finished downloading.
+        """
 
     def estimateDuration():
         """Estimate the build duration."""
@@ -130,6 +134,7 @@
 
         :param status: Slave build status string with 'BuildStatus.' stripped.
         :param slave_status: A dict as returned by IBuilder.slaveStatus
+        :return: A Deferred that fires when finished dealing with the build.
         """
 
     def queueBuild(suspended=False):

=== modified file 'lib/lp/buildmaster/model/builder.py'
--- lib/lp/buildmaster/model/builder.py	2010-11-26 18:36:09 +0000
+++ lib/lp/buildmaster/model/builder.py	2010-11-30 11:46:52 +0000
@@ -40,6 +40,7 @@
     reactor as default_reactor,
     )
 from twisted.web import xmlrpc
+from twisted.web.client import downloadPage
 
 from zope.component import getUtility
 from zope.interface import implements
@@ -213,12 +214,35 @@
         return self._with_timeout(self._server.callRemote(
             'ensurepresent', sha1sum, url, username, password))
 
-    def getFile(self, sha_sum):
-        """Construct a file-like object to return the named file."""
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        file_url = urlappend(self._file_cache_url, sha_sum)
-        return urllib2.urlopen(file_url)
+    def getFile(self, sha_sum, file_to_write):
+        """Fetch a file from the builder.
+
+        :param sha_sum: The sha of the file (which is also its name on the 
+            builder)
+        :param file_to_write: A file name or file-like object to write
+            the file to
+        :return: A Deferred that calls back when the download is done, or
+            errback with the error string.
+        """
+        file_url = urlappend(self._file_cache_url, sha_sum).encode('utf8')
+        # If desired we can pass a param "timeout" here but let's leave
+        # it at the default value if it becomes obvious we need to
+        # change it.
+        return downloadPage(file_url, file_to_write, followRedirect=0)
+
+    def getFiles(self, filemap):
+        """Fetch many files from the builder.
+
+        :param filemap: A Dictionary containing key values of the builder
+            file name to retrieve, which maps to a value containing the
+            file name or file object to write the file to.
+
+        :return: A DeferredList that calls back when the download is done.
+        """
+        dl = defer.gatherResults([
+            self.getFile(builder_file, filemap[builder_file])
+            for builder_file in filemap])
+        return dl
 
     def resume(self, clock=None):
         """Resume the builder in an asynchronous fashion.
@@ -607,38 +631,40 @@
         """See IBuilder."""
         out_file_fd, out_file_name = tempfile.mkstemp(suffix=".buildlog")
         out_file = os.fdopen(out_file_fd, "r+")
-        try:
-            # XXX 2010-10-18 bug=662631
-            # Change this to do non-blocking IO.
-            slave_file = self.slave.getFile(file_sha1)
-            copy_and_close(slave_file, out_file)
-            # If the requested file is the 'buildlog' compress it using gzip
-            # before storing in Librarian.
-            if file_sha1 == 'buildlog':
+
+        def got_file(ignored, filename, out_file, out_file_name):
+            try:
+                # If the requested file is the 'buildlog' compress it
+                # using gzip before storing in Librarian.
+                if file_sha1 == 'buildlog':
+                    out_file = open(out_file_name)
+                    filename += '.gz'
+                    out_file_name += '.gz'
+                    gz_file = gzip.GzipFile(out_file_name, mode='wb')
+                    copy_and_close(out_file, gz_file)
+                    os.remove(out_file_name.replace('.gz', ''))
+
+                # Reopen the file, seek to its end position, count and seek
+                # to beginning, ready for adding to the Librarian.
                 out_file = open(out_file_name)
-                filename += '.gz'
-                out_file_name += '.gz'
-                gz_file = gzip.GzipFile(out_file_name, mode='wb')
-                copy_and_close(out_file, gz_file)
-                os.remove(out_file_name.replace('.gz', ''))
-
-            # Reopen the file, seek to its end position, count and seek
-            # to beginning, ready for adding to the Librarian.
-            out_file = open(out_file_name)
-            out_file.seek(0, 2)
-            bytes_written = out_file.tell()
-            out_file.seek(0)
-
-            library_file = getUtility(ILibraryFileAliasSet).create(
-                filename, bytes_written, out_file,
-                contentType=filenameToContentType(filename),
-                restricted=private)
-        finally:
-            # Finally, remove the temporary file
-            out_file.close()
-            os.remove(out_file_name)
-
-        return library_file.id
+                out_file.seek(0, 2)
+                bytes_written = out_file.tell()
+                out_file.seek(0)
+
+                library_file = getUtility(ILibraryFileAliasSet).create(
+                    filename, bytes_written, out_file,
+                    contentType=filenameToContentType(filename),
+                    restricted=private)
+            finally:
+                # Remove the temporary file.  getFile() closes the file
+                # object.
+                os.remove(out_file_name)
+
+            return library_file.id
+
+        d = self.slave.getFile(file_sha1, out_file)
+        d.addCallback(got_file, filename, out_file, out_file_name)
+        return d
 
     def isAvailable(self):
         """See `IBuilder`."""

=== modified file 'lib/lp/buildmaster/model/buildfarmjobbehavior.py'
--- lib/lp/buildmaster/model/buildfarmjobbehavior.py	2010-10-27 14:25:19 +0000
+++ lib/lp/buildmaster/model/buildfarmjobbehavior.py	2010-11-30 11:46:52 +0000
@@ -191,9 +191,8 @@
         # XXX: dsilvers 2005-03-02: Confirm the builder has the right build?
 
         build = queueItem.specific_job.build
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        build.handleStatus(build_status, librarian, slave_status)
+        d = build.handleStatus(build_status, librarian, slave_status)
+        return d
 
 
 class IdleBuildBehavior(BuildFarmJobBehaviorBase):

=== modified file 'lib/lp/buildmaster/model/packagebuild.py'
--- lib/lp/buildmaster/model/packagebuild.py	2010-10-28 09:11:36 +0000
+++ lib/lp/buildmaster/model/packagebuild.py	2010-11-30 11:46:52 +0000
@@ -169,12 +169,11 @@
     def getLogFromSlave(package_build):
         """See `IPackageBuild`."""
         builder = package_build.buildqueue_record.builder
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        return builder.transferSlaveFileToLibrarian(
+        d = builder.transferSlaveFileToLibrarian(
             SLAVE_LOG_FILENAME,
             package_build.buildqueue_record.getLogFileName(),
             package_build.is_private)
+        return d
 
     def estimateDuration(self):
         """See `IPackageBuild`."""
@@ -183,21 +182,23 @@
     @staticmethod
     def storeBuildInfo(build, librarian, slave_status):
         """See `IPackageBuild`."""
-        # log, builder and date_finished are read-only, so we must
-        # currently remove the security proxy to set them.
-        naked_build = removeSecurityProxy(build)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        naked_build.log = build.getLogFromSlave(build)
-        naked_build.builder = build.buildqueue_record.builder
-        # XXX cprov 20060615 bug=120584: Currently buildduration includes
-        # the scanner latency, it should really be asking the slave for
-        # the duration spent building locally.
-        naked_build.date_finished = datetime.datetime.now(pytz.UTC)
-        if slave_status.get('dependencies') is not None:
-            build.dependencies = unicode(slave_status.get('dependencies'))
-        else:
-            build.dependencies = None
+        def got_log(lfa_id):
+            # log, builder and date_finished are read-only, so we must
+            # currently remove the security proxy to set them.
+            naked_build = removeSecurityProxy(build)
+            naked_build.log = lfa_id
+            naked_build.builder = build.buildqueue_record.builder
+            # XXX cprov 20060615 bug=120584: Currently buildduration includes
+            # the scanner latency, it should really be asking the slave for
+            # the duration spent building locally.
+            naked_build.date_finished = datetime.datetime.now(pytz.UTC)
+            if slave_status.get('dependencies') is not None:
+                build.dependencies = unicode(slave_status.get('dependencies'))
+            else:
+                build.dependencies = None
+
+        d = build.getLogFromSlave(build)
+        return d.addCallback(got_log)
 
     def verifySuccessfulUpload(self):
         """See `IPackageBuild`."""
@@ -284,9 +285,8 @@
             logger.critical("Unknown BuildStatus '%s' for builder '%s'"
                             % (status, self.buildqueue_record.builder.url))
             return
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        method(librarian, slave_status, logger)
+        d = method(librarian, slave_status, logger)
+        return d
 
     def _handleStatus_OK(self, librarian, slave_status, logger):
         """Handle a package that built successfully.
@@ -326,6 +326,7 @@
 
         slave = removeSecurityProxy(self.buildqueue_record.builder.slave)
         successful_copy_from_slave = True
+        filenames_to_download = {}
         for filename in filemap:
             logger.info("Grabbing file: %s" % filename)
             out_file_name = os.path.join(upload_path, filename)
@@ -338,46 +339,50 @@
                     "A slave tried to upload the file '%s' "
                     "for the build %d." % (filename, self.id))
                 break
-            out_file = open(out_file_name, "wb")
-            slave_file = slave.getFile(filemap[filename])
-            copy_and_close(slave_file, out_file)
-
+            filenames_to_download[filemap[filename]] = out_file_name
+
+        def build_info_stored(ignored):
+            # We only attempt the upload if we successfully copied all the
+            # files from the slave.
+            if successful_copy_from_slave:
+                logger.info(
+                    "Gathered %s %d completely. Moving %s to uploader queue." % (
+                    self.__class__.__name__, self.id, upload_leaf))
+                target_dir = os.path.join(root, "incoming")
+                self.status = BuildStatus.UPLOADING
+            else:
+                logger.warning(
+                    "Copy from slave for build %s was unsuccessful.", self.id)
+                self.status = BuildStatus.FAILEDTOUPLOAD
+                self.notify(extra_info='Copy from slave was unsuccessful.')
+                target_dir = os.path.join(root, "failed")
+
+            if not os.path.exists(target_dir):
+                os.mkdir(target_dir)
+
+            # Flush so there are no race conditions with archiveuploader about
+            # self.status.
+            Store.of(self).flush()
+
+            # Move the directory used to grab the binaries into
+            # the incoming directory so the upload processor never
+            # sees half-finished uploads.
+            os.rename(grab_dir, os.path.join(target_dir, upload_leaf))
+
+            # Release the builder for another job.
+            d = self.buildqueue_record.builder.cleanSlave()
+
+            # Remove BuildQueue record.
+            return d.addCallback(
+                lambda x:self.buildqueue_record.destroySelf())
+
+        d = slave.getFiles(filenames_to_download)
         # Store build information, build record was already updated during
         # the binary upload.
-        self.storeBuildInfo(self, librarian, slave_status)
-
-        # We only attempt the upload if we successfully copied all the
-        # files from the slave.
-        if successful_copy_from_slave:
-            logger.info(
-                "Gathered %s %d completely. Moving %s to uploader queue." % (
-                self.__class__.__name__, self.id, upload_leaf))
-            target_dir = os.path.join(root, "incoming")
-            self.status = BuildStatus.UPLOADING
-        else:
-            logger.warning(
-                "Copy from slave for build %s was unsuccessful.", self.id)
-            self.status = BuildStatus.FAILEDTOUPLOAD
-            self.notify(extra_info='Copy from slave was unsuccessful.')
-            target_dir = os.path.join(root, "failed")
-
-        if not os.path.exists(target_dir):
-            os.mkdir(target_dir)
-
-        # Flush so there are no race conditions with archiveuploader about
-        # self.status.
-        Store.of(self).flush()
-
-        # Move the directory used to grab the binaries into
-        # the incoming directory so the upload processor never
-        # sees half-finished uploads.
-        os.rename(grab_dir, os.path.join(target_dir, upload_leaf))
-
-        # Release the builder for another job.
-        self.buildqueue_record.builder.cleanSlave()
-
-        # Remove BuildQueue record.
-        self.buildqueue_record.destroySelf()
+        d.addCallback(
+            lambda x: self.storeBuildInfo(self, librarian, slave_status))
+        d.addCallback(build_info_stored)
+        return d
 
     def _handleStatus_PACKAGEFAIL(self, librarian, slave_status, logger):
         """Handle a package that had failed to build.
@@ -387,10 +392,14 @@
         remove Buildqueue entry.
         """
         self.status = BuildStatus.FAILEDTOBUILD
-        self.storeBuildInfo(self, librarian, slave_status)
-        self.buildqueue_record.builder.cleanSlave()
-        self.notify()
-        self.buildqueue_record.destroySelf()
+        def build_info_stored(ignored):
+            self.notify()
+            d = self.buildqueue_record.builder.cleanSlave()
+            return d.addCallback(
+                lambda x:self.buildqueue_record.destroySelf())
+
+        d = self.storeBuildInfo(self, librarian, slave_status)
+        return d.addCallback(build_info_stored)
 
     def _handleStatus_DEPFAIL(self, librarian, slave_status, logger):
         """Handle a package that had missing dependencies.
@@ -400,11 +409,15 @@
         entry and release builder slave for another job.
         """
         self.status = BuildStatus.MANUALDEPWAIT
-        self.storeBuildInfo(self, librarian, slave_status)
-        logger.critical("***** %s is MANUALDEPWAIT *****"
-                        % self.buildqueue_record.builder.name)
-        self.buildqueue_record.builder.cleanSlave()
-        self.buildqueue_record.destroySelf()
+        def build_info_stored(ignored):
+            logger.critical("***** %s is MANUALDEPWAIT *****"
+                            % self.buildqueue_record.builder.name)
+            d = self.buildqueue_record.builder.cleanSlave()
+            return d.addCallback(
+                lambda x:self.buildqueue_record.destroySelf())
+
+        d = self.storeBuildInfo(self, librarian, slave_status)
+        return d.addCallback(build_info_stored)
 
     def _handleStatus_CHROOTFAIL(self, librarian, slave_status, logger):
         """Handle a package that had failed when unpacking the CHROOT.
@@ -414,12 +427,16 @@
         and release the builder.
         """
         self.status = BuildStatus.CHROOTWAIT
-        self.storeBuildInfo(self, librarian, slave_status)
-        logger.critical("***** %s is CHROOTWAIT *****" %
-                        self.buildqueue_record.builder.name)
-        self.buildqueue_record.builder.cleanSlave()
-        self.notify()
-        self.buildqueue_record.destroySelf()
+        def build_info_stored(ignored):
+            logger.critical("***** %s is CHROOTWAIT *****" %
+                            self.buildqueue_record.builder.name)
+            self.notify()
+            d = self.buildqueue_record.builder.cleanSlave()
+            return d.addCallback(
+                lambda x:self.buildqueue_record.destroySelf())
+
+        d = self.storeBuildInfo(self, librarian, slave_status)
+        return d.addCallback(build_info_stored)
 
     def _handleStatus_BUILDERFAIL(self, librarian, slave_status, logger):
         """Handle builder failures.
@@ -432,9 +449,11 @@
                        % self.buildqueue_record.builder.name)
         self.buildqueue_record.builder.failBuilder(
             "Builder returned BUILDERFAIL when asked for its status")
-        # simply reset job
-        self.storeBuildInfo(self, librarian, slave_status)
-        self.buildqueue_record.reset()
+        def build_info_stored(ignored):
+            # simply reset job
+            self.buildqueue_record.reset()
+        d = self.storeBuildInfo(self, librarian, slave_status)
+        return d.addCallback(build_info_stored)
 
     def _handleStatus_GIVENBACK(self, librarian, slave_status, logger):
         """Handle automatic retry requested by builder.
@@ -446,13 +465,17 @@
         logger.warning("***** %s is GIVENBACK by %s *****"
                        % (self.buildqueue_record.specific_job.build.title,
                           self.buildqueue_record.builder.name))
-        self.storeBuildInfo(self, librarian, slave_status)
-        # XXX cprov 2006-05-30: Currently this information is not
-        # properly presented in the Web UI. We will discuss it in
-        # the next Paris Summit, infinity has some ideas about how
-        # to use this content. For now we just ensure it's stored.
-        self.buildqueue_record.builder.cleanSlave()
-        self.buildqueue_record.reset()
+        def build_info_stored(ignored):
+            # XXX cprov 2006-05-30: Currently this information is not
+            # properly presented in the Web UI. We will discuss it in
+            # the next Paris Summit, infinity has some ideas about how
+            # to use this content. For now we just ensure it's stored.
+            d = self.buildqueue_record.builder.cleanSlave()
+            self.buildqueue_record.reset()
+            return d
+
+        d = self.storeBuildInfo(self, librarian, slave_status)
+        return d.addCallback(build_info_stored)
 
 
 class PackageBuildSet:

=== modified file 'lib/lp/buildmaster/tests/mock_slaves.py'
--- lib/lp/buildmaster/tests/mock_slaves.py	2010-11-12 12:54:42 +0000
+++ lib/lp/buildmaster/tests/mock_slaves.py	2010-11-30 11:46:52 +0000
@@ -23,6 +23,7 @@
 
 import fixtures
 import os
+import types
 
 from StringIO import StringIO
 import xmlrpclib
@@ -161,6 +162,12 @@
         return self.sendFileToSlave(
             libraryfilealias.content.sha1, libraryfilealias.http_url)
 
+    def getFiles(self, filemap):
+        dl = defer.gatherResults([
+            self.getFile(builder_file, filemap[builder_file])
+            for builder_file in filemap])
+        return dl
+
 
 class BuildingSlave(OkSlave):
     """A mock slave that looks like it's currently building."""
@@ -175,13 +182,14 @@
         return defer.succeed(
             ('BuilderStatus.BUILDING', self.build_id, buildlog))
 
-    def getFile(self, sum):
-        # XXX: This needs to be updated to return a Deferred.
+    def getFile(self, sum, file_to_write):
         self.call_log.append('getFile')
         if sum == "buildlog":
-            s = StringIO("This is a build log")
-            s.headers = {'content-length': 19}
-            return s
+            if isinstance(file_to_write, types.StringTypes):
+                file_to_write = open(file_to_write, 'wb')
+            file_to_write.write("This is a build log")
+            file_to_write.close()
+        return defer.succeed(None)
 
 
 class WaitingSlave(OkSlave):
@@ -208,14 +216,15 @@
             'BuilderStatus.WAITING', self.state, self.build_id, self.filemap,
             self.dependencies))
 
-    def getFile(self, hash):
-        # XXX: This needs to be updated to return a Deferred.
+    def getFile(self, hash, file_to_write):
         self.call_log.append('getFile')
         if hash in self.valid_file_hashes:
             content = "This is a %s" % hash
-            s = StringIO(content)
-            s.headers = {'content-length': len(content)}
-            return s
+            if isinstance(file_to_write, types.StringTypes):
+                file_to_write = open(file_to_write, 'wb')
+            file_to_write.write(content)
+            file_to_write.close()
+        return defer.succeed(None)
 
 
 class AbortingSlave(OkSlave):
@@ -314,7 +323,7 @@
         Points to a fixed URL that is also used by `BuilddSlaveTestSetup`.
         """
         return BuilderSlave.makeBuilderSlave(
-            self.TEST_URL, 'vmhost', reactor, proxy)
+            self.BASE_URL, 'vmhost', reactor, proxy)
 
     def makeCacheFile(self, tachandler, filename):
         """Make a cache file available on the remote slave.

=== modified file 'lib/lp/buildmaster/tests/test_builder.py'
--- lib/lp/buildmaster/tests/test_builder.py	2010-11-26 18:36:09 +0000
+++ lib/lp/buildmaster/tests/test_builder.py	2010-11-30 11:46:52 +0000
@@ -5,6 +5,7 @@
 
 import os
 import signal
+import tempfile
 import xmlrpclib
 
 from testtools.deferredruntest import (
@@ -14,7 +15,10 @@
     SynchronousDeferredRunTest,
     )
 
-from twisted.internet.defer import CancelledError
+from twisted.internet.defer import (
+    CancelledError,
+    DeferredList,
+    )
 from twisted.internet.task import Clock
 from twisted.python.failure import Failure
 from twisted.web.client import getPage
@@ -1150,3 +1154,43 @@
             d = getPage(expected_url.encode('utf8'))
             return d.addCallback(self.assertEqual, content)
         return d.addCallback(check_file)
+
+    def test_getFiles(self):
+        # Test BuilderSlave.getFiles().
+        # It also implicitly tests getFile() - I don't want to test that
+        # separately because it increases test run time and it's going
+        # away at some point anyway, in favour of getFiles().
+        contents = ["content1", "content2", "content3"]
+        self.slave_helper.getServerSlave()
+        slave = self.slave_helper.getClientSlave()
+        filemap = {}
+        content_map = {}
+
+        def got_files(ignored):
+            # Called back when getFiles finishes.  Make sure all the
+            # content is as expected.
+            got_contents = []
+            for sha1 in filemap:
+                local_file = filemap[sha1]
+                file = open(local_file)
+                self.assertEqual(content_map[sha1], file.read())
+                file.close()
+
+        def finished_uploading(ignored):
+            d = slave.getFiles(filemap)
+            return d.addCallback(got_files)
+
+        # Set up some files on the builder and store details in
+        # content_map so we can compare downloads later.
+        dl = []
+        for content in contents:
+            filename = content + '.txt'
+            lf = self.factory.makeLibraryFileAlias(filename, content=content)
+            content_map[lf.content.sha1] = content
+            fd, filemap[lf.content.sha1] = tempfile.mkstemp()
+            self.addCleanup(os.remove, filemap[lf.content.sha1])
+            self.layer.txn.commit()
+            d = slave.ensurepresent(lf.content.sha1, lf.http_url, "", "")
+            dl.append(d)
+
+        return DeferredList(dl).addCallback(finished_uploading)

=== modified file 'lib/lp/buildmaster/tests/test_packagebuild.py'
--- lib/lp/buildmaster/tests/test_packagebuild.py	2010-10-28 09:11:36 +0000
+++ lib/lp/buildmaster/tests/test_packagebuild.py	2010-11-30 11:46:52 +0000
@@ -8,6 +8,7 @@
 from datetime import datetime
 import hashlib
 import os
+import shutil
 
 from storm.store import Store
 from zope.component import getUtility
@@ -43,6 +44,7 @@
     login_person,
     TestCaseWithFactory,
     )
+from lp.testing.factory import LaunchpadObjectFactory
 from lp.testing.fakemethod import FakeMethod
 
 
@@ -98,8 +100,6 @@
         self.assertRaises(
             NotImplementedError, self.package_build.verifySuccessfulUpload)
         self.assertRaises(NotImplementedError, self.package_build.notify)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
         self.assertRaises(
             NotImplementedError, self.package_build.handleStatus,
             None, None, None)
@@ -283,6 +283,7 @@
 class TestHandleStatusMixin:
     """Tests for `IPackageBuild`s handleStatus method.
 
+    This should be run with a Trial TestCase.
     """
 
     layer = LaunchpadZopelessLayer
@@ -293,6 +294,7 @@
 
     def setUp(self):
         super(TestHandleStatusMixin, self).setUp()
+        self.factory = LaunchpadObjectFactory()
         self.build = self.makeBuild()
         # For the moment, we require a builder for the build so that
         # handleStatus_OK can get a reference to the slave.
@@ -304,7 +306,10 @@
         builder.setSlaveForTesting(self.slave)
 
         # We overwrite the buildmaster root to use a temp directory.
-        self.upload_root = self.makeTemporaryDirectory()
+        tempdir = self.mktemp()
+        os.mkdir(tempdir)
+        self.addCleanup(shutil.rmtree, tempdir)
+        self.upload_root = tempdir
         tmp_builddmaster_root = """
         [builddmaster]
         root: %s
@@ -325,56 +330,58 @@
         # A filemap with plain filenames should not cause a problem.
         # The call to handleStatus will attempt to get the file from
         # the slave resulting in a URL error in this test case.
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        self.build.handleStatus('OK', None, {
+        def got_status(ignored):
+            self.assertEqual(BuildStatus.UPLOADING, self.build.status)
+            self.assertResultCount(1, "incoming")
+
+        d = self.build.handleStatus('OK', None, {
                 'filemap': {'myfile.py': 'test_file_hash'},
                 })
-
-        self.assertEqual(BuildStatus.UPLOADING, self.build.status)
-        self.assertResultCount(1, "incoming")
+        return d.addCallback(got_status)
 
     def test_handleStatus_OK_absolute_filepath(self):
         # A filemap that tries to write to files outside of
         # the upload directory will result in a failed upload.
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        self.build.handleStatus('OK', None, {
+        def got_status(ignored):
+            self.assertEqual(BuildStatus.FAILEDTOUPLOAD, self.build.status)
+            self.assertResultCount(0, "failed")
+            self.assertIdentical(None, self.build.buildqueue_record)
+
+        d = self.build.handleStatus('OK', None, {
             'filemap': {'/tmp/myfile.py': 'test_file_hash'},
             })
-        self.assertEqual(BuildStatus.FAILEDTOUPLOAD, self.build.status)
-        self.assertResultCount(0, "failed")
-        self.assertIs(None, self.build.buildqueue_record)
+        return d.addCallback(got_status)
 
     def test_handleStatus_OK_relative_filepath(self):
         # A filemap that tries to write to files outside of
         # the upload directory will result in a failed upload.
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        self.build.handleStatus('OK', None, {
+        def got_status(ignored):
+            self.assertEqual(BuildStatus.FAILEDTOUPLOAD, self.build.status)
+            self.assertResultCount(0, "failed")
+
+        d = self.build.handleStatus('OK', None, {
             'filemap': {'../myfile.py': 'test_file_hash'},
             })
-        self.assertEqual(BuildStatus.FAILEDTOUPLOAD, self.build.status)
-        self.assertResultCount(0, "failed")
+        return d.addCallback(got_status)
 
     def test_handleStatus_OK_sets_build_log(self):
         # The build log is set during handleStatus.
         removeSecurityProxy(self.build).log = None
         self.assertEqual(None, self.build.log)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        self.build.handleStatus('OK', None, {
+        d = self.build.handleStatus('OK', None, {
                 'filemap': {'myfile.py': 'test_file_hash'},
                 })
-        self.assertNotEqual(None, self.build.log)
+        def got_status(ignored):
+            self.assertNotEqual(None, self.build.log)
+        return d.addCallback(got_status)
 
     def test_date_finished_set(self):
         # The date finished is updated during handleStatus_OK.
         removeSecurityProxy(self.build).date_finished = None
         self.assertEqual(None, self.build.date_finished)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        self.build.handleStatus('OK', None, {
+        d = self.build.handleStatus('OK', None, {
                 'filemap': {'myfile.py': 'test_file_hash'},
                 })
-        self.assertNotEqual(None, self.build.date_finished)
+        def got_status(ignored):
+            self.assertNotEqual(None, self.build.date_finished)
+        return d.addCallback(got_status)

=== modified file 'lib/lp/code/model/sourcepackagerecipebuild.py'
--- lib/lp/code/model/sourcepackagerecipebuild.py	2010-11-11 20:56:21 +0000
+++ lib/lp/code/model/sourcepackagerecipebuild.py	2010-11-30 11:46:52 +0000
@@ -312,11 +312,13 @@
 
     def _handleStatus_OK(self, librarian, slave_status, logger):
         """See `IPackageBuild`."""
-        super(SourcePackageRecipeBuild, self)._handleStatus_OK(
+        d = super(SourcePackageRecipeBuild, self)._handleStatus_OK(
             librarian, slave_status, logger)
-        # base implementation doesn't notify on success.
-        if self.status == BuildStatus.FULLYBUILT:
-            self.notify()
+        def uploaded_build(ignored):
+            # Base implementation doesn't notify on success.
+            if self.status == BuildStatus.FULLYBUILT:
+                self.notify()
+        return d.addCallback(uploaded_build)
 
     def getUploader(self, changes):
         """See `IPackageBuild`."""

=== modified file 'lib/lp/code/model/tests/test_sourcepackagerecipebuild.py'
--- lib/lp/code/model/tests/test_sourcepackagerecipebuild.py	2010-10-27 14:20:21 +0000
+++ lib/lp/code/model/tests/test_sourcepackagerecipebuild.py	2010-11-30 11:46:52 +0000
@@ -14,6 +14,8 @@
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from twisted.trial.unittest import TestCase as TrialTestCase
+
 from canonical.launchpad.interfaces.lpstorm import IStore
 from canonical.launchpad.webapp.authorization import check_permission
 from canonical.launchpad.webapp.testing import verifyObject
@@ -424,7 +426,7 @@
 
 
 class TestHandleStatusForSPRBuild(
-    MakeSPRecipeBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
+    MakeSPRecipeBuildMixin, TestHandleStatusMixin, TrialTestCase):
     """IPackageBuild.handleStatus works with SPRecipe builds."""
 
 

=== modified file 'lib/lp/soyuz/tests/test_binarypackagebuild.py'
--- lib/lp/soyuz/tests/test_binarypackagebuild.py	2010-10-06 11:46:51 +0000
+++ lib/lp/soyuz/tests/test_binarypackagebuild.py	2010-11-30 11:46:52 +0000
@@ -13,6 +13,8 @@
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from twisted.trial.unittest import TestCase as TrialTestCase
+
 from canonical.testing.layers import LaunchpadZopelessLayer
 from lp.buildmaster.enums import BuildStatus
 from lp.buildmaster.interfaces.builder import IBuilderSet
@@ -466,5 +468,5 @@
 
 
 class TestHandleStatusForBinaryPackageBuild(
-    MakeBinaryPackageBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
+    MakeBinaryPackageBuildMixin, TestHandleStatusMixin, TrialTestCase):
     """IPackageBuild.handleStatus works with binary builds."""

=== modified file 'lib/lp/soyuz/tests/test_binarypackagebuildbehavior.py'
--- lib/lp/soyuz/tests/test_binarypackagebuildbehavior.py	2010-11-25 17:37:34 +0000
+++ lib/lp/soyuz/tests/test_binarypackagebuildbehavior.py	2010-11-30 11:46:52 +0000
@@ -435,37 +435,46 @@
         self.build.status = BuildStatus.FULLYBUILT
         old_tmps = sorted(os.listdir('/tmp'))
 
-        # Grabbing logs should not leave new files in /tmp (bug #172798)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        logfile_lfa_id = self.build.getLogFromSlave(self.build)
-        logfile_lfa = getUtility(ILibraryFileAliasSet)[logfile_lfa_id]
-        new_tmps = sorted(os.listdir('/tmp'))
-        self.assertEqual(old_tmps, new_tmps)
-
-        # The new librarian file is stored compressed with a .gz
-        # extension and text/plain file type for easy viewing in
-        # browsers, as it decompresses and displays the file inline.
-        self.assertTrue(logfile_lfa.filename.endswith('_FULLYBUILT.txt.gz'))
-        self.assertEqual('text/plain', logfile_lfa.mimetype)
-        self.layer.txn.commit()
-
-        # LibrarianFileAlias does not implement tell() or seek(), which
-        # are required by gzip.open(), so we need to read the file out
-        # of the librarian first.
-        fd, fname = tempfile.mkstemp()
-        self.addCleanup(os.remove, fname)
-        tmp = os.fdopen(fd, 'wb')
-        tmp.write(logfile_lfa.read())
-        tmp.close()
-        uncompressed_file = gzip.open(fname).read()
-
-        # XXX: 2010-10-18 bug=662631
-        # When the mock slave is changed to return a Deferred,
-        # update this test too.
-        orig_file = removeSecurityProxy(self.builder.slave).getFile(
-            'buildlog').read()
-        self.assertEqual(orig_file, uncompressed_file)
+        def got_log(logfile_lfa_id):
+            # Grabbing logs should not leave new files in /tmp (bug #172798)
+            logfile_lfa = getUtility(ILibraryFileAliasSet)[logfile_lfa_id]
+            new_tmps = sorted(os.listdir('/tmp'))
+            self.assertEqual(old_tmps, new_tmps)
+
+            # The new librarian file is stored compressed with a .gz
+            # extension and text/plain file type for easy viewing in
+            # browsers, as it decompresses and displays the file inline.
+            self.assertTrue(
+                logfile_lfa.filename.endswith('_FULLYBUILT.txt.gz'))
+            self.assertEqual('text/plain', logfile_lfa.mimetype)
+            self.layer.txn.commit()
+
+            # LibrarianFileAlias does not implement tell() or seek(), which
+            # are required by gzip.open(), so we need to read the file out
+            # of the librarian first.
+            fd, fname = tempfile.mkstemp()
+            self.addCleanup(os.remove, fname)
+            tmp = os.fdopen(fd, 'wb')
+            tmp.write(logfile_lfa.read())
+            tmp.close()
+            uncompressed_file = gzip.open(fname).read()
+
+            # Now make a temp filename that getFile() can write to.
+            fd, tmp_orig_file_name = tempfile.mkstemp()
+            self.addCleanup(os.remove, tmp_orig_file_name)
+
+            # Check that the original file from the slave matches the
+            # uncompressed file in the librarian.
+            def got_orig_log(ignored):
+                orig_file_content = open(tmp_orig_file_name).read()
+                self.assertEqual(orig_file_content, uncompressed_file)
+
+            d = removeSecurityProxy(self.builder.slave).getFile(
+                'buildlog', tmp_orig_file_name)
+            return d.addCallback(got_orig_log)
+
+        d = self.build.getLogFromSlave(self.build)
+        return d.addCallback(got_log)
 
     def test_private_build_log_storage(self):
         # Builds in private archives should have their log uploaded to

=== modified file 'lib/lp/translations/model/translationtemplatesbuildbehavior.py'
--- lib/lp/translations/model/translationtemplatesbuildbehavior.py	2010-10-27 14:25:19 +0000
+++ lib/lp/translations/model/translationtemplatesbuildbehavior.py	2010-11-30 11:46:52 +0000
@@ -11,6 +11,11 @@
     'TranslationTemplatesBuildBehavior',
     ]
 
+import os
+import tempfile
+
+from twisted.internet import defer
+
 from zope.component import getUtility
 from zope.interface import implements
 from zope.security.proxy import removeSecurityProxy
@@ -74,17 +79,20 @@
         """Read tarball with generated translation templates from slave."""
         if filemap is None:
             logger.error("Slave returned no filemap.")
-            return None
+            return defer.succeed(None)
 
         slave_filename = filemap.get(self.templates_tarball_path)
         if slave_filename is None:
             logger.error("Did not find templates tarball in slave output.")
-            return None
+            return defer.succeed(None)
 
         slave = removeSecurityProxy(buildqueue.builder.slave)
-        # XXX 2010-10-18 bug=662631
-        # Change this to do non-blocking IO.
-        return slave.getFile(slave_filename).read()
+
+        fd, fname = tempfile.mkstemp()
+        tarball_file = os.fdopen(fd, 'wb')
+        d = slave.getFile(slave_filename, tarball_file)
+        # getFile will close the file object.
+        return d.addCallback(lambda ignored: fname)
 
     def _uploadTarball(self, branch, tarball, logger):
         """Upload tarball to productseries that want it."""
@@ -120,20 +128,40 @@
             queue_item.specific_job.branch.bzr_identity,
             build_status))
 
+        def clean_slave(ignored):
+            d = queue_item.builder.cleanSlave()
+            return d.addCallback(lambda ignored: queue_item.destroySelf())
+
+        def got_tarball(filename):
+            # XXX 2010-11-12 bug=674575
+            # Please make addOrUpdateEntriesFromTarball() take files on
+            # disk; reading arbitrarily sized files into memory is
+            # dangerous.
+            if filename is None:
+                logger.error("Build produced no tarball.")
+                return
+
+            tarball_file = open(filename)
+            try:
+                tarball = tarball_file.read()
+                if tarball is None:
+                    logger.error("Build produced empty tarball.")
+                else:
+                    logger.debug("Uploading translation templates tarball.")
+                    self._uploadTarball(
+                        queue_item.specific_job.branch, tarball, logger)
+                    logger.debug("Upload complete.")
+            finally:
+                tarball_file.close()
+                os.remove(filename)
+
         if build_status == 'OK':
             logger.debug("Processing successful templates build.")
             filemap = slave_status.get('filemap')
-            # XXX 2010-10-18 bug=662631
-            # Change this to do non-blocking IO.
-            tarball = self._readTarball(queue_item, filemap, logger)
-
-            if tarball is None:
-                logger.error("Build produced no tarball.")
-            else:
-                logger.debug("Uploading translation templates tarball.")
-                self._uploadTarball(
-                    queue_item.specific_job.branch, tarball, logger)
-                logger.debug("Upload complete.")
-
-        d = queue_item.builder.cleanSlave()
-        return d.addCallback(lambda ignored: queue_item.destroySelf())
+            d = self._readTarball(queue_item, filemap, logger)
+            d.addCallback(got_tarball)
+            d.addCallback(clean_slave)
+            return d
+
+        return clean_slave(None)
+

=== modified file 'lib/lp/translations/tests/test_translationtemplatesbuildbehavior.py'
--- lib/lp/translations/tests/test_translationtemplatesbuildbehavior.py	2010-11-09 00:04:11 +0000
+++ lib/lp/translations/tests/test_translationtemplatesbuildbehavior.py	2010-11-30 11:46:52 +0000
@@ -14,9 +14,12 @@
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from twisted.internet import defer
+
 from canonical.config import config
 from canonical.launchpad.interfaces.launchpad import ILaunchpadCelebrities
 from canonical.launchpad.interfaces.librarian import ILibraryFileAliasSet
+from canonical.librarian.utils import copy_and_close
 from canonical.testing.layers import (
     LaunchpadZopelessLayer,
     )
@@ -166,9 +169,17 @@
         path = behavior.templates_tarball_path
         # Poke the file we're expecting into the mock slave.
         behavior._builder.slave.valid_file_hashes.append(path)
-        self.assertEqual(
-            "This is a %s" % path,
-            behavior._readTarball(buildqueue, {path: path}, logging))
+        def got_tarball(filename):
+            tarball = open(filename, 'r')
+            try:
+                self.assertEqual(
+                    "This is a %s" % path, tarball.read())
+            finally:
+                tarball.close()
+                os.remove(filename)
+
+        d = behavior._readTarball(buildqueue, {path: path}, logging)
+        return d.addCallback(got_tarball)
 
     def test_updateBuild_WAITING_OK(self):
         # Hopefully, a build will succeed and produce a tarball.
@@ -190,8 +201,10 @@
         def got_status(status):
             slave_call_log = behavior._builder.slave.call_log
             slave_status = {
-                'builder_status': status[0], 'build_status': status[1]}
-            behavior.updateSlaveStatus(status, slave_status)
+                'builder_status': status[0],
+                'build_status': status[1],
+                'filemap': {'translation-templates.tar.gz': 'foo'},
+                }
             return behavior.updateBuild_WAITING(
                 queue_item, slave_status, None, logging), slave_call_log
 
@@ -199,7 +212,7 @@
             slave_call_log = behavior._builder.slave.call_log
             self.assertEqual(1, queue_item.destroySelf.call_count)
             self.assertIn('clean', slave_call_log)
-            self.assertEqual(0, behavior._uploadTarball.call_count)
+            self.assertEqual(1, behavior._uploadTarball.call_count)
 
         d.addCallback(got_dispatch)
         d.addCallback(got_status)
@@ -292,12 +305,15 @@
 
         d = behavior.dispatchBuildToSlave(queue_item, logging)
 
-        def got_dispatch((status, info)):
+        def fake_getFile(sum, file):
             dummy_tar = os.path.join(
                 os.path.dirname(__file__), 'dummy_templates.tar.gz')
-            # XXX 2010-10-18 bug=662631
-            # Change this to do non-blocking IO.
-            builder.slave.getFile = lambda sum: open(dummy_tar)
+            tar_file = open(dummy_tar)
+            copy_and_close(tar_file, file)
+            return defer.succeed(None)
+
+        def got_dispatch((status, info)):
+            builder.slave.getFile = fake_getFile
             builder.slave.filemap = {
                 'translation-templates.tar.gz': 'foo'}
             return builder.slave.status()

=== modified file 'lib/lp_sitecustomize.py'
--- lib/lp_sitecustomize.py	2010-10-27 14:25:19 +0000
+++ lib/lp_sitecustomize.py	2010-11-30 11:46:52 +0000
@@ -9,7 +9,10 @@
 import warnings
 import logging
 
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import (
+    Deferred,
+    DeferredList,
+    )
 
 from bzrlib.branch import Branch
 from lp.services.log import loglevels
@@ -105,6 +108,7 @@
     customizeMimetypes()
     dont_wrap_class_and_subclasses(Branch)
     checker.BasicTypes.update({Deferred: checker.NoProxy})
+    checker.BasicTypes.update({DeferredList: checker.NoProxy})
     checker.BasicTypes[itertools.groupby] = checker._iteratorChecker
     # The itertools._grouper type is not exposed by name, so we must get it
     # through actually using itertools.groupby.