[PATCH] Execute the yum transaction in another process

Brian C. Lane bcl at redhat.com
Tue Mar 26 19:24:26 UTC 2013


From: "Brian C. Lane" <bcl at redhat.com>

This moves the transaction into a multiprocessing thread and uses the
yum transaction save/load feature to transfer the data to it.
This should prevent yum thread problems and rpm chroot from interfering
with the main anaconda process.
---
 pyanaconda/packaging/yumpayload.py | 352 ++++++++++++++++++++++++++-----------
 1 file changed, 250 insertions(+), 102 deletions(-)

diff --git a/pyanaconda/packaging/yumpayload.py b/pyanaconda/packaging/yumpayload.py
index 4fc6177..3685184 100644
--- a/pyanaconda/packaging/yumpayload.py
+++ b/pyanaconda/packaging/yumpayload.py
@@ -42,6 +42,9 @@ import shutil
 import sys
 import time
 import tempfile
+from Queue import Empty
+from multiprocessing import Process, Queue
+from urlgrabber.grabber import URLGrabError
 
 from . import *
 
@@ -92,6 +95,14 @@ import itertools
 
 from pykickstart.constants import KS_MISSING_IGNORE
 
+STATUS_INFO = 0
+STATUS_DEBUG = 1
+STATUS_ERROR = 2
+STATUS_PROGRESS = 3
+STATUS_EXCEPTION = 4
+STATUS_QUIT = 5
+
+YUM_PLUGINS = ["blacklist", "whiteout", "fastestmirror", "langpacks"]
 default_repos = [productName.lower(), "rawhide"]
 
 import inspect
@@ -115,6 +126,123 @@ class YumLock(object):
 _yum_lock = YumLock()
 _yum_cache_dir = "/tmp/yum.cache"
 
+def run_yum_transaction(statusQ, release, yum_conf, install_root, ts_file, script_log, testing=False):
+    """ Execute a yum transaction loaded from a transaction file
+
+        :param statusQ: Used to communicate status back to calling process
+        :type statusQ: multiprocessing.Queue
+        :param release: The release version to use
+        :type release: string
+        :param yum_conf: Path to yum config file to use
+        :type yum_conf: string
+        :param install_root: Path to install root
+        :type install_root: string
+        :param ts_file: Path to yum transaction file to load and execute
+        :type ts_file: string
+        :param script_log: Path to file to store rpm script logs in
+        :type script_log: string
+        :param testing: True sets RPMTRANS_FLAG_TEST (default is false)
+        :type testing: bool
+        :returns: Nothing
+
+        This is used to run the yum transaction in a separate process, preventing
+        problems with threads and rpm chrooting during the install.
+    """
+    from yum.Errors import PackageSackError, RepoError, YumBaseError, YumRPMTransError
+
+    try:
+        # Setup the basics, point to the config file and install_root
+        _yum = yum.YumBase()
+
+        # Reset the pycurl object that may be left over from before the fork
+        from urlgrabber import grabber
+        grabber.reset_curl_obj()
+
+        _yum.use_txmbr_in_callback = True
+
+        # Set some configuration parameters that don't get set through a config
+        # file.  yum will know what to do with these.
+        # Enable all types of yum plugins. We're somewhat careful about what
+        # plugins we put in the environment.
+        _yum.preconf.plugin_types = yum.plugins.ALL_TYPES
+        _yum.preconf.enabled_plugins = YUM_PLUGINS
+        _yum.preconf.fn = yum_conf
+        _yum.preconf.root = install_root
+        # set this now to the best default we've got ; we'll update it if/when
+        # we get a base repo set up
+        _yum.preconf.releasever = release
+        # Set the yum verbosity to 6, and update yum's internal logger
+        # objects to the debug level.  This is a bit of a hack requiring
+        # internal knowledge of yum, that will hopefully go away in the
+        # future with API improvements.
+        _yum.preconf.debuglevel = 10
+        _yum.preconf.errorlevel = 10
+        _yum.logger.setLevel(logging.DEBUG)
+        _yum.verbose_logger.setLevel(logging.DEBUG)
+
+        # Setup yum cache dir outside the installroot
+        if _yum.conf.cachedir.startswith(_yum.conf.installroot):
+            root = _yum.conf.installroot
+            _yum.conf.cachedir = _yum.conf.cachedir[len(root):]
+
+        # Load the transaction file and execute it
+        _yum.load_ts(ts_file)
+        _yum.initActionTs()
+
+        if rpmUtils and rpmUtils.arch.isMultiLibArch():
+            _yum.ts.ts.setColor(3)
+
+        statusQ.put((STATUS_DEBUG, ["populate transaction set"]))
+        try:
+            # uses dsCallback.transactionPopulation
+            _yum.populateTs(keepold=0)
+        except RepoError as e:
+            statusQ.put((STATUS_ERROR, ["error populating transaction: %s" % e]))
+            statusQ.put((STATUS_QUIT, []))
+            return
+
+        statusQ.put((STATUS_DEBUG, ["check transaction set"]))
+        _yum.ts.check()
+        statusQ.put((STATUS_DEBUG, ["order transaction set"]))
+        _yum.ts.order()
+        _yum.ts.clean()
+
+        # Write scriptlet output to a file to be logged later
+        logfile = open(script_log, "w")
+        _yum.ts.ts.scriptFd = logfile.fileno()
+        rpm.setLogFile(logfile)
+
+        # create the install callback
+        rpmcb = RPMCallback(_yum, logfile, statusQ)
+
+        if testing:
+            _yum.ts.setFlags(rpm.RPMTRANS_FLAG_TEST)
+
+        # Reset the curl object again. If we don't do this it will use the last
+        # range request and only download part of the file.
+        grabber.reset_curl_obj()
+        statusQ.put((STATUS_INFO, ["running transaction"]))
+        try:
+            _yum.runTransaction(cb=rpmcb)
+        except PackageSackError as e:
+            statusQ.put((STATUS_ERROR, ["PackageSackError: %s" % e]))
+        except YumRPMTransError as e:
+            statusQ.put((STATUS_ERROR, ["YumRPMTransError: %s" % e]))
+        except YumBaseError as e:
+            statusQ.put((STATUS_ERROR, ["YumBaseError: %s" % e]))
+            for error in e.errors:
+                statusQ.put((STATUS_ERROR, ["    %s" % error]))
+        else:
+            statusQ.put((STATUS_INFO, ["transaction complete"]))
+        finally:
+            _yum.ts.close()
+            iutil.resetRpmDb()
+            logfile.close()
+    except Exception as e:
+        statusQ.put((STATUS_ERROR, ["transaction error: %s" % e]))
+    finally:
+        statusQ.put((STATUS_QUIT, []))
+
 class YumPayload(PackagePayload):
     """ A YumPayload installs packages onto the target system using yum.
 
@@ -217,8 +345,7 @@ class YumPayload(PackagePayload):
             # Enable all types of yum plugins. We're somewhat careful about what
             # plugins we put in the environment.
             self._yum.preconf.plugin_types = yum.plugins.ALL_TYPES
-            self._yum.preconf.enabled_plugins = ["blacklist", "whiteout", "fastestmirror",
-                                                 "langpacks"]
+            self._yum.preconf.enabled_plugins = YUM_PLUGINS
             self._yum.preconf.fn = "/tmp/anaconda-yum.conf"
             self._yum.preconf.root = root
             # set this now to the best default we've got ; we'll update it if/when
@@ -228,8 +355,8 @@ class YumPayload(PackagePayload):
             # objects to the debug level.  This is a bit of a hack requiring
             # internal knowledge of yum, that will hopefully go away in the
             # future with API improvements.
-            self._yum.preconf.debuglevel = 6
-            self._yum.preconf.errorlevel = 6
+            self._yum.preconf.debuglevel = 10
+            self._yum.preconf.errorlevel = 10
             self._yum.logger.setLevel(logging.DEBUG)
             self._yum.verbose_logger.setLevel(logging.DEBUG)
 
@@ -246,6 +373,7 @@ metadata_expire=never
 pluginpath=/usr/lib/yum-plugins,/tmp/updates/yum-plugins
 pluginconfpath=/etc/yum/pluginconf.d,/tmp/updates/pluginconf.d
 plugins=1
+debuglevel=10
 reposdir=%s
 """ % (_yum_cache_dir, self._repos_dir)
 
@@ -1460,82 +1588,93 @@ reposdir=%s
         return retval
 
     def install(self):
-        """ Install the payload. """
-        from yum.Errors import PackageSackError, RepoError, YumBaseError, YumRPMTransError
+        """ Install the payload.
+
+            This writes out the yum transaction and then uses a Process thread
+            to execute it in a totally separate process.
 
-        log.info("preparing transaction")
-        log.debug("initialize transaction set")
+            It monitors the status of the install and logs debug info, updates
+            the progress meter and cleans up when it is done.
+
+            TODO: Properly handle raising an exception on errors.
+        """
+        def _check_queue(statusQ):
+            """ Process all pending statusQ messages
+
+                :param statusQ: the status queue to receive messages from
+                :type statusQ: Queue
+                :returns: Exit needed, Errors received
+                :rtype: tuple
+            """
+            error = False
+            while True:
+                try:
+                    (code, args) = statusQ.get(True, 1)
+                except Empty:
+                    return False, error
+
+                if code == STATUS_INFO:
+                    for arg in args:
+                        log.info(arg)
+                elif code == STATUS_DEBUG:
+                    for arg in args:
+                        log.debug(arg)
+                elif code == STATUS_ERROR:
+                    for arg in args:
+                        log.error(arg)
+                    error = True
+                elif code == STATUS_PROGRESS:
+                    for arg in args:
+                        log.info(arg)
+                        progress.send_message(arg)
+                elif code == STATUS_QUIT:
+                    return True, error
+
+        ts_file = ROOT_PATH+"/anaconda-yum.yumtx"
         with _yum_lock:
-            self._yum.initActionTs()
+            # Save the transaction, this will be loaded and executed by the new
+            # process.
+            self._yum.save_ts(ts_file)
 
-            if rpmUtils and rpmUtils.arch.isMultiLibArch():
-                self._yum.ts.ts.setColor(3)
+            # Try and clean up yum before the fork
+            self.release()
+            self.deleteYumTS()
+            self._yum.close()
+            del self._yum
+
+        statusQ = Queue()
+        script_log = "/tmp/rpm-script.log"
+        release = self._getReleaseVersion(None)
+        p = Process(target=run_yum_transaction,
+                    args=(statusQ, release, "/tmp/anaconda-yum.conf",
+                          ROOT_PATH, ts_file, script_log))
+        p.start()
+        success = True
+        while p.is_alive():
+            exit, error = _check_queue(statusQ)
+            if error:
+                success = False
+            if exit:
+                break
 
-            log.debug("populate transaction set")
-            try:
-                # uses dsCallback.transactionPopulation
-                self._yum.populateTs(keepold=0)
-            except RepoError as e:
-                log.error("error populating transaction: %s" % e)
-                exn = PayloadInstallError(str(e))
-                if errorHandler.cb(exn) == ERROR_RAISE:
-                    raise exn
-
-            log.debug("check transaction set")
-            self._yum.ts.check()
-            log.debug("order transaction set")
-            self._yum.ts.order()
-            self._yum.ts.clean()
-
-            # Write scriptlet output to a file to be logged later
-            script_log = tempfile.NamedTemporaryFile(delete=False)
-            self._yum.ts.ts.scriptFd = script_log.fileno()
-            rpm.setLogFile(script_log)
-
-            # create the install callback
-            rpmcb = RPMCallback(self._yum, script_log,
-                                upgrade=self.data.upgrade.upgrade)
-
-            if flags.testing:
-                self._yum.ts.setFlags(rpm.RPMTRANS_FLAG_TEST)
-
-            log.info("running transaction")
-            progress.send_step()
-            try:
-                self._yum.runTransaction(cb=rpmcb)
-            except PackageSackError as e:
-                log.error("error [1] running transaction: %s" % e)
-                exn = PayloadInstallError(str(e))
-                if errorHandler.cb(exn) == ERROR_RAISE:
-                    raise exn
-            except YumRPMTransError as e:
-                log.error("error [2] running transaction: %s" % e)
-                exn = PayloadInstallError(self._transactionErrors(e.errors))
-                if errorHandler.cb(exn) == ERROR_RAISE:
-                    progress.send_quit(1)
-                    sys.exit(1)
-            except YumBaseError as e:
-                log.error("error [3] running transaction: %s" % e)
-                for error in e.errors:
-                    log.error("%s" % error[0])
-                exn = PayloadInstallError(str(e))
-                if errorHandler.cb(exn) == ERROR_RAISE:
-                    raise exn
-            else:
-                log.info("transaction complete")
-                progress.send_step()
-            finally:
-                self._yum.ts.close()
-                iutil.resetRpmDb()
-                script_log.close()
-
-                # log the contents of the scriptlet logfile
-                log.info("==== start rpm scriptlet logs ====")
-                with open(script_log.name) as f:
-                    for l in f:
-                        log.info(l)
-                log.info("==== end rpm scriptlet logs ====")
-                os.unlink(script_log.name)
+        log.debug("Waiting for pid %s to exit." % p.pid)
+        p.join()
+        log.debug("%s exited with %s" % (p.pid, p.exitcode))
+
+        # Need to check the queue one last time?
+        _check_queue(statusQ)
+        if os.path.exists(ts_file):
+            os.unlink(ts_file)
+
+        # log the contents of the scriptlet logfile
+        log.info("==== start rpm scriptlet logs ====")
+        with open(script_log) as f:
+            for l in f:
+                log.info(l)
+        log.info("==== end rpm scriptlet logs ====")
+        os.unlink(script_log)
+
+        return success
 
     def writeMultiLibConfig(self):
         if not self.data.packages.multiLib:
@@ -1595,10 +1734,17 @@ reposdir=%s
         super(YumPayload, self).postInstall()
 
 class RPMCallback(object):
-    def __init__(self, yb, log, upgrade=False):
+    def __init__(self, yb, log, statusQ):
+        """ :param yb: YumBase object
+            :type yb: YumBase
+            :param log: script logfile
+            :type log: string
+            :param statusQ: status communication back to other process
+            :type statusQ: Queue
+        """
         self._yum = yb              # yum.YumBase
         self.install_log = log      # logfile for yum script logs
-        self.upgrade = upgrade      # boolean
+        self.statusQ = statusQ
 
         self.package_file = None    # file instance (package file management)
 
@@ -1621,9 +1767,11 @@ class RPMCallback(object):
 
     def callback(self, event, amount, total, key, userdata):
         """ Yum install callback. """
+        self.statusQ.put((STATUS_DEBUG, ["callback: %s %s %s %s %s" % (event, amount, total, key, userdata)]))
+
         if event == rpm.RPMCALLBACK_TRANS_START:
             if amount == 6:
-                progress.send_message(_("Preparing transaction from installation source"))
+                self.statusQ.put((STATUS_PROGRESS, [_("Preparing transaction from installation source")]))
             self.total_actions = total
             self.completed_actions = 0
         elif event == rpm.RPMCALLBACK_TRANS_PROGRESS:
@@ -1636,17 +1784,14 @@ class RPMCallback(object):
             # update status that we're installing/upgrading %h
             # return an open fd to the file
             txmbr = self._get_txmbr(key)[1]
+            self.statusQ.put((STATUS_DEBUG, ["txmbr = %s" % txmbr]))
 
             # If self.completed_actions is still None, that means this package
             # is being opened to retrieve a %pretrans script. Don't log that
             # we're installing the package unless we've been called with a
             # TRANS_START event.
             if self.completed_actions is not None:
-                if self.upgrade:
-                    mode = _("Upgrading")
-                else:
-                    mode = _("Installing")
-
+                mode = _("Installing")
                 self.completed_actions += 1
                 msg_format = "%s %s (%d/%d)"
                 progress_package = txmbr.name
@@ -1659,14 +1804,19 @@ class RPMCallback(object):
                 log_msg = msg_format % (mode, txmbr.po,
                                         self.completed_actions,
                                         self.total_actions)
-                log.info(log_msg)
                 self.install_log.write(log_msg+"\n")
                 self.install_log.flush()
 
-                progress.send_message(progress_msg)
+                self.statusQ.put((STATUS_PROGRESS, [progress_msg]))
 
             self.package_file = None
-            repo = self._yum.repos.getRepo(txmbr.po.repoid)
+            try:
+                repo = self._yum.repos.getRepo(txmbr.po.repoid)
+            except Exception as e:
+                self.statusQ.put((STATUS_DEBUG, ["getRepo failed: %s" % e]))
+                raise Exception("rpmcallback getRepo failed")
+            else:
+                self.statusQ.put((STATUS_DEBUG, ["repo = %s" % repo]))
 
             while self.package_file is None:
                 try:
@@ -1681,24 +1831,23 @@ class RPMCallback(object):
                     #     obj.filename = '/tmp/stuff'
                     #     obj.url = 'http://foo.com/stuff'
                     checkfunc = (self._yum.verifyPkg, (txmbr.po, 1), {})
+                    self.statusQ.put((STATUS_DEBUG, ["getPackage %s" % txmbr.name]))
                     package_path = repo.getPackage(txmbr.po, checkfunc=checkfunc)
                 except URLGrabError as e:
-                    log.error("URLGrabError: %s" % (e,))
-                    exn = PayloadInstallError("failed to get package")
-                    if errorHandler.cb(exn, package=txmbr.po) == ERROR_RAISE:
-                        raise exn
+                    self.statusQ.put((STATUS_ERROR, ["URLGrabError: %s" % (e,)]))
+                    raise Exception("rpmcallback failed")
                 except (yum.Errors.NoMoreMirrorsRepoError, IOError) as e:
+                    self.statusQ.put((STATUS_ERROR, ["getPackage error: %s" % (e,)]))
                     if os.path.exists(txmbr.po.localPkg()):
                         os.unlink(txmbr.po.localPkg())
-                        log.debug("retrying download of %s" % txmbr.po)
+                        self.statusQ.put((STATUS_DEBUG, ["retrying download of %s" % txmbr.po]))
                         continue
-                    log.error("getPackage error: %s" % (e,))
-                    exn = PayloadInstallError("failed to open package")
-                    if errorHandler.cb(exn, package=txmbr.po) == ERROR_RAISE:
-                        raise exn
-                except yum.Errors.RepoError:
+                    raise Exception("rpmcallback failed")
+                except yum.Errors.RepoError as e:
+                    self.statusQ.put((STATUS_DEBUG, ["RepoError: %s" % e]))
                     continue
 
+                self.statusQ.put((STATUS_DEBUG, ["opening package %s" % package_path]))
                 self.package_file = open(package_path)
 
             return self.package_file.fileno()
@@ -1713,13 +1862,13 @@ class RPMCallback(object):
                 try:
                     os.unlink(package_path)
                 except OSError as e:
-                    log.debug("unable to remove file %s" % e.strerror)
+                    self.statusQ.put((STATUS_DEBUG, ["unable to remove file %s" % e.strerror]))
 
             # rpm doesn't tell us when it's started post-trans stuff which can
             # take a very long time.  So when it closes the last package, just
             # display the message.
             if self.completed_actions == self.total_actions:
-                progress.send_message(_("Performing post-installation setup tasks"))
+                self.statusQ.put((STATUS_PROGRESS, [_("Performing post-installation setup tasks")]))
         elif event == rpm.RPMCALLBACK_UNINST_START:
             # update status that we're cleaning up %key
             #progress.set_text(_("Cleaning up %s" % key))
@@ -1733,6 +1882,5 @@ class RPMCallback(object):
             # we should only error out for fatal script errors or the cpio and
             # unpack problems.
             if event != rpm.RPMCALLBACK_SCRIPT_ERROR or total:
-                exn = PayloadInstallError("cpio, unpack, or fatal script error")
-                if errorHandler.cb(exn, package=name) == ERROR_RAISE:
-                    raise exn
+                self.statusQ.put((STATUS_ERROR, ["%s: cpio, unpack, or fatal script error" % name]))
+                raise Exception("rpmcallback failed")
-- 
1.8.1.4



More information about the anaconda-patches mailing list