Merge lp:~vila/udd/use-new-driver into lp:udd

[Vincent Ladeuil]

Build on top of lp:~vila/udd/robust-driver , this simplify the mass_import.py script.

ThreadedImporter has been renamed Importer for clarity, and just redefine a couple of methods mainly for logger calls() and db access.

AllQueue has been adapted to plug into the new design and will need more work, but this step is just to replace the existing implementation.

ImportDriver also redefine a couple of methods for the same reasons (logger and db).

ImportController uses the new ImportDriver features (well, ThreadDriver really) to simplify the control.

One feature has been lost here: changing the number of threads dynamically. I only realize that after doing a bunch of tests on package-import.local so I'll add it again in a next submission (or during review).

Note that this refactoring addresses the 'still active' spamming issue (but there wasn't a bug number for that right ?).

bug #589523 (adding a timeout for imports) will be far easier to fix now.

[lp:~vila/udd/use-new-driver]

425. By Vincent Ladeuil on 2011-02-22

Renamed SubprocessThread to SubprocessMonitor (and fixes lp:714420)

[Vincent Ladeuil (vila)]

There *was* a bug: #714420. So it's fixed now :)

[lp:~vila/udd/use-new-driver]

426. By Vincent Ladeuil on 2011-02-22

Merge robust-driver into use-new-driver

427. By Vincent Ladeuil on 2011-02-22

This also fixes bug #717204

[James Westby (james-w)]

157 + except Queue.Empty:
158 + self.unfinished_tasks += 1
159 + job_id, package_name = self.next_job()

186 + except StopIteration:
187 + raise Queue.Empty

Won't this mean that Queue.Empty would bubble to the top and stop the importer in this case?
Is that intentional?

This looks like a good simplification, thanks.

James

review: Needs Information

[Vincent Ladeuil (vila)]

> 157 + except Queue.Empty:
> 158 + self.unfinished_tasks += 1
> 159 + job_id, package_name = self.next_job()
>
> 186 + except StopIteration:
> 187 + raise Queue.Empty
>
> Won't this mean that Queue.Empty would

be caught by start_new_thread which handles it as meaning: there is no job for me yet, return None.

> bubble to the top and stop the importer

So no.

> in this case?
> Is that intentional?

Yup, it's a nice property of Queue that you could chain them this way almost transparently (first the jobs, then the packages (and I intentionally left the interrupted ones re-queuing in the controller since that occurs only once)).

If needed, we could add some priority handling in the queue itself, but I think a better plan is to maintain a debian and a lp canaris that will just suspend the execution of imports when one of the sites is down (and kill and requeue the ones in progress at the top of the queue)

> This looks like a good simplification, thanks.

Very much appreciated :-)

[James Westby (james-w)]

> > 157 + except Queue.Empty:
> > 158 + self.unfinished_tasks += 1
> > 159 + job_id, package_name = self.next_job()
> >
> > 186 + except StopIteration:
> > 187 + raise Queue.Empty
> >
> > Won't this mean that Queue.Empty would
>
> be caught by start_new_thread which handles it as meaning: there is no job for
> me yet, return None.

Ah missed that.

> If needed, we could add some priority handling in the queue itself, but I
> think a better plan is to maintain a debian and a lp canaris that will just
> suspend the execution of imports when one of the sites is down (and kill and
> requeue the ones in progress at the top of the queue)

I think the canaries are a good idea. I still think that the priority handling
in the old code is useful. It's useful to be able to requeue a package as it is
blocking someone and have it done as soon as a slot is free. I've never felt the
need for more fine-grained than the three levels we have now though (ASAP, soon, and
idle).

Thanks,

James

review: Approve

[Vincent Ladeuil (vila)]

>
> > If needed, we could add some priority handling in the queue itself, but I
> > think a better plan is to maintain a debian and a lp canaris that will just
> > suspend the execution of imports when one of the sites is down (and kill and
> > requeue the ones in progress at the top of the queue)
>
> I think the canaries are a good idea. I still think that the priority handling
> in the old code is useful.

The one in the db ? Oh my, I wouldn't touch that, it's perfect there.

> It's useful to be able to requeue a package as it
> is
> blocking someone and have it done as soon as a slot is free. I've never felt
> the
> need for more fine-grained than the three levels we have now though (ASAP,
> soon, and
> idle).

Yup. Good point about re-queuing on-demand, but that's already covered by requeue-package --priority no ? ISTM the way you wrote next_job() should give us that since it queries the db for each job... I'm not familiar enough with SQL (nor the requeue_package implementation) but if the query takes the priority into account my implementation will respect it.

When I was mentioning multiple queues/priority I was thinking about the losa proxied queries we may need in the future (including of course requeue-package or any other python script) which is the subject of my next submission.

[James Westby (james-w)]

On Wed, 23 Feb 2011 21:27:58 -0000, Vincent Ladeuil <email address hidden> wrote:
> The one in the db ? Oh my, I wouldn't touch that, it's perfect there.

Exactly.

> Yup. Good point about re-queuing on-demand, but that's already covered
> by requeue-package --priority no ?

Indeed.

> ISTM the way you wrote next_job()
> should give us that since it queries the db for each job... I'm not
> familiar enough with SQL (nor the requeue_package implementation) but
> if the query takes the priority into account my implementation will
> respect it.

That's perfect then. IIRC the SQL does an ORDER BY prio DESC or similar.

> When I was mentioning multiple queues/priority I was thinking about
> the losa proxied queries we may need in the future (including of
> course requeue-package or any other python script) which is the
> subject of my next submission.

Ok, let's discuss it then.

I was merely trying to highlight the utility of the priority scheme that
is in place now, rather than arguing against any specific change.

Thanks,

James

[Vincent Ladeuil (vila)]

> Ok, let's discuss it then.

Cool, we are in agreement then, see you on my next proposal ;)

> I was merely trying to highlight the utility of the priority scheme that
> is in place now,

Sorry, I should have been clearer about that but I don't fully understand yet all details (which is also why I tried to not change these parts :)

=== modified file 'icommon.py'
--- icommon.py	2011-02-22 16:11:07 +0000
+++ icommon.py	2011-02-22 16:11:07 +0000
@@ -1938,13 +1938,13 @@
     """
 
 
-class SubprocessThread(CatchingExceptionThread):
+class SubprocessMonitor(CatchingExceptionThread):
 
     def __init__(self, cmd):
         self.started = threading.Event()
         self.stopped = threading.Event()
-        super(SubprocessThread, self).__init__(target=self.spawn,
-                                               sync_event=self.started)
+        super(SubprocessMonitor, self).__init__(target=self.spawn,
+                                                sync_event=self.started)
         self.cmd = cmd
         self.proc_pid = None
         self.retcode = None

=== modified file 'mass_import.py'
--- mass_import.py	2011-02-20 19:32:59 +0000
+++ mass_import.py	2011-02-22 16:11:07 +0000
@@ -3,6 +3,7 @@
 import codecs
 import logging
 import os
+import Queue
 import random
 import signal
 from stat import ST_DEV, ST_INO
@@ -111,223 +112,116 @@
 d_archive = debian.main_archive
 u_archive = ubuntu.main_archive
 
-lp_lock = threading.Lock()
-
-
-def subprocess_setup():
-    signal.signal(signal.SIGPIPE, signal.SIG_DFL)
-
-
-def pool_base(name):
-    if name.startswith("lib"):
-        return name[:4]
-    return name[0]
-
-
-class ThreadedImporter(threading.Thread):
-
-    import_cmd = 'import_package.py %s'
-
-    def __init__(self, package, job_id):
-        super(ThreadedImporter, self).__init__()
-        self.package = package
+
+class Importer(icommon.SubprocessMonitor):
+
+    def __init__(self, cmd, package_name, job_id):
+        super(Importer, self).__init__(cmd % (package_name,))
+        self.package_name = package_name
         self.job_id = job_id
-        self.success = None
-        self.output = None
-        self.proc_pid = None
-        self.stopped = threading.Event()
+        self.status_db = icommon.StatusDatabase(icommon.sqlite_file)
 
-    def run(self):
-        success = False
-        output = icommon.running_sentinel
-        logger.info("Trying %s" % self.package)
-        proc = subprocess.Popen(self.import_cmd % self.package,
-            shell=True, stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT, stdin=subprocess.PIPE,
-            preexec_fn=subprocess_setup)
-        # Save the pid so the process can be killed from another thread.
-        self.proc_pid = proc.pid
-        output, _ = proc.communicate()
-        if proc.returncode == icommon.no_lock_returncode:
-            logger.info("Couldn't lock %s, skipping" % self.package)
+    def collect(self, timeout=None):
+        super(Importer, self).collect(timeout)
+        if self.retcode == icommon.no_lock_returncode:
+            logger.info("Couldn't lock %s, skipping" % self.package_name)
         else:
-            self.success = (proc.returncode == 0)
-            self.output = output
-        self.stopped.set()
-
-    def kill(self):
-        if self.proc_id is not None:
-            try:
-                os.kill(self.proc_pid, signal.SIGTERM)
-                self.proc_pid = None
-            except OSError, e:
-                if e.errno == errno.ESRCH:
-                    # The process doesn't exist anymore.
-                    self.proc_pid = None
-                    # We ignore all other exceptions but don't reset proc_pid
-                    # for them so that we can try to kill it again. We don't
-                    # re-raise either to let the caller continue.
-                else:
-                    pass
-            except:
-                pass
-
-
-class Stop(Exception):
-    pass
-
-
-class AllQueue(object):
-    """A Queue that always returns a package, even if we are not sure that
-       it is needed"""
+            unicode_output = self.out.decode("utf-8", "replace")
+            ascii_output = unicode_output.encode("ascii", "replace")
+            success = self.retcode == 0
+            if success:
+                logger.info("Success %s: %s"
+                            % (self.package_name,
+                               ascii_output.replace("\n", " ")))
+            else:
+                logger.warning("Importing %s failed:\n%s" % (self.package_name,
+                                                             ascii_output))
+            self.status_db.finish_job(
+                self.package_name, self.job_id, success,
+                unicode_output.encode("utf-8", "replace"))
+        logger.info("thread for %s finished" % self.package_name)
+
+
+class AllQueue(Queue.Queue):
+    """A Queue that always returns a package.
+
+    The jobs in the status db are tried first. 
+
+    When no more jobs are pending, the package db is used, even if we are not
+    sure that it is needed.
+    """
 
     def __init__(self):
+        Queue.Queue.__init__(self)
         self.tried = set()
-        self.packages_db = icommon.PackageDatabase(
-                icommon.sqlite_package_file)
+        self.packages_db = icommon.PackageDatabase(icommon.sqlite_package_file)
         self.status_db = icommon.StatusDatabase(icommon.sqlite_file)
 
+    def get_nowait(self):
+        # FIXME: It's bogus to implement get_nowait() this way (without
+        # locking) but we don't really care since we have a single consumer
+        # and no producer. This should be changed to a different model anyway
+        # so different kinds of jobs could be queued -- vila 20110221
+        try:
+            job_id, package_name = Queue.Queue.get_nowait(self)
+        except Queue.Empty:
+            self.unfinished_tasks += 1
+            job_id, package_name = self.next_job()
+        return Importer('import_package.py %s',
+                        package_name, job_id)
+
     def next_job(self):
         while True:
-            try:
-                to_try = self.packages_db.get_one(self.tried)
-            except StopIteration:
-                return (None, None)
-            if to_try is not None:
-                self.tried.add(to_try)
-                job_id = self.status_db.start_package(to_try)
-                if job_id is not None:
-                    return (job_id, to_try)
+            # jobs first
+            job_id, package = self.status_db.next_job()
+            if package is not None:
+                return job_id, package
             else:
-                self.tried = set()
-
-
-class ImportDriver(threading.Thread):
+                # Now the packages
+                try:
+                    to_try = self.packages_db.get_one(self.tried)
+                except StopIteration:
+                    raise Queue.Empty
+                if to_try is not None:
+                    self.tried.add(to_try)
+                    job_id = self.status_db.start_package(to_try)
+                    if job_id is not None:
+                        return (job_id, to_try)
+                else:
+                    logger.info("All packages requeued, start again")
+                    self.tried = set()
+
+
+class ImportDriver(icommon.ThreadDriver):
     """Monitor the ThreadedImporter.
 
     This includes planning and spawning imports, tracking their success or
     failures and shutting them down.
     """
 
-    MAX_THREADS = 6
-
-    def __init__(self):
-        super(ImportDriver, self).__init__()
-        self.threads = []
-        self.stop_requested = threading.Event()
-        self.stop_now = threading.Event()
-        self.started = threading.Event()
-        self.stopped = threading.Event()
-
-    def request_stop(self):
-        self.stop_requested.set()
-
-    def should_stop(self):
-        return self.stop_requested.isSet() or self.must_stop()
-
-    def must_stop(self):
-        return self.stop_now.isSet()
-
-    def get_next(self):
-        job_id, package = self.status_db.next_job()
-        if package is None:
-            (job_id, package) = self.queue.next_job()
-        return (job_id, package)
-
-    def sleep(self, max):
-        for i in range(max):
-            if self.should_stop():
-                return True
-            time.sleep(1)
-        return self.should_stop()
-
-    def deep_sleep(self, max):
-        for i in range(max):
-            if self.must_stop():
-                return True
-            time.sleep(1)
-        return self.must_stop()
-
-    def get_max_threads(self):
-        if os.path.exists(icommon.max_threads_file):
-            f = open(icommon.max_threads_file)
-            try:
-                return int(f.read().splitlines()[0].strip())
-            except Exception, e:
-                logger.warning("Error reading max threads file: %s", str(e))
-                return self.MAX_THREADS
-            finally:
-                f.close()
-        else:
-            return self.MAX_THREADS
-
-    def package_finished(self, package, job_id, success, output):
-        unicode_output = output.decode("utf-8", "replace")
-        ascii_output = unicode_output.encode("ascii", "replace")
-        if success:
-            logger.info("Success %s: %s"
-                        % (package, ascii_output.replace("\n", " ")))
-        else:
-            logger.warning("Importing %s failed:\n%s" % (package, ascii_output))
-        self.status_db.finish_job(package, job_id, success,
-                                  unicode_output.encode("utf-8", "replace"))
-
-    def _wait_until_threads_reaches(self, target):
-        i = 0
-        while len(self.threads) > target:
-            if i % 6 == 0:
-                logger.info("threads for %s still active"
-                        % str([t.package for t in self.threads]))
-            removed = self._retire_finished_threads()
-            if not removed:
-                self.sleep(10)
-            i += 1
-
-    def _retire_finished_threads(self):
-        removed = False
-        for thread in self.threads[:]:
-            if not thread.isAlive():
-                removed = True
-                logger.info("thread for %s finished" % thread.package)
-                if thread.success is not None:
-                    self.package_finished(thread.package, thread.job_id,
-                            thread.success, thread.output)
-                self.threads.remove(thread)
-            elif self.must_stop():
-                thread.kill()
-        return removed
-
-    def run(self):
-        try:
-            self.status_db = icommon.StatusDatabase(icommon.sqlite_file)
-            self.queue = AllQueue()
-            self.threads = []
-            # Release caller
-            self.started.set()
-            asked_to_stop = False
-            while not self.should_stop():
-                self._retire_finished_threads()
-                max_threads = self.get_max_threads()
-                job_id, next = self.get_next()
-                if next is None:
-                    logger.debug("No package in queue, sleeping")
-                    self.sleep(10)
-                    continue
-                if self.should_stop():
-                    continue
-                logger.debug("Starting thread for %s" % next)
-                new_thread = ThreadedImporter(next, job_id)
-                new_thread.start()
-                self.threads.append(new_thread)
-                self._wait_until_threads_reaches(max_threads-1)
-            # We've been asked to stop
-            logger.info("Driver asked to stop")
-            self._wait_until_threads_reaches(0)
-        except Exception, e:
-            logger.critical("Driver hit %s" % str(e))
-        finally:
-            logger.info("Driver stopping")
-            self.stopped.set()
+    def __init__(self, max_threads):
+        super(ImportDriver, self).__init__(None, max_threads)
+
+    def before_start(self):
+        self.queue = AllQueue()
+
+    def start_new_thread(self):
+        t = super(ImportDriver, self).start_new_thread()
+        if t is not None:
+            logger.debug("Starting thread for %s" % (t.package_name,))
+
+    def collect_terminated_threads(self):
+        before = len(self.threads)
+        super(ImportDriver, self).collect_terminated_threads()
+        after = len(self.threads)
+        if before != after:
+            # Only mention the running threads when we add or remove some
+            logger.info("threads for %r still active"
+                        % [t.package_name for t in self.threads])
+
+
+class Stop(Exception):
+    pass
 
 
 class ImportController(object):
@@ -335,38 +229,8 @@
 
     def __init__(self):
         self.status_db = icommon.StatusDatabase(icommon.sqlite_file)
-        self.stop_requested = False
-        self.driver = ImportDriver()
-        self.stopped = threading.Event()
-        self.stopped.set()
-
-    def request_stop(self):
-        self.stop_requested = True
-
-    def should_graceful_stop(self):
-        if os.path.exists(icommon.stop_file):
-            return True
-        return self.should_stop()
-
-    def should_stop(self):
-        return self.stop_requested
-
-    def check_stop(self):
-        if self.should_stop():
-            self.driver.stop_now.set()
-            raise Stop
-        if self.should_graceful_stop():
-            self.driver.stop_requested.set()
-            raise Stop
-
-    def sleep(self, max):
-        for i in range(max):
-            self.check_stop()
-            time.sleep(1)
-        self.check_stop()
-
-    def add_jobs_for_interrupted(self):
-        self.status_db.add_jobs_for_interrupted()
+        # Start with a 8-way Driver
+        self.driver = ImportDriver(8)
 
     def run(self):
         lock = icommon.lock_main()
@@ -374,25 +238,36 @@
             logger.info("Another main process is running")
             raise Stop
         try:
-            self.add_jobs_for_interrupted()
-            self.stopped.clear()
-            try:
-                self.driver.start()
-                self.driver.started.wait()
-                while True:
-                    if self.driver.stopped.isSet():
-                        self.stopped.set()
-                        break
-                    self.sleep(10)
-            finally:
-                self.driver.stop_requested.set()
-                logger.info("Waiting for driver to finish")
-                self.driver.stopped.wait()
-                self.stopped.set()
-                logger.info("Finished")
+            # First, re-queue previously interrupted jobs.
+            self.status_db.add_jobs_for_interrupted()
+            self.driver.start()
+            self.driver.started.wait()
+            while True:
+                # First, check the driver
+                self.driver.stopped.wait(0.1)
+                if self.driver.stopped.isSet():
+                    break
+                # Then check the graceful stop file
+                if os.path.exists(icommon.stop_file):
+                    self.driver.queue_closed.set()
+                    continue
+            # Catch driver exception if any
+            self.report_driver_exception()
+            self.driver.join()
+            logger.info("Finished")
         finally:
             lock.close()
 
+    def report_driver_exception(self):
+        if self.driver.exception is not None:
+            import traceback
+            exc_class, exc_value, exc_tb = self.driver.exception
+            logger.info('Driver failed with exception:\n%s'
+                        % ''.join(traceback.format_exception(
+                        exc_class, exc_value, exc_tb)))
+            # No need to ever re-raise this exception, we just reported it
+            self.driver.exception = None
+
 
 controller = ImportController()
 
@@ -401,9 +276,11 @@
     logger.info("Received signal")
     signal.signal(signum, signal.SIG_DFL)
 
-    controller.driver.stop_now.set()
+    controller.driver.stop.set()
     logger.info("Waiting for driver to finish")
     controller.driver.stopped.wait()
+    controller.report_driver_exception()
+    controller.driver.join()
     logger.info("Driver finished: stopping")
     sys.exit(1)
 

=== modified file 'tests.py'
--- tests.py	2011-02-22 16:11:07 +0000
+++ tests.py	2011-02-22 16:11:07 +0000
@@ -378,11 +378,11 @@
     pass
 
 
-class TestSubprocessThread(tests.TestCase):
+class TestSubprocessMonitor(tests.TestCase):
 
     def run_in_subprocess(self, cmd, kls=None):
         if kls is None:
-            kls = icommon.SubprocessThread
+            kls = icommon.SubprocessMonitor
         sub = kls(cmd)
         self.addCleanup(sub.join, 0)
         sub.start()
@@ -420,7 +420,7 @@
         sub = self.run_in_subprocess('yes')
         sub.started.wait()
         # For oscure reasons, using SIGABRT instead of SIGKILL and stopping
-        # below left the 'yes' process alive and the SubprocessThread happily
+        # below left the 'yes' process alive and the SubprocessMonitor happily
         # consuming memroy (8GB mark reached while sitting under pdb, which is
         # good enough to consider bug #589532 addressed.
         # import pdb; pdb.set_trace()
@@ -430,16 +430,16 @@
 
     def test_kill_catch_zombie(self):
         control = threading.Event()
-        class TestSubprocessThread(icommon.SubprocessThread):
+        class TestSubprocessMonitor(icommon.SubprocessMonitor):
 
             def switch_and_set(self, new):
-                super(TestSubprocessThread, self).switch_and_set(new)
+                super(TestSubprocessMonitor, self).switch_and_set(new)
                 if new is self.stopped:
                     # The process is running but we haven't called
                     # proc.communicate yet.
                     control.wait()
 
-        sub = self.run_in_subprocess('yes', TestSubprocessThread)
+        sub = self.run_in_subprocess('yes', TestSubprocessMonitor)
         self.addCleanup(sub.join, 0)
         sub.started.wait()
         # Kill the subprocess ourselves
@@ -454,7 +454,7 @@
         self.assertEquals(-signal.SIGTERM, sub.retcode)
 
 
-class Sleeper(icommon.SubprocessThread):
+class Sleeper(icommon.SubprocessMonitor):
 
     def __init__(self):
         # sleep can't be killed (or interrupted) so we don't sleep for long but
@@ -544,7 +544,7 @@
         self.started = threading.Event()
         self.resumed = threading.Event()
         self.terminated = threading.Event()
-        class TestThread(icommon.SubprocessThread):
+        class TestThread(icommon.SubprocessMonitor):
 
             def collect(thread):
                 self.assertIs(self.thread, thread)
@@ -584,7 +584,7 @@
         self.assertLength(0, driver.threads)
 
     def test_thread_exception_raised_in_driver(self):
-        class TestThread(icommon.SubprocessThread):
+        class TestThread(icommon.SubprocessMonitor):
 
             def collect(thread):
                 # false returns 1