[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