Merge lp:~mwhudson/launchpad/puller-job-scheduling into lp:launchpad

[Michael Hudson-Doyle]

[Michael Hudson-Doyle (mwhudson)]

Hi jml,

This long awaited branch makes the puller scheduler work in a job-style fashion.

In some sense, after all the preliminary branches, it's fairly simple, applying the task infrastructure.

A fair portion of the changes from the lines-of-diff POV are due to the fact that the cronscript doesn't take an argument any more.

I'm not sure if the JobScheduler needs more tests, tell me what you think.

Cheers,
mwh

[Jonathan Lange (jml)]

On Thu, Jul 23, 2009 at 1:00 PM, Michael
Hudson<email address hidden> wrote:
> Michael Hudson has proposed merging lp:~mwhudson/launchpad/puller-job-scheduling into lp:~launchpad-pqm/launchpad/devel.
>
> Requested reviews:
>    Jonathan Lange (jml)
>
> Hi jml,
>
> This long awaited branch makes the puller scheduler work in a job-style fashion.
>
> In some sense, after all the preliminary branches, it's fairly simple, applying the task infrastructure.
>

Hooray!

> A fair portion of the changes from the lines-of-diff POV are due to the fact that the cronscript doesn't take an argument any more.
>

Yeah I see.

I can't recall that we ever discussed the impact of mirroring all
types of branches with the one process. I mention it in the review
below.

> I'm not sure if the JobScheduler needs more tests, tell me what you think.
>

I don't think it does.

> === modified file 'configs/development/launchpad-lazr.conf'
> --- configs/development/launchpad-lazr.conf 2009-07-16 21:12:16 +0000
> +++ configs/development/launchpad-lazr.conf 2009-07-23 02:07:29 +0000
> @@ -185,6 +185,10 @@
> [rosetta]
> global_suggestions_enabled: True
>
> +[supermirror_puller]
> +error_dir: /var/tmp/codehosting.test
> +oops_prefix: SMP
> +

If we're adding a new section, do you think we could call it something better
than supermirror_puller? branch_puller perhaps?

> === modified file 'cronscripts/supermirror-pull.py'
> --- cronscripts/supermirror-pull.py 2009-06-24 20:52:01 +0000
> +++ cronscripts/supermirror-pull.py 2009-07-23 02:07:29 +0000
> @@ -12,7 +12,6 @@
> from twisted.python import log as tplog
> from twisted.web.xmlrpc import Proxy
>
> -from lp.code.enums import BranchType
> from lp.codehosting.puller import mirror, scheduler
> from canonical.config import config
> from canonical.launchpad.scripts import logger_options
> @@ -40,26 +39,11 @@
> parser = OptionParser()
> logger_options(parser)
> (options, arguments) = parser.parse_args()
> - which = arguments.pop(0)
> if arguments:
> parser.error("Unhandled arguments %s" % repr(arguments))
> -
> - branch_type_map = {
> - 'upload': BranchType.HOSTED,
> - 'mirror': BranchType.MIRRORED,
> - 'import': BranchType.IMPORTED
> - }
> -
> - try:
> - branch_type = branch_type_map[which]
> - except KeyError:
> - parser.error(
> - 'Expected one of %s, but got: %r'
> - % (branch_type_map.keys(), which))
> -
> - log = set_up_logging_for_script(options, 'supermirror_%s_puller' % which)
> + log = set_up_logging_for_script(options, 'supermirror_puller')
> manager = scheduler.JobScheduler(
> - Proxy(config.codehosting.branch_puller_endpoint), log, branch_type)
> + Proxy(config.codehosting.branch_puller_endpoint), log)
>

It's pleasing to see this complexity go. I guess this threatens to increase
the lag for hosted branches. Do we need to worry about this?

> reactor.callWhenRunning(run_mirror, log, manager)
> reactor.run()
>
> === modified file 'lib/canonical/config/schema-lazr.conf'
> --- lib/canonical/config/schema-lazr.conf 2009-07-16 21:12:16 +0000
> +++ lib/canonical/config/sch...

[Jonathan Lange (jml)]

All of which combines to be needs-fixing.

review: Needs Fixing (code)

[Michael Hudson-Doyle (mwhudson)]

Jonathan Lange wrote:
> On Thu, Jul 23, 2009 at 1:00 PM, Michael
> Hudson<email address hidden> wrote:
>> Michael Hudson has proposed merging lp:~mwhudson/launchpad/puller-job-scheduling into lp:~launchpad-pqm/launchpad/devel.
>>
>> Requested reviews:
>> Jonathan Lange (jml)
>>
>> Hi jml,
>>
>> This long awaited branch makes the puller scheduler work in a job-style fashion.
>>
>> In some sense, after all the preliminary branches, it's fairly simple, applying the task infrastructure.
>>
>
> Hooray!
>
>> A fair portion of the changes from the lines-of-diff POV are due to the fact that the cronscript doesn't take an argument any more.
>>
>
> Yeah I see.
>
> I can't recall that we ever discussed the impact of mirroring all
> types of branches with the one process. I mention it in the review
> below.

Hmm, I guess we should up the limit from the current 4 per type to say,
10 across all types. I also increased the worker_timeout, which should
help the initial pull of large branches.

We should set up some graphs to measure the problem at least. It would
be easy enough to graph the number of branches waiting to be pulled, I
don't think we keep enough information currently to track how long
branches wait on average. We will when we move to a BranchJob though --
which should be a fairly easy follow up to this branch.

>> I'm not sure if the JobScheduler needs more tests, tell me what you think.
>>
>
> I don't think it does.

Cool.

>> === modified file 'configs/development/launchpad-lazr.conf'
>> --- configs/development/launchpad-lazr.conf 2009-07-16 21:12:16 +0000
>> +++ configs/development/launchpad-lazr.conf 2009-07-23 02:07:29 +0000
>> @@ -185,6 +185,10 @@
>> [rosetta]
>> global_suggestions_enabled: True
>>
>> +[supermirror_puller]
>> +error_dir: /var/tmp/codehosting.test
>> +oops_prefix: SMP
>> +
>
> If we're adding a new section, do you think we could call it something better
> than supermirror_puller? branch_puller perhaps?

Probably makes sense.

>> === modified file 'cronscripts/supermirror-pull.py'
>> --- cronscripts/supermirror-pull.py 2009-06-24 20:52:01 +0000
>> +++ cronscripts/supermirror-pull.py 2009-07-23 02:07:29 +0000
>> @@ -12,7 +12,6 @@
>> from twisted.python import log as tplog
>> from twisted.web.xmlrpc import Proxy
>>
>> -from lp.code.enums import BranchType
>> from lp.codehosting.puller import mirror, scheduler
>> from canonical.config import config
>> from canonical.launchpad.scripts import logger_options
>> @@ -40,26 +39,11 @@
>> parser = OptionParser()
>> logger_options(parser)
>> (options, arguments) = parser.parse_args()
>> - which = arguments.pop(0)
>> if arguments:
>> parser.error("Unhandled arguments %s" % repr(arguments))
>> -
>> - branch_type_map = {
>> - 'upload': BranchType.HOSTED,
>> - 'mirror': BranchType.MIRRORED,
>> - 'import': BranchType.IMPORTED
>> - }
>> -
>> - try:
>> - branch_type = branch_type_map[which]
>> - except KeyError:
>> - parser.error(
>> - 'Expected one of %s, but got: %r'
>> - % (branch_type_map.keys(), which))
>> -
>> - log = set_up_logging_for_script(options, '...

[Jonathan Lange (jml)]

Thanks for that.

review: Approve (code)

=== modified file 'configs/development/launchpad-lazr.conf'
--- configs/development/launchpad-lazr.conf	2009-07-16 21:12:16 +0000
+++ configs/development/launchpad-lazr.conf	2009-07-23 02:07:29 +0000
@@ -185,6 +185,10 @@
 [rosetta]
 global_suggestions_enabled: True
 
+[supermirror_puller]
+error_dir: /var/tmp/codehosting.test
+oops_prefix: SMP
+
 [supermirror_import_puller]
 error_dir: /var/tmp/codehosting.test
 oops_prefix: ISMP

=== modified file 'cronscripts/supermirror-pull.py'
--- cronscripts/supermirror-pull.py	2009-06-24 20:52:01 +0000
+++ cronscripts/supermirror-pull.py	2009-07-23 02:07:29 +0000
@@ -12,7 +12,6 @@
 from twisted.python import log as tplog
 from twisted.web.xmlrpc import Proxy
 
-from lp.code.enums import BranchType
 from lp.codehosting.puller import mirror, scheduler
 from canonical.config import config
 from canonical.launchpad.scripts import logger_options
@@ -40,26 +39,11 @@
     parser = OptionParser()
     logger_options(parser)
     (options, arguments) = parser.parse_args()
-    which = arguments.pop(0)
     if arguments:
         parser.error("Unhandled arguments %s" % repr(arguments))
-
-    branch_type_map = {
-        'upload': BranchType.HOSTED,
-        'mirror': BranchType.MIRRORED,
-        'import': BranchType.IMPORTED
-        }
-
-    try:
-        branch_type = branch_type_map[which]
-    except KeyError:
-        parser.error(
-            'Expected one of %s, but got: %r'
-            % (branch_type_map.keys(), which))
-
-    log = set_up_logging_for_script(options, 'supermirror_%s_puller' % which)
+    log = set_up_logging_for_script(options, 'supermirror_puller')
     manager = scheduler.JobScheduler(
-        Proxy(config.codehosting.branch_puller_endpoint), log, branch_type)
+        Proxy(config.codehosting.branch_puller_endpoint), log)
 
     reactor.callWhenRunning(run_mirror, log, manager)
     reactor.run()

=== modified file 'lib/canonical/config/schema-lazr.conf'
--- lib/canonical/config/schema-lazr.conf	2009-07-16 21:12:16 +0000
+++ lib/canonical/config/schema-lazr.conf	2009-07-23 02:07:29 +0000
@@ -1513,6 +1513,17 @@
 maximum_workers: 4
 
 
+[supermirror_puller]
+# See [error_reports].
+error_dir: none
+
+# See [error_reports].
+oops_prefix: none
+
+# See [error_reports].
+copy_to_zlog: false
+
+
 [supermirror_import_puller]
 # See [error_reports].
 error_dir: none

=== modified file 'lib/lp/codehosting/puller/scheduler.py'
--- lib/lp/codehosting/puller/scheduler.py	2009-06-25 04:06:00 +0000
+++ lib/lp/codehosting/puller/scheduler.py	2009-07-23 02:35:18 +0000
@@ -18,7 +18,6 @@
 import os
 from StringIO import StringIO
 import socket
-import sys
 
 from twisted.internet import defer, error, reactor
 from twisted.protocols.basic import NetstringReceiver, NetstringParseError
@@ -28,7 +27,8 @@
 
 import canonical
 from canonical.cachedproperty import cachedproperty
-from lp.codehosting.vfs import branch_id_to_path
+from canonical.twistedsupport.task import (
+    ParallelLimitedTaskConsumer, PollingTaskSource)
 from lp.codehosting.puller.worker import (
     get_canonical_url_for_branch_name)
 from lp.codehosting.puller import get_lock_id_for_branch_id
@@ -276,7 +276,7 @@
         'scripts/mirror-branch.py')
     protocol_class = PullerMonitorProtocol
 
-    def __init__(self, branch_id, source_url, unique_name, branch_type,
+    def __init__(self, branch_id, source_url, unique_name, branch_type_name,
                  default_stacked_on_url, logger, client,
                  available_oops_prefixes):
         """Construct a PullerMaster object.
@@ -300,10 +300,9 @@
         """
         self.branch_id = branch_id
         self.source_url = source_url.strip()
-        path = branch_id_to_path(branch_id)
         self.destination_url = 'lp-mirrored:///%s' % (unique_name,)
         self.unique_name = unique_name
-        self.branch_type = branch_type
+        self.branch_type_name = branch_type_name
         self.default_stacked_on_url = default_stacked_on_url
         self.logger = logger
         self.branch_puller_endpoint = client
@@ -335,7 +334,7 @@
         command = [
             interpreter, self.path_to_script, self.source_url,
             self.destination_url, str(self.branch_id), str(self.unique_name),
-            self.branch_type.name, self.oops_prefix,
+            self.branch_type_name, self.oops_prefix,
             self.default_stacked_on_url]
         self.logger.debug("executing %s", command)
         env = os.environ.copy()
@@ -416,12 +415,11 @@
     branches.
     """
 
-    def __init__(self, branch_puller_endpoint, logger, branch_type):
+    def __init__(self, branch_puller_endpoint, logger):
         self.branch_puller_endpoint = branch_puller_endpoint
         self.logger = logger
         self.actualLock = None
-        self.branch_type = branch_type
-        self.name = 'branch-puller-%s' % branch_type.name.lower()
+        self.name = 'branch-puller'
         self.lockfilename = '/var/lock/launchpad-%s.lock' % self.name
 
     @cachedproperty
@@ -435,43 +433,36 @@
         return set(
             [str(i) for i in range(config.supermirror.maximum_workers)])
 
-    def _run(self, puller_masters):
-        """Run all branches_to_mirror registered with the JobScheduler."""
-        self.logger.info('%d branches to mirror', len(puller_masters))
-        assert config.supermirror.maximum_workers is not None, (
-            "config.supermirror.maximum_workers is not defined.")
-        semaphore = defer.DeferredSemaphore(
+    def _turnJobTupleIntoTask(self, job_tuple):
+        if len(job_tuple) == 0:
+            return None
+        (branch_id, pull_url, unique_name,
+         default_stacked_on_url, branch_type_name) = job_tuple
+        master = PullerMaster(
+            branch_id, pull_url, unique_name, branch_type_name,
+            default_stacked_on_url, self.logger,
+            self.branch_puller_endpoint, self.available_oops_prefixes)
+        return master.run
+
+    def _poll(self):
+        deferred = self.branch_puller_endpoint.callRemote(
+            'acquireBranchToPull')
+        deferred.addCallback(self._turnJobTupleIntoTask)
+        return deferred
+
+    def run(self):
+        consumer = ParallelLimitedTaskConsumer(
             config.supermirror.maximum_workers)
-        deferreds = [
-            semaphore.run(puller_master.run)
-            for puller_master in puller_masters]
-        deferred = defer.gatherResults(deferreds)
+        self.consumer = consumer
+        source = PollingTaskSource(10, self._poll)
+        deferred = consumer.consume(source)
         deferred.addCallback(self._finishedRunning)
         return deferred
 
-    def run(self):
-        deferred = self.branch_puller_endpoint.callRemote(
-            'getBranchPullQueue', self.branch_type.name)
-        deferred.addCallback(self.getPullerMasters)
-        deferred.addCallback(self._run)
-        return deferred
-
     def _finishedRunning(self, ignored):
         self.logger.info('Mirroring complete')
         return ignored
 
-    def getPullerMaster(self, branch_id, branch_src, unique_name,
-                        default_stacked_on_url):
-        branch_src = branch_src.strip()
-        return PullerMaster(
-            branch_id, branch_src, unique_name, self.branch_type,
-            default_stacked_on_url, self.logger,
-            self.branch_puller_endpoint, self.available_oops_prefixes)
-
-    def getPullerMasters(self, branches_to_pull):
-        return [
-            self.getPullerMaster(*branch) for branch in branches_to_pull]
-
     def lock(self):
         self.actualLock = GlobalLock(self.lockfilename)
         try:

=== modified file 'lib/lp/codehosting/puller/tests/test_acceptance.py'
--- lib/lp/codehosting/puller/tests/test_acceptance.py	2009-06-25 04:06:00 +0000
+++ lib/lp/codehosting/puller/tests/test_acceptance.py	2009-07-23 02:35:18 +0000
@@ -10,7 +10,6 @@
 import os
 import shutil
 from subprocess import PIPE, Popen
-import sys
 import unittest
 from urlparse import urlparse
 
@@ -123,7 +122,7 @@
         output, error = process.communicate()
         return process.returncode, output, error
 
-    def runPuller(self, branch_type):
+    def runPuller(self):
         """Run the puller script for the given branch type.
 
         :param branch_type: One of 'upload', 'mirror' or 'import'
@@ -133,7 +132,7 @@
             stdout and stderr respectively.
         """
         command = [
-            '%s/bin/py' % config.root, self._puller_script, '-q', branch_type]
+            '%s/bin/py' % config.root, self._puller_script, '-q']
         retcode, output, error = self.runSubprocess(command)
         return command, retcode, output, error
 
@@ -214,7 +213,7 @@
         db_branch = self.factory.makeAnyBranch(branch_type=BranchType.HOSTED)
         transaction.commit()
         self.pushBranch(db_branch)
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch)
 
@@ -225,7 +224,7 @@
         transaction.commit()
         pack_tree = self.make_branch_and_tree('pack', format='pack-0.92')
         self.pushBranch(db_branch, tree=pack_tree)
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch)
         # Then we upgrade the to a different format and ask for it to be
@@ -234,7 +233,7 @@
         transaction.begin()
         db_branch.requestMirror()
         transaction.commit()
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch)
 
@@ -244,7 +243,7 @@
         transaction.commit()
         loom_tree = self.makeLoomBranchAndTree('loom')
         self.pushBranch(db_branch, tree=loom_tree)
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch)
 
@@ -257,7 +256,7 @@
             branch=db_branch, person=accessing_user)
         transaction.commit()
         self.pushBranch(db_branch)
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch, accessing_user=accessing_user)
 
@@ -267,7 +266,7 @@
             branch_type=BranchType.MIRRORED)
         tree = self.setUpMirroredBranch(db_branch)
         transaction.commit()
-        command, retcode, output, error = self.runPuller('mirror')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         self.assertMirrored(db_branch, source_branch=tree.branch)
 
@@ -292,13 +291,11 @@
         series.branch = default_branch
         # Arrange for it to be pulled.
         if branch_type == BranchType.HOSTED:
-            puller_type = 'upload'
             transaction.commit()
             # For hosted branches, we just push it into the hosted area via
             # the codehosting vfs.
             self.pushBranch(default_branch)
         elif branch_type == BranchType.MIRRORED:
-            puller_type = 'mirror'
             # For mirrored branches, we serve the branch over HTTP, point the
             # database branch at this HTTP server and call requestMirror()
             self.setUpMirroredBranch(default_branch, format='1.6')
@@ -308,7 +305,7 @@
                 "don't know how to make a %s default branch"
                 % branch_type.TITLE)
         # Pull it.
-        command, retcode, output, error = self.runPuller(puller_type)
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         return default_branch
 
@@ -320,7 +317,7 @@
             branch_type=BranchType.MIRRORED, product=default_branch.product)
         tree = self.setUpMirroredBranch(db_branch, format='1.6')
         transaction.commit()
-        command, retcode, output, error = self.runPuller('mirror')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         mirrored_branch = self.assertMirrored(
             db_branch, source_branch=tree.branch)
@@ -339,7 +336,7 @@
             branch_type=BranchType.HOSTED, product=default_branch.product)
         transaction.commit()
         self.pushBranch(db_branch, format='1.6')
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         mirrored_branch = self.assertMirrored(db_branch)
         self.assertEqual(
@@ -369,7 +366,7 @@
         branch_config.set_option(
             'stacked_on_location',
             'bzr+ssh://bazaar.launchpad.dev/' + default_branch.unique_name)
-        command, retcode, output, error = self.runPuller('upload')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         # We clear the stacking information again here so that assertMirrored
         # can open the branch in the hosted area.
@@ -388,7 +385,7 @@
 
         tree = self.setUpMirroredBranch(db_branch, format='1.6')
         transaction.commit()
-        command, retcode, output, error = self.runPuller('mirror')
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
         mirrored_branch = self.assertMirrored(
             db_branch, source_branch=tree.branch)
@@ -423,26 +420,26 @@
         self.serveOverHTTP(self._getImportMirrorPort())
 
         # Run the puller.
-        command, retcode, output, error = self.runPuller("import")
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
 
         self.assertMirrored(db_branch, source_branch=tree.branch)
 
     def test_mirror_empty(self):
         # Run the puller on an empty pull queue.
-        command, retcode, output, error = self.runPuller("upload")
+        command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
 
     def test_records_script_activity(self):
         # A record gets created in the ScriptActivity table.
         script_activity_set = getUtility(IScriptActivitySet)
         self.assertIs(
-            script_activity_set.getLastActivity("branch-puller-hosted"),
+            script_activity_set.getLastActivity("branch-puller"),
             None)
-        self.runPuller("upload")
+        self.runPuller()
         transaction.abort()
         self.assertIsNot(
-            script_activity_set.getLastActivity("branch-puller-hosted"),
+            script_activity_set.getLastActivity("branch-puller"),
             None)
 
     # Possible tests to add:
@@ -455,4 +452,3 @@
 
 def test_suite():
     return unittest.TestLoader().loadTestsFromName(__name__)
-    return unittest.TestSuite()

=== modified file 'lib/lp/codehosting/puller/tests/test_scheduler.py'
--- lib/lp/codehosting/puller/tests/test_scheduler.py	2009-06-25 04:06:00 +0000
+++ lib/lp/codehosting/puller/tests/test_scheduler.py	2009-07-23 02:35:18 +0000
@@ -73,55 +73,31 @@
 
     def setUp(self):
         self.masterlock = 'master.lock'
-        # We set the log level to CRITICAL so that the log messages
-        # are suppressed.
-        logging.basicConfig(level=logging.CRITICAL)
 
     def tearDown(self):
         reset_logging()
-
-    def makeFakeClient(self, hosted, mirrored, imported):
-        return FakePullerEndpointProxy(
-            {'HOSTED': hosted, 'MIRRORED': mirrored, 'IMPORTED': imported})
-
-    def makeJobScheduler(self, branch_type, branch_tuples):
-        if branch_type == BranchType.HOSTED:
-            client = self.makeFakeClient(branch_tuples, [], [])
-        elif branch_type == BranchType.MIRRORED:
-            client = self.makeFakeClient([], branch_tuples, [])
-        elif branch_type == BranchType.IMPORTED:
-            client = self.makeFakeClient([], [], branch_tuples)
-        else:
-            self.fail("Unknown branch type: %r" % (branch_type,))
-        return scheduler.JobScheduler(
-            client, logging.getLogger(), branch_type)
+        if os.path.exists(self.masterlock):
+            os.unlink(self.masterlock)
+
+    def makeJobScheduler(self):
+        return scheduler.JobScheduler(None, logging.getLogger())
 
     def testManagerCreatesLocks(self):
-        try:
-            manager = self.makeJobScheduler(BranchType.HOSTED, [])
-            manager.lockfilename = self.masterlock
-            manager.lock()
-            self.failUnless(os.path.exists(self.masterlock))
-            manager.unlock()
-        finally:
-            self._removeLockFile()
+        manager = self.makeJobScheduler()
+        manager.lockfilename = self.masterlock
+        manager.lock()
+        self.failUnless(os.path.exists(self.masterlock))
+        manager.unlock()
 
     def testManagerEnforcesLocks(self):
-        try:
-            manager = self.makeJobScheduler(BranchType.HOSTED, [])
-            manager.lockfilename = self.masterlock
-            manager.lock()
-            anothermanager = self.makeJobScheduler(BranchType.HOSTED, [])
-            anothermanager.lockfilename = self.masterlock
-            self.assertRaises(scheduler.LockError, anothermanager.lock)
-            self.failUnless(os.path.exists(self.masterlock))
-            manager.unlock()
-        finally:
-            self._removeLockFile()
-
-    def _removeLockFile(self):
-        if os.path.exists(self.masterlock):
-            os.unlink(self.masterlock)
+        manager = self.makeJobScheduler()
+        manager.lockfilename = self.masterlock
+        manager.lock()
+        anothermanager = self.makeJobScheduler()
+        anothermanager.lockfilename = self.masterlock
+        self.assertRaises(scheduler.LockError, anothermanager.lock)
+        self.failUnless(os.path.exists(self.masterlock))
+        manager.unlock()
 
 
 class TestPullerWireProtocol(TrialTestCase):
@@ -541,7 +517,6 @@
     def setUp(self):
         from twisted.internet import reactor
         self.factory = ObjectFactory()
-        status_client = FakePullerEndpointProxy()
         self.available_oops_prefixes = set(['foo'])
         self.eventHandler = self.makePullerMaster(
             BranchType.HOSTED, oops_prefixes=self.available_oops_prefixes)
@@ -825,8 +800,7 @@
 
         # We need to create a branch at the destination_url, so that the
         # subprocess can actually create a lock.
-        destination_branch = BzrDir.create_branch_convenience(
-            puller_master.destination_url)
+        BzrDir.create_branch_convenience(puller_master.destination_url)
 
         deferred = puller_master.mirror().addErrback(self._dumpError)
 
@@ -935,8 +909,7 @@
 
         # We need to create a branch at the destination_url, so that the
         # subprocess can actually create a lock.
-        destination_branch = BzrDir.create_branch_convenience(
-            locking_puller_master.destination_url)
+        BzrDir.create_branch_convenience(locking_puller_master.destination_url)
 
         # Because when the deferred returned by 'func' is done we kill the
         # locking subprocess, we know that when the subprocess is done, the