Merge lp:~maxb/udd/modular-threads into lp:udd

[Max Bowsher]

A chunk of splitting up icommon.

I ran a local mass_import.py to check it still worked.

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 6/15/2011 9:32 AM, Max Bowsher wrote:
> This includes planning and spawning imports, tracking their success or
> @@ -301,7 +302,7 @@
> # doing all the filesystem calls every milisecond but the user
> # shouldn't have to wait too long to see his modifications
> # taken into account either.
> - self.driver.stopped.wait(0.2)
> + self.driver.stopped.wait(5)
> if self.driver.stopped.isSet():
> break
> # Then check the graceful stop file
>

That is more than just moving the code around.

The rest seems fine to me, but I remember talking with vila quite a bit
about his threading patches, so I'd like him to take a look at this.

John
=:->
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk34nF0ACgkQJdeBCYSNAAPiwACeMTfzREOn35a2JYxSq4viAthn
yzcAn0YiAeITR4UjQxlR8JucUGr5k5Lq
=/CNm
-----END PGP SIGNATURE-----

[Vincent Ladeuil (vila)]

Please don't mix up refactoring and changes like:

40 - self.driver.stopped.wait(0.2)
41 + self.driver.stopped.wait(5)

What's the point of this change ? The comment above this line
clearly states that the value has been tweaked to match user
expectations. In this case, you're making the user waits up to 5
seconds before getting some feedback that the importer will stop
and there have been cases where a losa was trigger happy and
killed the job instead of letting it finish properly which lost
some valuable feedback (I don't remember the exact details, only
that killing wasn't the right solution).

I don't mind discussing this kind of change if you feel strongly
about it but let's do that in a specific proposal then.
For the rest, I prefer importing modules rather than symbols, but...
well, I won't block on this.

So feel free to land the rest.

review: Needs Fixing

[Max Bowsher (maxb)]

Whoops!

That was a temporary local hack whilst trying to figure out why mass_import was tight-looping for me locally. It should never have been committed.

[Max Bowsher (maxb)]

> I prefer importing modules rather than symbols, but...
> well, I won't block on this.

I do not have any strong preference in this case, so I am quite happy to go along with yours here.

I only feel strongly about importing symbols directly when the same symbol is used many times, and the advantage of brevity is significant.

[Vincent Ladeuil (vila)]

Cool, thanks.

review: Approve

=== modified file 'mass_import.py'
--- mass_import.py	2011-06-15 00:28:18 +0000
+++ mass_import.py	2011-06-15 14:39:30 +0000
@@ -17,9 +17,7 @@
 
 from udd import icommon
 from udd import iconfig
-
-#import httplib2
-#httplib2.debuglevel = 1
+from udd import threads
 
 
 class WatchedFileHandler(logging.FileHandler):
@@ -129,7 +127,7 @@
                    ''.join(traceback.format_exception(exc_class, exc_value,
                                                       exc_tb))))
 
-class Importer(icommon.SubprocessMonitor):
+class Importer(threads.SubprocessMonitor):
 
     def __init__(self, args, package_name, job_id):
         # FIXME: We should put the max_duration in a config file when it
@@ -223,7 +221,7 @@
                     self.tried = set()
 
 
-class ImportDriver(icommon.ThreadDriver):
+class ImportDriver(threads.ThreadDriver):
     """Monitor the ThreadedImporter.
 
     This includes planning and spawning imports, tracking their success or

=== modified file 'selftest.py'
--- selftest.py	2011-06-14 23:58:08 +0000
+++ selftest.py	2011-06-15 14:39:30 +0000
@@ -44,10 +44,10 @@
             'udd.tests.test_config',
             'udd.tests.test_import_list',
             'udd.tests.test_import_package',
-            'udd.tests.test_mass_import',
             'udd.tests.test_package_to_import',
             'udd.tests.test_revid_database',
             'udd.tests.test_status_database',
+            'udd.tests.test_threads',
             ]))
         self.test = suite
 

=== modified file 'udd/icommon.py'
--- udd/icommon.py	2011-06-15 06:54:16 +0000
+++ udd/icommon.py	2011-06-15 14:39:30 +0000
@@ -1,23 +1,17 @@
-#!/usr/bin/python
-
 import cgi
 import datetime
 import errno
 import fcntl
 import operator
 import os
-import Queue
 import random
 import re
 import shutil
 import simplejson
-import signal
 import socket
 import sqlite3
 import StringIO
-import subprocess
 import sys
-import threading
 import time
 import urllib
 import urllib2
@@ -42,12 +36,6 @@
     )
 from bzrlib.trace import mutter
 
-try:
-    from bzrlib import cethread
-except ImportError:
-    # Use our copy then
-    from udd import cethread
-
 from launchpadlib.credentials import Credentials
 from launchpadlib.errors import HTTPError
 from launchpadlib.launchpad import Launchpad, EDGE_SERVICE_ROOT
@@ -2007,234 +1995,6 @@
             f.close()
 
 
-class CatchingExceptionThread(cethread.CatchingExceptionThread):
-    """Isolating features that needs to be backported to bzrlib.
-
-    None so far.
-    """
-
-
-class SubprocessMonitor(CatchingExceptionThread):
-
-    def __init__(self, args, max_duration=None):
-        """Create a SubprocessMonitor.
-
-        :param args: A list of arguments to give to python.
-
-        :param max_duration: The max number of seconds the command is expected
-            to be run (elapsed time).
-        """
-        self.started = threading.Event()
-        self.stopped = threading.Event()
-        super(SubprocessMonitor, self).__init__(target=self.spawn,
-                                                sync_event=self.started)
-        self.args = args
-        self.proc_pid = None
-        self.retcode = None
-        self.out = self.err = None
-        self.started_at = self.ended_at = None
-        self.max_duration = max_duration
-        self.killed_with = None
-        self.killed_at = None
-
-    def spawn(self):
-        """Spawn the python command in a subprocess."""
-        proc = subprocess.Popen(['python'] + self.args,
-                                executable=sys.executable,
-                                stdout=subprocess.PIPE,
-                                stderr=subprocess.PIPE,
-                                stdin=subprocess.PIPE)
-        self.proc_pid = proc.pid
-        self.started_at = time.time()
-        self.switch_and_set(self.stopped)
-        try:
-            self.out, self.err = proc.communicate()
-            self.ended_at = time.time()
-            self.retcode = proc.returncode
-        except OSError, e:
-            if e.errno in (errno.ECHILD, errno.ESRCH):
-                # The process doesn't exist anymore
-                pass
-            else:
-                raise
-        self.stopped.set()
-
-    def collect(self, timeout=None):
-        """Collect relevant data before discarding the thread.
-
-        This must be called by the controlling thread. Be aware that since this
-        calls join() it may raise the pending exception which must be cleared
-        if this behaviour is not the intended one.
-
-        Daughter classes can override this method to collect any relevant data
-        (including the pending exception) before the thread is discarded.
-        """
-        self.join(timeout)
-
-    def kill_with_escalation(self, grace_period=None):
-        """Attempt to kill the subprocess.
-
-        :param grace_period: A floating number of seconds to wait before using
-            SIGKILL to kill the subprocess.
-
-        When first called, `signal.SIGTERM` is used to allow the subprocess to
-        handle the kill cleanly.  If called repeatedly, this will not try to
-        kill the process again until the end of the specified
-        `grace_period`. When the grace period expires, `signal.SIGKILL` is used
-        so that subprocesses that don't shutdown when receiving
-        `signal.SIGTERM` can still be shut down.
-
-        This must be called by the controlling thread.
-        """
-        killed = False
-        if self.killed_with is None:
-            killed = self._kill(signal.SIGTERM)
-        else:
-            # We already tried to kill this this subprocess, if it didn't
-            # during the grace period, we'll kill it harder
-            now = time.time()
-            if grace_period is None:
-                # FIXME: This should go into a config file -- vila 2011-03-18
-                # 10 seconds should be more than enough for a process to die or
-                # something wrong is happening in which case it's better to get
-                # this problematic subprocess down as quickly as possible.
-                grace_period = 10.0
-            if now - self.killed_at > grace_period:
-                killed = self._kill(signal.SIGKILL)
-        return killed
-
-    def _kill(self, signum):
-        killed = False
-        # Don't attempt to kill the subprocess if it hasn't been started
-        if self.proc_pid is not None:
-            try:
-                os.kill(self.proc_pid, signum)
-                killed = True
-                self.killed_with = signum
-                self.killed_at = time.time()
-            except OSError, e:
-                if e.errno in (errno.ECHILD, errno.ESRCH):
-                    # The process doesn't exist anymore.
-                    pass
-                else:
-                    raise
-        return killed
-
-    def has_exceeded_time_quota(self, duration=None):
-        """Is the thread running for too long ?
-
-        :param duration: If specified will override the ``max_duration``
-            attribute.
-        """
-        if duration is None:
-            duration = self.max_duration
-        if (duration is not None
-            and self.started_at is not None and self.ended_at is None):
-            # A timeout is defined, the thread has been started but is not
-            # finished yet
-            if (time.time() - self.started_at) > self.max_duration:
-                return True
-        return False
-
-
-class ThreadDriver(CatchingExceptionThread):
-
-    def __init__(self, queue, max_threads, target=None):
-        """Create a ThreadDriver.
-
-        :param queue: A thread provider.
-
-        :param max_threads: Maximum number of threads allowed to run
-            concurrently.
-
-        :param target: What callable should be run in the thread.
-        """
-        self.started = threading.Event()
-        self.stopped = threading.Event()
-        self.stop = threading.Event()
-        if target is None:
-            target = self.drive
-        super(ThreadDriver, self).__init__(target=target,
-                                           sync_event=self.started)
-        self.queue = queue
-        self.queue_closed = threading.Event()
-        self.max_threads = max_threads
-        self.threads = []
-
-    def drive(self):
-        self.before_start()
-        # From now on the caller will let us run until we stop or die
-        self.switch_and_set(self.stopped)
-        while True:
-            # Calling wait ensures we can be interrupted properly, but we don't
-            # want to block so we use a timeout. We don't use 0 though or other
-            # threads may not get any cycle otherwise.
-            self.stop.wait(0.001)
-            if self.stop.isSet():
-                break
-            # Check for threads exceeding their expected lifetime
-            self.check_time_quota()
-            self.do_one_step()
-        self.before_stop()
-        self.stopped.set()
-
-    def before_start(self):
-        pass
-
-    def check_time_quota(self):
-        """Ensure threads don't exceed their elapsed time quota.
-
-        :returns: The list of killed threads.
-        """
-        killed = []
-        for t in self.threads:
-            if t.has_exceeded_time_quota():
-                # Die you sucker
-                t.kill_with_escalation()
-                killed.append(t)
-        return killed
-
-    def do_one_step(self):
-        self.collect_terminated_threads()
-        if (not self.stop.isSet() and not self.queue_closed.isSet()
-            and len(self.threads) < self.max_threads):
-            self.start_new_thread()
-        if self.queue_closed.isSet() and not self.threads:
-            # The queue is closed and there are no more running threads
-            self.stop.set()
-
-    def before_stop(self):
-        self.kill_remaining_threads()
-
-    def start_new_thread(self):
-        # Get a new thread from the queue and start it
-        try:
-            new_thread = self.queue.get_nowait()
-        except Queue.Empty:
-            return None
-        self.threads.append(new_thread)
-        new_thread.start()
-        new_thread.started.wait()
-        # Tell the queue we're done with this thread
-        self.queue.task_done()
-        return new_thread
-
-    def collect_terminated_threads(self):
-        # If there are terminated threads, collect them
-        for t in self.threads[:]:
-            if not t.isAlive():
-                t.collect()
-                self.threads.remove(t)
-
-    def kill_remaining_threads(self):
-        # Kill all the remaining threads
-        for t in self.threads[:]:
-            t.kill_with_escalation()
-            # Collect the thread whether we could kill it or not
-            t.collect()
-            self.threads.remove(t)
-
-
 def refresh_possible_transports(possible_transports):
     """Prune stale connections from possible_transports.
     

=== renamed file 'udd/tests/test_mass_import.py' => 'udd/tests/test_threads.py'
--- udd/tests/test_mass_import.py	2011-06-15 00:28:18 +0000
+++ udd/tests/test_threads.py	2011-06-15 14:39:30 +0000
@@ -4,11 +4,9 @@
 import threading
 import time
 
-
 from bzrlib import tests
 
-
-from udd import icommon
+from udd import threads
 
 
 def get_test_script_path(name):
@@ -28,7 +26,7 @@
 
     def run_in_subprocess(self, args, klass=None):
         if klass is None:
-            klass = icommon.SubprocessMonitor
+            klass = threads.SubprocessMonitor
         sub = klass(args)
         self.addCleanup(sub.join, 0)
         sub.start()
@@ -83,7 +81,7 @@
     def test_kill_catch_zombie(self):
         kill_me = threading.Event()
         resume = threading.Event()
-        class TestMonitor(icommon.SubprocessMonitor):
+        class TestMonitor(threads.SubprocessMonitor):
 
             def switch_and_set(self, new):
                 super(TestMonitor, self).switch_and_set(new)
@@ -121,7 +119,7 @@
 
     def run_in_subprocess(self, args, klass=None):
         if klass is None:
-            klass = icommon.SubprocessMonitor
+            klass = threads.SubprocessMonitor
         sub = klass(args)
         self.addCleanup(sub.join, 0)
         sub.start()
@@ -187,7 +185,7 @@
         self.assertEquals(sub.out, 'Try harder\n')
 
 
-class Sleeper(icommon.SubprocessMonitor):
+class Sleeper(threads.SubprocessMonitor):
 
     def __init__(self, sleep_time=0.001, max_duration=None):
         # sleep can't be killed (or interrupted) so we don't sleep for long but
@@ -206,7 +204,7 @@
     def get_driver(self, *args, **kwargs):
         klass = kwargs.pop('klass', None)
         if klass is None:
-            klass = icommon.ThreadDriver
+            klass = threads.ThreadDriver
         return klass(*args, **kwargs)
 
     def get_started_driver(self, target=None, override=None, queue=None,
@@ -286,7 +284,7 @@
         self.started = threading.Event()
         self.resumed = threading.Event()
         self.terminated = threading.Event()
-        class TestThread(icommon.SubprocessMonitor):
+        class TestThread(threads.SubprocessMonitor):
 
             def collect(thread):
                 self.assertIs(self.thread, thread)
@@ -297,7 +295,7 @@
         q = Queue.Queue(1)
         q.put(self.thread)
 
-        class TestDriver(icommon.ThreadDriver):
+        class TestDriver(threads.ThreadDriver):
 
             def start_new_thread(driver):
                 t = super(TestDriver, driver).start_new_thread()
@@ -326,7 +324,7 @@
         self.assertLength(0, driver.threads)
 
     def test_thread_exception_raised_in_driver(self):
-        class TestThread(icommon.SubprocessMonitor):
+        class TestThread(threads.SubprocessMonitor):
 
             def collect(thread):
                 self.assertEquals(1, thread.retcode)
@@ -436,7 +434,7 @@
         self.addCleanup(sleeper.kill_with_escalation)
         q.put(sleeper)
         self.killed = None
-        class TestDriver(icommon.ThreadDriver):
+        class TestDriver(threads.ThreadDriver):
 
             def check_time_quota(driver):
                 killed = super(TestDriver, driver).check_time_quota()

=== added file 'udd/threads.py'
--- udd/threads.py	1970-01-01 00:00:00 +0000
+++ udd/threads.py	2011-06-15 14:39:30 +0000
@@ -0,0 +1,242 @@
+import errno
+import os
+import Queue
+import signal
+import subprocess
+import sys
+import threading
+import time
+
+try:
+    from bzrlib import cethread
+except ImportError:
+    # Use our copy then
+    from udd import cethread
+
+
+class CatchingExceptionThread(cethread.CatchingExceptionThread):
+    """Isolating features that needs to be backported to bzrlib.
+
+    None so far.
+    """
+
+
+class SubprocessMonitor(CatchingExceptionThread):
+
+    def __init__(self, args, max_duration=None):
+        """Create a SubprocessMonitor.
+
+        :param args: A list of arguments to give to python.
+
+        :param max_duration: The max number of seconds the command is expected
+            to be run (elapsed time).
+        """
+        self.started = threading.Event()
+        self.stopped = threading.Event()
+        super(SubprocessMonitor, self).__init__(target=self.spawn,
+                                                sync_event=self.started)
+        self.args = args
+        self.proc_pid = None
+        self.retcode = None
+        self.out = self.err = None
+        self.started_at = self.ended_at = None
+        self.max_duration = max_duration
+        self.killed_with = None
+        self.killed_at = None
+
+    def spawn(self):
+        """Spawn the python command in a subprocess."""
+        proc = subprocess.Popen(['python'] + self.args,
+                                executable=sys.executable,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                stdin=subprocess.PIPE)
+        self.proc_pid = proc.pid
+        self.started_at = time.time()
+        self.switch_and_set(self.stopped)
+        try:
+            self.out, self.err = proc.communicate()
+            self.ended_at = time.time()
+            self.retcode = proc.returncode
+        except OSError, e:
+            if e.errno in (errno.ECHILD, errno.ESRCH):
+                # The process doesn't exist anymore
+                pass
+            else:
+                raise
+        self.stopped.set()
+
+    def collect(self, timeout=None):
+        """Collect relevant data before discarding the thread.
+
+        This must be called by the controlling thread. Be aware that since this
+        calls join() it may raise the pending exception which must be cleared
+        if this behaviour is not the intended one.
+
+        Daughter classes can override this method to collect any relevant data
+        (including the pending exception) before the thread is discarded.
+        """
+        self.join(timeout)
+
+    def kill_with_escalation(self, grace_period=None):
+        """Attempt to kill the subprocess.
+
+        :param grace_period: A floating number of seconds to wait before using
+            SIGKILL to kill the subprocess.
+
+        When first called, `signal.SIGTERM` is used to allow the subprocess to
+        handle the kill cleanly.  If called repeatedly, this will not try to
+        kill the process again until the end of the specified
+        `grace_period`. When the grace period expires, `signal.SIGKILL` is used
+        so that subprocesses that don't shutdown when receiving
+        `signal.SIGTERM` can still be shut down.
+
+        This must be called by the controlling thread.
+        """
+        killed = False
+        if self.killed_with is None:
+            killed = self._kill(signal.SIGTERM)
+        else:
+            # We already tried to kill this this subprocess, if it didn't
+            # during the grace period, we'll kill it harder
+            now = time.time()
+            if grace_period is None:
+                # FIXME: This should go into a config file -- vila 2011-03-18
+                # 10 seconds should be more than enough for a process to die or
+                # something wrong is happening in which case it's better to get
+                # this problematic subprocess down as quickly as possible.
+                grace_period = 10.0
+            if now - self.killed_at > grace_period:
+                killed = self._kill(signal.SIGKILL)
+        return killed
+
+    def _kill(self, signum):
+        killed = False
+        # Don't attempt to kill the subprocess if it hasn't been started
+        if self.proc_pid is not None:
+            try:
+                os.kill(self.proc_pid, signum)
+                killed = True
+                self.killed_with = signum
+                self.killed_at = time.time()
+            except OSError, e:
+                if e.errno in (errno.ECHILD, errno.ESRCH):
+                    # The process doesn't exist anymore.
+                    pass
+                else:
+                    raise
+        return killed
+
+    def has_exceeded_time_quota(self, duration=None):
+        """Is the thread running for too long ?
+
+        :param duration: If specified will override the ``max_duration``
+            attribute.
+        """
+        if duration is None:
+            duration = self.max_duration
+        if (duration is not None
+            and self.started_at is not None and self.ended_at is None):
+            # A timeout is defined, the thread has been started but is not
+            # finished yet
+            if (time.time() - self.started_at) > self.max_duration:
+                return True
+        return False
+
+
+class ThreadDriver(CatchingExceptionThread):
+
+    def __init__(self, queue, max_threads, target=None):
+        """Create a ThreadDriver.
+
+        :param queue: A thread provider.
+
+        :param max_threads: Maximum number of threads allowed to run
+            concurrently.
+
+        :param target: What callable should be run in the thread.
+        """
+        self.started = threading.Event()
+        self.stopped = threading.Event()
+        self.stop = threading.Event()
+        if target is None:
+            target = self.drive
+        super(ThreadDriver, self).__init__(target=target,
+                                           sync_event=self.started)
+        self.queue = queue
+        self.queue_closed = threading.Event()
+        self.max_threads = max_threads
+        self.threads = []
+
+    def drive(self):
+        self.before_start()
+        # From now on the caller will let us run until we stop or die
+        self.switch_and_set(self.stopped)
+        while True:
+            # Calling wait ensures we can be interrupted properly, but we don't
+            # want to block so we use a timeout. We don't use 0 though or other
+            # threads may not get any cycle otherwise.
+            self.stop.wait(0.001)
+            if self.stop.isSet():
+                break
+            # Check for threads exceeding their expected lifetime
+            self.check_time_quota()
+            self.do_one_step()
+        self.before_stop()
+        self.stopped.set()
+
+    def before_start(self):
+        pass
+
+    def check_time_quota(self):
+        """Ensure threads don't exceed their elapsed time quota.
+
+        :returns: The list of killed threads.
+        """
+        killed = []
+        for t in self.threads:
+            if t.has_exceeded_time_quota():
+                # Die you sucker
+                t.kill_with_escalation()
+                killed.append(t)
+        return killed
+
+    def do_one_step(self):
+        self.collect_terminated_threads()
+        if (not self.stop.isSet() and not self.queue_closed.isSet()
+            and len(self.threads) < self.max_threads):
+            self.start_new_thread()
+        if self.queue_closed.isSet() and not self.threads:
+            # The queue is closed and there are no more running threads
+            self.stop.set()
+
+    def before_stop(self):
+        self.kill_remaining_threads()
+
+    def start_new_thread(self):
+        # Get a new thread from the queue and start it
+        try:
+            new_thread = self.queue.get_nowait()
+        except Queue.Empty:
+            return None
+        self.threads.append(new_thread)
+        new_thread.start()
+        new_thread.started.wait()
+        # Tell the queue we're done with this thread
+        self.queue.task_done()
+        return new_thread
+
+    def collect_terminated_threads(self):
+        # If there are terminated threads, collect them
+        for t in self.threads[:]:
+            if not t.isAlive():
+                t.collect()
+                self.threads.remove(t)
+
+    def kill_remaining_threads(self):
+        # Kill all the remaining threads
+        for t in self.threads[:]:
+            t.kill_with_escalation()
+            # Collect the thread whether we could kill it or not
+            t.collect()
+            self.threads.remove(t)