From 40e0f6617c5a9213e9399bf48ab89d5b98c29665 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?G=C3=B6k=C3=A7en=20Eraslan?= Date: Fri, 13 May 2011 11:50:02 +0000 Subject: [PATCH] index: Parallelize indexing procedure using Python's multiprocessing module. Create a pool of worker processes to make indexing in a parallel fashion. Normally, all metadata parsing and SHA1 calculation process are done sequentially per package. This commits distributes all package-based parsing, SHA1 calculation process to worker processes to make it faster in multiprocessor/multicore systems. --- pisi/index.py | 135 ++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 97 insertions(+), 38 deletions(-) diff --git a/pisi/index.py b/pisi/index.py index bda67992..e151cdb6 100644 --- a/pisi/index.py +++ b/pisi/index.py @@ -14,6 +14,7 @@ import os import shutil +import multiprocessing import gettext __trans = gettext.translation('pisi', fallback=True) @@ -30,6 +31,8 @@ import pisi.file import pisi.pxml.autoxml as autoxml import pisi.component as component import pisi.group as group +import pisi.operations.build + class Error(pisi.Error): pass @@ -85,7 +88,9 @@ class Index(xmlfile.XmlFile): self.repo_dir = repo_uri packages = [] + specs = [] deltas = {} + for root, dirs, files in os.walk(repo_uri): for fn in files: @@ -96,29 +101,66 @@ class Index(xmlfile.XmlFile): packages.append(os.path.join(root, fn)) if fn == 'components.xml': - self.add_components(os.path.join(root, fn)) + self.components.extend(add_components(os.path.join(root, fn))) if fn == 'pspec.xml' and not skip_sources: - self.add_spec(os.path.join(root, fn), repo_uri) + specs.append((os.path.join(root, fn), repo_uri)) if fn == 'distribution.xml': - self.add_distro(os.path.join(root, fn)) + self.distribution = add_distro(os.path.join(root, fn)) if fn == 'groups.xml': - self.add_groups(os.path.join(root, fn)) + self.groups.extend(add_groups(os.path.join(root, fn))) + + # Create a process pool, as many processes as the number of CPUs we + # have + pool = multiprocessing.Pool() + + try: + # Add source packages to index using a process pool + self.specs = pool.map(add_spec, specs) + except: + # If an exception occurs (like a keyboard interrupt), immediately + # terminate worker processes and propagate exception. (CLI honors + # KeyboardInterrupt exception, if you're not using CLI, you must + # handle KeyboardException yourself) + + pool.terminate() + raise try: obsoletes_list = map(str, self.distribution.obsoletes) except AttributeError: obsoletes_list = [] + latest_packages = [] + for pkg in util.filter_latest_packages(packages): pkg_name = util.parse_package_name(os.path.basename(pkg))[0] if pkg_name.endswith(ctx.const.debug_name_suffix): pkg_name = util.remove_suffix(ctx.const.debug_name_suffix, pkg_name) if pkg_name not in obsoletes_list: - ctx.ui.info(_('Adding %s to package index') % pkg) - self.add_package(pkg, deltas, repo_uri) + # Currently, multiprocessing.Pool.map method accepts methods + # with single parameters only. So we have to send our + # parameters as a tuple to workaround that + + latest_packages.append((pkg, deltas, repo_uri)) + + try: + # Add binary packages to index using a process pool + self.packages = pool.map(add_package, latest_packages) + except: + pool.terminate() + raise + else: + # Clean up output + ctx.ui.info("\r%-80.80s" % (_('Done.'))) + +def add_package(params): + try: + path, deltas, repo_uri = params + + ctx.ui.info("\r%-80.80s" % (_('Adding package to index: %s') % + os.path.basename(path)), noln = True) - def add_package(self, path, deltas, repo_uri): package = pisi.package.Package(path, 'r') md = package.get_metadata() md.package.packageSize = long(os.path.getsize(path)) @@ -160,43 +202,56 @@ class Index(xmlfile.XmlFile): md.package.deltaPackages.append(delta) - self.packages.append(md.package) + return md.package - def add_groups(self, path): - ctx.ui.info("Adding groups.xml to index...") - groups_xml = group.Groups() - groups_xml.read(path) - for grp in groups_xml.groups: - self.groups.append(grp) + except KeyboardInterrupt: + # Handle KeyboardInterrupt exception to prevent ugly backtrace of all + # worker processes and propagate the exception to main process. + # + # Probably it's better to use just 'raise' here, but multiprocessing + # module has some bugs about that: (python#8296, python#9205 and + # python#9207 ) + # + # For now, worker processes do not propagate exceptions other than + # Exception (like KeyboardInterrupt), so we have to manually propagate + # KeyboardInterrupt exception as an Exception. - def add_components(self, path): - ctx.ui.info("Adding components.xml to index...") - components_xml = component.Components() - components_xml.read(path) - #try: - for comp in components_xml.components: - self.components.append(comp) - #except: - # raise Error(_('Component in %s is corrupt') % path) - #ctx.ui.error(str(Error(*errs))) + raise Exception - def add_distro(self, path): - ctx.ui.info("Adding distribution.xml to index...") - distro = component.Distribution() - #try: - distro.read(path) - self.distribution = distro - #except: - # raise Error(_('Distribution in %s is corrupt') % path) - #ctx.ui.error(str(Error(*errs))) +def add_groups(path): + ctx.ui.info("Adding groups.xml to index...") + groups_xml = group.Groups() + groups_xml.read(path) + return groups_xml.groups - def add_spec(self, path, repo_uri): - import pisi.operations.build +def add_components(path): + ctx.ui.info("Adding components.xml to index...") + components_xml = component.Components() + components_xml.read(path) + #try: + return components_xml.components + #except: + # raise Error(_('Component in %s is corrupt') % path) + #ctx.ui.error(str(Error(*errs))) + +def add_distro(path): + ctx.ui.info("Adding distribution.xml to index...") + distro = component.Distribution() + #try: + distro.read(path) + return distro + #except: + # raise Error(_('Distribution in %s is corrupt') % path) + #ctx.ui.error(str(Error(*errs))) + +def add_spec(params): + try: + path , repo_uri = params ctx.ui.info(_('Adding %s to source index') % path) #TODO: may use try/except to handle this builder = pisi.operations.build.Builder(path) - #ctx.ui.error(_('SpecFile in %s is corrupt, skipping...') % path) - #ctx.ui.error(str(Error(*errs))) + #ctx.ui.error(_('SpecFile in %s is corrupt, skipping...') % path) + #ctx.ui.error(str(Error(*errs))) builder.fetch_component() sf = builder.spec if ctx.config.options and ctx.config.options.absolute_urls: @@ -204,4 +259,8 @@ class Index(xmlfile.XmlFile): else: # create relative path by default sf.source.sourceURI = util.removepathprefix(repo_uri, path) # check component - self.specs.append(sf) + return sf + + except KeyboardInterrupt: + # Multiprocessing hack, see add_package method for explanation + raise Exception