Merge lp:~mwhudson/launchpad/smarter-code-import-scheduling-bug-236973 into lp:launchpad

[Michael Hudson-Doyle]

Ask for work until none is provided in the code import dispatcher.

[Michael Hudson-Doyle (mwhudson)]

Hi Tim,

We've talked about this I think -- I modify the code import dispatcher to look for work until it finds none and sleeps for a period dependent on the load of the machine between the asks.

[Tim Penhey (thumper)]

Looks good. I'm eager to see how this works on staging :)

review: Approve

=== modified file 'cronscripts/code-import-dispatcher.py'
--- cronscripts/code-import-dispatcher.py	2010-02-22 01:36:30 +0000
+++ cronscripts/code-import-dispatcher.py	2010-02-23 00:34:17 +0000
@@ -37,7 +37,7 @@
         globalErrorUtility.configure('codeimportdispatcher')
 
         dispatcher = CodeImportDispatcher(self.logger, self.options.max_jobs)
-        dispatcher.findAndDispatchJob(
+        dispatcher.findAndDispatchJobs(
             ServerProxy(config.codeimportdispatcher.codeimportscheduler_url))
 
 

=== modified file 'lib/lp/codehosting/codeimport/dispatcher.py'
--- lib/lp/codehosting/codeimport/dispatcher.py	2010-02-22 01:36:30 +0000
+++ lib/lp/codehosting/codeimport/dispatcher.py	2010-02-23 00:34:17 +0000
@@ -16,6 +16,7 @@
 import os
 import socket
 import subprocess
+import time
 
 from canonical.config import config
 
@@ -32,13 +33,14 @@
     worker_script = os.path.join(
         config.root, 'scripts', 'code-import-worker-db.py')
 
-    def __init__(self, logger, worker_limit):
+    def __init__(self, logger, worker_limit, _sleep=time.sleep):
         """Initialize an instance.
 
         :param logger: A `Logger` object.
         """
         self.logger = logger
         self.worker_limit = worker_limit
+        self._sleep = _sleep
 
     def getHostname(self):
         """Return the hostname of this machine.
@@ -65,15 +67,38 @@
 
 
     def findAndDispatchJob(self, scheduler_client):
-        """Check for and dispatch a job if necessary."""
+        """Check for and dispatch a job if necessary.
+
+        :return: A boolean, true if a job was found and dispatched.
+        """
 
         job_id = scheduler_client.getJobForMachine(
             self.getHostname(), self.worker_limit)
 
         if job_id == 0:
             self.logger.info("No jobs pending.")
-            return
+            return False
 
         self.logger.info("Dispatching job %d." % job_id)
 
         self.dispatchJob(job_id)
+        return True
+
+    def _getSleepInterval(self):
+        """How long to sleep for until asking for a new job.
+
+        The basic idea is to wait longer if the machine is more heavily
+        loaded, so that less loaded slaves get a chance to grab some jobs.
+
+        We assume worker_limit will be roughly the number of CPUs in the
+        machine, so load/worker_limit is roughly how loaded the machine is.
+        """
+        return 5*os.getloadavg()[0]/self.worker_limit
+
+    def findAndDispatchJobs(self, scheduler_client):
+        """Call findAndDispatchJob until no job is found."""
+        while True:
+            found = self.findAndDispatchJob(scheduler_client)
+            if not found:
+                break
+            self._sleep(self._getSleepInterval())

=== modified file 'lib/lp/codehosting/codeimport/tests/test_dispatcher.py'
--- lib/lp/codehosting/codeimport/tests/test_dispatcher.py	2010-02-22 02:06:57 +0000
+++ lib/lp/codehosting/codeimport/tests/test_dispatcher.py	2010-02-23 00:34:17 +0000
@@ -24,11 +24,11 @@
 class StubSchedulerClient:
     """A scheduler client that returns a pre-arranged answer."""
 
-    def __init__(self, id_to_return):
-        self.id_to_return = id_to_return
+    def __init__(self, ids_to_return):
+        self.ids_to_return = ids_to_return
 
     def getJobForMachine(self, machine, limit):
-        return self.id_to_return
+        return self.ids_to_return.pop(0)
 
 
 class MockSchedulerClient:
@@ -51,9 +51,10 @@
         TestCase.setUp(self)
         self.pushConfig('codeimportdispatcher', forced_hostname='none')
 
-    def makeDispatcher(self, worker_limit=10):
+    def makeDispatcher(self, worker_limit=10, _sleep=lambda delay: None):
         """Make a `CodeImportDispatcher`."""
-        return CodeImportDispatcher(QuietFakeLogger(), worker_limit)
+        return CodeImportDispatcher(
+            QuietFakeLogger(), worker_limit, _sleep=_sleep)
 
     def test_getHostname(self):
         # By default, getHostname return the same as socket.gethostname()
@@ -111,16 +112,16 @@
         calls = []
         dispatcher = self.makeDispatcher()
         dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
-        dispatcher.findAndDispatchJob(StubSchedulerClient(10))
-        self.assertEqual([10], calls)
+        found = dispatcher.findAndDispatchJob(StubSchedulerClient([10]))
+        self.assertEqual(([10], True), (calls, found))
 
     def test_findAndDispatchJob_noJobWaiting(self):
         # If there is no job to dispatch, then we just exit quietly.
         calls = []
         dispatcher = self.makeDispatcher()
         dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
-        dispatcher.findAndDispatchJob(StubSchedulerClient(0))
-        self.assertEqual([], calls)
+        found = dispatcher.findAndDispatchJob(StubSchedulerClient([0]))
+        self.assertEqual(([], False), (calls, found))
 
     def test_findAndDispatchJob_calls_getJobForMachine_with_limit(self):
         # findAndDispatchJob calls getJobForMachine on the scheduler client
@@ -133,5 +134,29 @@
             [(dispatcher.getHostname(), worker_limit)],
             scheduler_client.calls)
 
+    def test_findAndDispatchJobs(self):
+        # findAndDispatchJobs calls getJobForMachine on the scheduler_client,
+        # dispatching jobs, until it indicates that there are no more jobs to
+        # dispatch.
+        calls = []
+        dispatcher = self.makeDispatcher()
+        dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
+        dispatcher.findAndDispatchJobs(StubSchedulerClient([10, 9, 0]))
+        self.assertEqual([10, 9], calls)
+
+    def test_findAndDispatchJobs_sleeps(self):
+        # After finding a job, findAndDispatchJobs sleeps for an interval as
+        # returned by _getSleepInterval.
+        sleep_calls = []
+        interval = self.factory.getUniqueInteger()
+        def _sleep(delay):
+            sleep_calls.append(delay)
+        dispatcher = self.makeDispatcher(_sleep=_sleep)
+        dispatcher.dispatchJob = lambda job_id: None
+        dispatcher._getSleepInterval = lambda : interval
+        dispatcher.findAndDispatchJobs(StubSchedulerClient([10, 0]))
+        self.assertEqual([interval], sleep_calls)
+
+
 def test_suite():
     return TestLoader().loadTestsFromName(__name__)