Merge lp:~stub/launchpad/garbo into lp:launchpad

[Stuart Bishop]

Let garbo tasks timeout while waiting for replication lag or long running transactions

garbo-*.py lets a timeout be set for tasks so they abort if they don't complete withing a given time frame. This timeout is not checked for when blocking for long running transactions or for replication lag.

This branch fixes that. Now tasks will correctly be aborted when they timeout and report this, rather than block indefinitely.

[Jeroen T. Vermeulen (jtv)]

Hi Stuart,

Nice to see loop tuning continue to get better. One design question: your timeout check accompanies time.sleep calls. Why not have:

 * An attribute flag that says "I consider myself timed out" (even if it hasn't happened yet but is bound to happen because of impending sleep)

 * A _checkTimeLeft(seconds=0) method that you call in various places to set the timeout flag if the requisite number of seconds isn't available

 * A self.sleep method that checks for remaining time and time.sleeps only if the sleep ends before the deadline, so you don't have to check this for every time.sleep

 * Instead of time checks, a check for the timeout flag.

On a more nitpicky note, _is_timed_out is not the right spelling for a method; that'd be _isTimedOut. And I don't see the point of a leading underscore in _msg_counter.

Jeroen

review: Needs Information

[Stuart Bishop (stub)]

On Mon, Apr 5, 2010 at 7:29 PM, Jeroen T. Vermeulen <email address hidden> wrote:

> Nice to see loop tuning continue to get better.  One design question: your timeout check accompanies time.sleep calls.  Why not have:
>
>  * An attribute flag that says "I consider myself timed out" (even if it hasn't happened yet but is bound to happen because of impending sleep)
>
>  * A _checkTimeLeft(seconds=0) method that you call in various places to set the timeout flag if the requisite number of seconds isn't available
>
>  * A self.sleep method that checks for remaining time and time.sleeps only if the sleep ends before the deadline, so you don't have to check this for every time.sleep
>
>  * Instead of time checks, a check for the timeout flag.

I've added a self._sleep method to simplify the 'if not timed out
sleep' stanzas.

Having a flag in addition to the check method is less clear, and we
are only doing this a few times per second so it isn't a performance
issue.

> On a more nitpicky note, _is_timed_out is not the right spelling for a method; that'd be _isTimedOut.  And I don't see the point of a leading underscore in _msg_counter.

I've fixed both of these. And changed the 'extra' parameter to 'extra seconds'.

--
Stuart Bishop <email address hidden>
http://www.stuartbishop.net/

[Jeroen T. Vermeulen (jtv)]

The reason I brought up an "I'm timed out" flag was concern that the tuner might decide to go straight back to work after a sleep was short-circuited because it would have timed out. But turning some "while True" loops into ones with timeout checks in the loop condition is better. Approved.

review: Approve (code)

=== modified file 'lib/canonical/launchpad/doc/looptuner.txt'
--- lib/canonical/launchpad/doc/looptuner.txt	2009-07-14 05:50:46 +0000
+++ lib/canonical/launchpad/doc/looptuner.txt	2010-04-07 16:01:48 +0000
@@ -72,7 +72,7 @@
     ...         self.clock = 0
     ...
     ...     def isDone(self):
-    ...         done = (self.iteration > len(self.timings)+1)
+    ...         done = self.iteration >= len(self.timings)
     ...         if done:
     ...             print "done"
     ...         return done
@@ -80,6 +80,8 @@
     ...     def __call__(self, chunk_size):
     ...         print_change(self.last_chunk_size, chunk_size)
     ...         self.last_chunk_size = chunk_size
+    ...         self.clock += self.timings[self.iteration]
+    ...         self.iteration += 1
 
 
 In combination with that, we tweak LoopTuner to simulate the timings we gave
@@ -88,12 +90,7 @@
 
     >>> class TestTuner(LoopTuner):
     ...     def _time(self):
-    ...         clock_now = float(self.operation.clock)
-    ...         if self.operation.iteration < len(self.operation.timings):
-    ...             next = self.operation.timings[self.operation.iteration]
-    ...             self.operation.clock += next
-    ...         self.operation.iteration += 1
-    ...         return clock_now
+    ...         return float(self.operation.clock)
 
 
 === Trivial Case ===
@@ -103,7 +100,6 @@
     >>> body = PlannedLoop([])
     >>> loop = TestTuner(body, goal_seconds, 100)
     >>> loop.run()
-    start
     done
 
 
@@ -131,7 +127,6 @@
     increased (0.0)
     same (0.0)
     decreased (-0.0)
-    same (0.0)
     done
 
 
@@ -152,7 +147,6 @@
     same (0.0)
     same (0.0)
     same (0.0)
-    same (0.0)
     done
 
 
@@ -177,7 +171,6 @@
     decreased (-0.1)
     increased (0.1)
     same (0.0)
-    same (0.0)
     done
 
 

=== modified file 'lib/canonical/launchpad/utilities/looptuner.py'
--- lib/canonical/launchpad/utilities/looptuner.py	2010-03-25 12:22:13 +0000
+++ lib/canonical/launchpad/utilities/looptuner.py	2010-04-07 16:01:48 +0000
@@ -94,17 +94,21 @@
         else:
             self.log = log
 
+    def _isTimedOut(self, extra_seconds=0):
+        """Return True if the task will be timed out in extra_seconds."""
+        return self.abort_time is not None and (
+            self._time() + extra_seconds >= self.start_time + self.abort_time)
+
     def run(self):
         """Run the loop to completion."""
         chunk_size = self.minimum_chunk_size
         iteration = 0
         total_size = 0
-        start_time = self._time()
-        last_clock = start_time
+        self.start_time = self._time()
+        last_clock = self.start_time
         while not self.operation.isDone():
 
-            if (self.abort_time is not None
-                and last_clock >= start_time + self.abort_time):
+            if self._isTimedOut():
                 self.log.warn(
                     "Task aborted after %d seconds." % self.abort_time)
                 break
@@ -137,7 +141,7 @@
             chunk_size = min(chunk_size, self.maximum_chunk_size)
             iteration += 1
 
-        total_time = last_clock - start_time
+        total_time = last_clock - self.start_time
         average_size = total_size/max(1, iteration)
         average_speed = total_size/max(1, total_time)
         self.log.debug(
@@ -161,7 +165,7 @@
         if self.cooldown_time is None or self.cooldown_time <= 0.0:
             return bedtime
         else:
-            time.sleep(self.cooldown_time)
+            self._sleep(self.cooldown_time)
             return self._time()
 
     def _time(self):
@@ -172,6 +176,15 @@
         """
         return time.time()
 
+    def _sleep(self, seconds):
+        """Sleep.
+
+        If the sleep interval would put us over the tasks timeout,
+        do nothing.
+        """
+        if not self._isTimedOut(self.cooldown_time):
+            time.sleep(self.cooldown_time)
+
 
 def timedelta_to_seconds(td):
     return 24 * 60 * td.days + td.seconds
@@ -208,7 +221,7 @@
         """When database replication lag is high, block until it drops."""
         # Lag is most meaningful on the master.
         store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)
-        while True:
+        while not self._isTimedOut():
             lag = store.execute("SELECT replication_lag()").get_one()[0]
             if lag is None:
                 lag = timedelta(seconds=0)
@@ -225,7 +238,7 @@
                 % time_to_sleep)
 
             transaction.abort() # Don't become a long running transaction!
-            time.sleep(time_to_sleep)
+            self._sleep(time_to_sleep)
 
     def _blockForLongRunningTransactions(self):
         """If there are long running transactions, block to avoid making
@@ -233,7 +246,8 @@
         if self.long_running_transaction is None:
             return
         store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)
-        while True:
+        msg_counter = 0
+        while not self._isTimedOut():
             results = list(store.execute("""
                 SELECT
                     CURRENT_TIMESTAMP - xact_start,
@@ -247,13 +261,18 @@
                 """ % self.long_running_transaction).get_all())
             if not results:
                 break
-            for runtime, procpid, usename, datname, query in results:
-                self.log.info(
-                    "Blocked on %s old xact %s@%s/%d - %s."
-                    % (runtime, usename, datname, procpid, query))
-            self.log.info("Sleeping for 5 minutes.")
+
+            # Check for long running transactions every 30 seconds, but
+            # only report every 10 minutes to avoid log spam.
+            msg_counter += 1
+            if msg_counter % 20 == 1:
+                for runtime, procpid, usename, datname, query in results:
+                    self.log.info(
+                        "Blocked on %s old xact %s@%s/%d - %s."
+                        % (runtime, usename, datname, procpid, query))
+                self.log.info("Sleeping for up to 10 minutes.")
             transaction.abort() # Don't become a long running transaction!
-            time.sleep(5*60)
+            self._sleep(30)
 
     def _coolDown(self, bedtime):
         """As per LoopTuner._coolDown, except we always wait until there
@@ -263,7 +282,7 @@
         self._blockWhenLagged()
         if self.cooldown_time is not None and self.cooldown_time > 0.0:
             remaining_nap = self._time() - bedtime + self.cooldown_time
-            if remaining_nap > 0.0:
+            if remaining_nap > 0.0 and not self._is_timed_out(remaining_nap):
                 time.sleep(remaining_nap)
         return self._time()