~vila/udd/717204-stop-too-fast

#!/usr/bin/python

import codecs
import logging
import os
import random
import signal
from stat import ST_DEV, ST_INO
import subprocess
import sys
import threading
import time

from launchpadlib.errors import HTTPError

sys.path.insert(0, os.path.dirname(__file__))
import icommon

#import httplib2
#httplib2.debuglevel = 1


class WatchedFileHandler(logging.FileHandler):
    """
    A handler for logging to a file, which watches the file
    to see if it has changed while in use. This can happen because of
    usage of programs such as newsyslog and logrotate which perform
    log file rotation. This handler, intended for use under Unix,
    watches the file to see if it has changed since the last emit.
    (A file has changed if its device or inode have changed.)
    If it has changed, the old file stream is closed, and the file
    opened to get a new stream.

    This handler is not appropriate for use under Windows, because
    under Windows open files cannot be moved or renamed - logging
    opens the files with exclusive locks - and so there is no need
    for such a handler. Furthermore, ST_INO is not supported under
    Windows; stat always returns zero for this value.

    This handler is based on a suggestion and patch by Chad J.
    Schroeder.
    """
    def __init__(self, filename, mode='a', encoding=None):
        logging.FileHandler.__init__(self, filename, mode, encoding)
        if not os.path.exists(self.baseFilename):
            self.dev, self.ino = -1, -1
        else:
            stat = os.stat(self.baseFilename)
            self.dev, self.ino = stat[ST_DEV], stat[ST_INO]
        self.encoding = encoding

    def _open(self):
        """
        Open the current base file with the (original) mode and encoding.
        Return the resulting stream.
        """
        if self.encoding is None:
            stream = open(self.baseFilename, self.mode)
        else:
            stream = codecs.open(self.baseFilename, self.mode, self.encoding)
        return stream


    def emit(self, record):
        """
        Emit a record.

        First check if the underlying file has changed, and if it
        has, close the old stream and reopen the file to get the
        current stream.
        """
        if not os.path.exists(self.baseFilename):
            stat = None
            changed = 1
        else:
            stat = os.stat(self.baseFilename)
            changed = (stat[ST_DEV] != self.dev) or (stat[ST_INO] != self.ino)
        if changed and self.stream is not None:
            self.stream.close()
            self.stream = self._open()
            if stat is None:
                stat = os.stat(self.baseFilename)
            self.dev, self.ino = stat[ST_DEV], stat[ST_INO]
        logging.FileHandler.emit(self, record)


logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

debug_handler = WatchedFileHandler(icommon.debug_log_file)
progress_handler = WatchedFileHandler(icommon.progress_log_file)

debug_handler.setLevel(logging.DEBUG)
progress_handler.setLevel(logging.INFO)

formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")

debug_handler.setFormatter(formatter)
progress_handler.setFormatter(formatter)

logger.addHandler(debug_handler)
logger.addHandler(progress_handler)

logger.info("Starting up")


logger.debug("Getting Launchpad object")
lp = icommon.get_lp()
debian = lp.distributions['debian']
ubuntu = lp.distributions['ubuntu']
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):

    def __init__(self, package, job_id):
        super(ThreadedImporter, self).__init__()
        self.package = package
        self.job_id = job_id
        self.success = None
        self.output = None
        self.should_stop = threading.Event()

    def run(self):
        success = False
        output = icommon.running_sentinel
        logger.info("Trying %s" % self.package)
        proc = subprocess.Popen("import_package.py %s" % self.package,
            shell=True, stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT, stdin=subprocess.PIPE,
            preexec_fn=subprocess_setup)
        proc.stdin.close()
        while proc.poll() is None:
            if self.should_stop.isSet():
                os.kill(proc.pid, signal.SIGTERM)
            time.sleep(1)
        output = proc.stdout.read()
        if proc.returncode == icommon.no_lock_returncode:
            logger.info("Couldn't lock %s, skipping" % self.package)
            return
        self.success = (proc.returncode == 0)
        self.output = output



class Stop(Exception):
    pass


class AllQueue(object):
    """A Queue that always returns a package, even if we are not sure that
       it is needed"""

    def __init__(self):
        self.tried = set()
        self.packages_db = icommon.PackageDatabase(
                icommon.sqlite_package_file)
        self.status_db = icommon.StatusDatabase(icommon.sqlite_file)

    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)
            else:
                self.tried = set()


class ImportDriver(threading.Thread):

    MAX_THREADS = 6

    def __init__(self):
        super(ImportDriver, self).__init__()
        self.threads = []
        self.stop_requested = threading.Event()
        self.stop_now = threading.Event()
        self.stopped = threading.Event()
        # Avoid a race when starting, we clear this when we actually
        # start
        self.stopped.set()

    def request_stop(self):
        self.stop_requested.set()

    def force_stop(self):
        self.stop_now.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):
        if success:
            logger.info("Success %s: %s" % (package,
                        output.decode("utf-8", "replace").encode("ascii",
                            "replace").replace("\n", " ")))
        else:
            logger.warning("Importing %s failed:\n%s"
                    % (package, output.decode("utf-8", "replace").encode("ascii",
                            "replace")))
        self.status_db.finish_job(package, job_id, success,
                output.decode("utf-8", "replace").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.should_stop.set()
        return removed


    def run(self):
        self.stopped.clear()
        try:
            self.status_db = icommon.StatusDatabase(icommon.sqlite_file)
            self.queue = AllQueue()
            self.threads = []
            asked_to_stop = False
            while not asked_to_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")
                    asked_to_stop = self.sleep(10)
                    continue
                if self.should_stop():
                    asked_to_stop = True
                    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)
                if self.should_stop():
                    asked_to_stop = True
            # 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()


class ImportController(object):

    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()

    def run(self):
        lock = icommon.lock_main()
        if lock is None:
            logger.info("Another main process is running")
            raise Stop
        try:
            self.add_jobs_for_interrupted()
            self.stopped.clear()
            try:
                self.driver.start()
                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")
        finally:
            lock.close()


controller = ImportController()


def die(signum, frame):
    logger.info("Received signal")
    signal.signal(signum, signal.SIG_DFL)

    controller.driver.stop_now.set()
    logger.info("Waiting for driver to finish")
    controller.driver.stopped.wait()
    logger.info("Driver finished: stopping")
    sys.exit(1)


signal.signal(signal.SIGINT, die)
signal.signal(signal.SIGTERM, die)

try:
    controller.run()
except HTTPError, e:
    logger.critical(e.content)
    raise
except Exception, e:
    logger.critical(str(e))
    raise