Merge packages asynchronously in Portage.

pym/portage/dbapi/vartree.py

 # Copyright 1998-2011 Gentoo Foundation
 # Distributed under the terms of the GNU General Public License v2
 
 __all__ = [
         "vardbapi", "vartree", "dblink"] + \
         ["write_contents", "tar_contents"]
 
 import portage
 portage.proxy.lazyimport.lazyimport(globals(),
         'portage.checksum:_perform_md5_merge@perform_md5',
         'portage.data:portage_gid,portage_uid,secpass',
         'portage.dbapi.dep_expand:dep_expand',
         'portage.dbapi._MergeProcess:MergeProcess',
         'portage.dep:dep_getkey,isjustname,match_from_list,' + \
                 'use_reduce,_slot_re',
-        'portage.elog:elog_process,_preload_elog_modules',
+        'portage.elog:collect_ebuild_messages,collect_messages,' + \
+                'elog_process,_merge_logentries,_preload_elog_modules',
         'portage.locks:lockdir,unlockdir',
         'portage.output:bold,colorize',
         'portage.package.ebuild.doebuild:doebuild_environment,' + \
                 '_spawn_phase',
         'portage.package.ebuild.prepare_build_dirs:prepare_build_dirs',
         'portage.update:fixdbentries',
         'portage.util:apply_secpass_permissions,ConfigProtect,ensure_dirs,' + \
                 'writemsg,writemsg_level,write_atomic,atomic_ofstream,writedict,' + \
                 'grabdict,normalize_path,new_protect_filename',
         'portage.util.digraph:digraph',
@@ skipping 1166 matching lines @@
                 r'(?P<sym>(sym) (.+) -> (.+) ((\d+)|(?P<oldsym>(' + \
                 r'\(\d+, \d+L, \d+L, \d+, \d+, \d+, \d+L, \d+, (\d+), \d+\)))))' + \
                 r')$'
         )
 
         # When looping over files for merge/unmerge, temporarily yield to the
         # scheduler each time this many files are processed.
         _file_merge_yield_interval = 20
 
         def __init__(self, cat, pkg, myroot=None, settings=None, treetype=None,
-                vartree=None, blockers=None, scheduler=None):
+                vartree=None, blockers=None, scheduler=None, pipe=None):
                 """
                 Creates a DBlink object for a given CPV.
                 The given CPV may not be present in the database already.
                 
                 @param cat: Category
                 @type cat: String
                 @param pkg: Package (PV)
                 @type pkg: String
                 @param myroot: ignored, settings['ROOT'] is used instead
                 @type myroot: String (Path)
@@ skipping 38 matching lines @@
 
                 self.myroot=myroot
                 self._installed_instance = None
                 self.contentscache = None
                 self._contents_inodes = None
                 self._contents_basenames = None
                 self._linkmap_broken = False
                 self._md5_merge_map = {}
                 self._hash_key = (self.myroot, self.mycpv)
                 self._protect_obj = None
+                self._pipe = pipe
 
         def __hash__(self):
                 return hash(self._hash_key)
 
         def __eq__(self, other):
                 return isinstance(other, dblink) and \
                         self._hash_key == other._hash_key
 
         def _get_protect_obj(self):
 
@@ skipping 224 matching lines @@
                 if others_in_slot is None:
                         slot = self.vartree.dbapi.aux_get(self.mycpv, ["SLOT"])[0]
                         slot_matches = self.vartree.dbapi.match(
                                 "%s:%s" % (portage.cpv_getkey(self.mycpv), slot))
                         others_in_slot = []
                         for cur_cpv in slot_matches:
                                 if cur_cpv == self.mycpv:
                                         continue
                                 others_in_slot.append(dblink(self.cat, catsplit(cur_cpv)[1],
                                         settings=self.settings, vartree=self.vartree,
-                                        treetype="vartree"))
+                                        treetype="vartree", pipe=self._pipe))
 
                         retval = self._security_check([self] + others_in_slot)
                         if retval:
                                 return retval
 
                 contents = self.getcontents()
                 # Now, don't assume that the name of the ebuild is the same as the
                 # name of the dir; the package may have been moved.
                 myebuildpath = None
                 failures = 0
@@ skipping 143 matching lines @@
                                                         "sourced and the eclasses "
                                                         "from the current portage tree will be used "
                                                         "when necessary. Removal of "
                                                         "the ebuild file will cause the "
                                                         "pkg_prerm() and pkg_postrm() removal "
                                                         "phases to be skipped entirely.")
                                                         msg_lines.extend(wrap(msg, 72))
 
                                                         self._eerror(ebuild_phase, msg_lines)
 
-                                                # process logs created during pre/postrm
-                                                elog_process(self.mycpv, self.settings,
-                                                        phasefilter=('prerm', 'postrm'))
+                                        self._elog_process()
 
                                         if retval == os.EX_OK:
                                                 # myebuildpath might be None, so ensure
                                                 # it has a sane value for the clean phase,
                                                 # even though it won't really be sourced.
                                                 myebuildpath = os.path.join(self.dbdir,
                                                         self.pkg + ".ebuild")
                                                 doebuild_environment(myebuildpath, "cleanrm",
                                                         settings=self.settings, db=self.vartree.dbapi)
                                                 if scheduler is None:
@@ skipping 76 matching lines @@
                 if others_in_slot is None:
                         others_in_slot = []
                         slot = self.vartree.dbapi.aux_get(self.mycpv, ["SLOT"])[0]
                         slot_matches = self.vartree.dbapi.match(
                                 "%s:%s" % (portage.cpv_getkey(self.mycpv), slot))
                         for cur_cpv in slot_matches:
                                 if cur_cpv == self.mycpv:
                                         continue
                                 others_in_slot.append(dblink(self.cat, catsplit(cur_cpv)[1],
                                         settings=self.settings,
-                                        vartree=self.vartree, treetype="vartree"))
+                                        vartree=self.vartree, treetype="vartree", pipe=self._pipe))
 
                 dest_root = self._eroot
                 dest_root_len = len(dest_root) - 1
 
                 conf_mem_file = os.path.join(dest_root, CONFIG_MEMORY_FILE)
                 cfgfiledict = grabdict(conf_mem_file)
                 stale_confmem = []
 
                 unmerge_orphans = "unmerge-orphans" in self.settings.features
                 calc_prelink = "prelink-checksums" in self.settings.features
@@ skipping 999 matching lines @@
 
         def _eerror(self, phase, lines):
                 from portage.elog.messages import eerror as _eerror
                 if self._scheduler is None:
                         for l in lines:
                                 _eerror(l, phase=phase, key=self.settings.mycpv)
                 else:
                         self._scheduler.dblinkElog(self,
                                 phase, _eerror, lines)
 
-        def _elog_subprocess(self, funcname, phase, lines):
-                """
-                Subprocesses call this in order to create elog messages in
-                $T, for collection by the main process.
-                """
-                cmd = "source %s/isolated-functions.sh ; " % \
-                        portage._shell_quote(self.settings["PORTAGE_BIN_PATH"])
-                for line in lines:
-                        cmd += "%s %s ; " % (funcname, portage._shell_quote(line))
-                env = self.settings.environ()
-                env['EBUILD_PHASE'] = phase
-                subprocess.call([portage.const.BASH_BINARY, "-c", cmd],
-                        env=env)
+        def _elog_process(self):
+                cpv = self.mycpv
+                if self._pipe is None:
+                        elog_process(cpv, self.settings)
+                else:
+                        logdir = os.path.join(self.settings["T"], "logging")
+                        ebuild_logentries = collect_ebuild_messages(logdir)
+                        py_logentries = collect_messages(key=cpv).get(cpv, {})
+                        logentries = _merge_logentries(py_logentries, ebuild_logentries)
+                        funcnames = {
+                                "INFO": "einfo",
+                                "LOG": "elog",
+                                "WARN": "ewarn",
+                                "QA": "eqawarn",
+                                "ERROR": "eerror"
+                        }
+                        buffer = []
+                        for phase, messages in logentries.items():
+                                for key, lines in messages:
+                                        funcname = funcnames[key]
+                                        if isinstance(lines, basestring):
+                                                lines = [lines]
+                                        for line in lines:
+                                                fields = (funcname, phase, cpv, line.rstrip('\n'))
+                                                buffer.append(' '.join(fields))
+                                                buffer.append('\n')
+                        if buffer:
+                                os.write(self._pipe, ''.join(buffer))
 
         def treewalk(self, srcroot, destroot, inforoot, myebuild, cleanup=0,
                 mydbapi=None, prev_mtimes=None):
                 """
                 
                 This function does the following:
                 
                 calls self._preserve_libs if FEATURES=preserve-libs
                 calls self._collision_protect if FEATURES=collision-protect
                 calls doebuild(mydo=pkg_preinst)
                 Merges the package to the livefs
                 unmerges old version (if required)
                 calls doebuild(mydo=pkg_postinst)
                 calls env_update
-                calls elog_process
                 
                 @param srcroot: Typically this is ${D}
                 @type srcroot: String (Path)
                 @param destroot: ignored, self.settings['ROOT'] is used instead
                 @type destroot: String (Path)
                 @param inforoot: root of the vardb entry ?
                 @type inforoot: String (Path)
                 @param myebuild: path to the ebuild that we are processing
                 @type myebuild: String (Path)
                 @param mydbapi: dbapi which is handed to doebuild.
@@ skipping 89 matching lines @@
                         slot_matches.append(self.mycpv)
 
                 others_in_slot = []
                 from portage import config
                 for cur_cpv in slot_matches:
                         # Clone the config in case one of these has to be unmerged since
                         # we need it to have private ${T} etc... for things like elog.
                         others_in_slot.append(dblink(self.cat, catsplit(cur_cpv)[1],
                                 settings=config(clone=self.settings),
                                 vartree=self.vartree, treetype="vartree",
-                                scheduler=self._scheduler))
+                                scheduler=self._scheduler, pipe=self._pipe))
 
                 retval = self._security_check(others_in_slot)
                 if retval:
                         return retval
 
                 self.settings["REPLACING_VERSIONS"] = " ".join( 
                         [portage.versions.cpv_getversion(other.mycpv) for other in others_in_slot] )
                 self.settings.backup_changes("REPLACING_VERSIONS")
 
                 if slot_matches:
@@ skipping 127 matching lines @@
                                         "/etc/make.conf if you do not want to "
                                         "abort in cases like this.") % other_dblink.mycpv,
                                         wrap_width))
                                 eerror(msg)
                         if installed_files:
                                 return 1
 
                 # check for package collisions
                 blockers = None
                 if self._blockers is not None:
-                        # This is only supposed to be called when
-                        # the vdb is locked, like it is here.
                         blockers = self._blockers()
                 if blockers is None:
                         blockers = []
                 collisions, plib_collisions = \
                         self._collision_protect(srcroot, destroot,
                         others_in_slot + blockers, myfilelist + mylinklist)
 
                 # Make sure the ebuild environment is initialized and that ${T}/elog
                 # exists for logging of collision-protect eerror messages.
                 if myebuild is None:
@@ skipping 151 matching lines @@
 
                 # Always behave like --noconfmem is enabled for downgrades
                 # so that people who don't know about this option are less
                 # likely to get confused when doing upgrade/downgrade cycles.
                 pv_split = catpkgsplit(self.mycpv)[1:]
                 for other in others_in_slot:
                         if pkgcmp(pv_split, catpkgsplit(other.mycpv)[1:]) < 0:
                                 cfgfiledict["IGNORE"] = 1
                                 break
 
-                merge_task = MergeProcess(
-                        background=(self.settings.get('PORTAGE_BACKGROUND') == '1'),
-                        cfgfiledict=cfgfiledict, conf_mem_file=conf_mem_file, dblink=self,
-                        destroot=destroot,
-                        logfile=self.settings.get('PORTAGE_LOG_FILE'),
-                        scheduler=(scheduler or PollScheduler().sched_iface),
-                        srcroot=srcroot)
-
-                merge_task.start()
-                rval = merge_task.wait()
+                rval = self._merge_contents(srcroot, destroot, cfgfiledict,
+                        conf_mem_file)
                 if rval != os.EX_OK:
                         return rval
 
                 # These caches are populated during collision-protect and the data
                 # they contain is now invalid. It's very important to invalidate
                 # the contents_inodes cache so that FEATURES=unmerge-orphans
                 # doesn't unmerge anything that belongs to this package that has
                 # just been merged.
                 for dblnk in others_in_slot:
                         dblnk._clear_contents_cache()
@@ skipping 180 matching lines @@
                 while True:
                         x += 1
                         backup_p = p + '.backup.' + str(x).rjust(4, '0')
                         try:
                                 os.lstat(backup_p)
                         except OSError:
                                 break
 
                 return backup_p
 
-        def _merge_process(self, srcroot, destroot, cfgfiledict, conf_mem_file):
+        def _merge_contents(self, srcroot, destroot, cfgfiledict, conf_mem_file):
 
                 cfgfiledict_orig = cfgfiledict.copy()
 
                 # open CONTENTS file (possibly overwriting old one) for recording
                 outfile = codecs.open(_unicode_encode(
                         os.path.join(self.dbtmpdir, 'CONTENTS'),
                         encoding=_encodings['fs'], errors='strict'),
                         mode='w', encoding=_encodings['repo.content'],
                         errors='backslashreplace')
 
@@ skipping 208 matching lines @@
                                         else:
                                                 # a non-directory and non-symlink-to-directory.  Won't work for us.  Move out of the way.
                                                 backup_dest = self._new_backup_path(mydest)
                                                 msg = []
                                                 msg.append("")
                                                 msg.append(_("Installation of a directory is blocked by a file:"))
                                                 msg.append("  '%s'" % mydest)
                                                 msg.append(_("This file will be renamed to a different name:"))
                                                 msg.append("  '%s'" % backup_dest)
                                                 msg.append("")
-                                                self._elog_subprocess("eerror", "preinst", msg)
+                                                self._eerror("preinst", msg)
                                                 if movefile(mydest, backup_dest,
                                                         mysettings=self.settings,
                                                         encoding=_encodings['merge']) is None:
                                                         return 1
                                                 showMessage(_("bak %s %s.backup\n") % (mydest, mydest),
                                                         level=logging.ERROR, noiselevel=-1)
                                                 #now create our directory
                                                 try:
                                                         if self.settings.selinux_enabled():
                                                                 _selinux_merge.mkdir(mydest, mysrc)
@@ skipping 57 matching lines @@
                                         if stat.S_ISDIR(mydmode):
                                                 # install of destination is blocked by an existing directory with the same name
                                                 newdest = self._new_backup_path(mydest)
                                                 msg = []
                                                 msg.append("")
                                                 msg.append(_("Installation of a regular file is blocked by a directory:"))
                                                 msg.append("  '%s'" % mydest)
                                                 msg.append(_("This file will be merged with a different name:"))
                                                 msg.append("  '%s'" % newdest)
                                                 msg.append("")
-                                                self._elog_subprocess("eerror", "preinst", msg)
+                                                self._eerror("preinst", msg)
                                                 mydest = newdest
 
                                         elif stat.S_ISREG(mydmode) or (stat.S_ISLNK(mydmode) and os.path.exists(mydest) and stat.S_ISREG(os.stat(mydest)[stat.ST_MODE])):
                                                 # install of destination is blocked by an existing regular file,
                                                 # or by a symlink to an existing regular file;
                                                 # now, config file management may come into play.
                                                 # we only need to tweak mydest if cfg file management is in play.
                                                 if protected:
                                                         # we have a protection path; enable config file management.
                                                         cfgprot = 0
@@ skipping 163 matching lines @@
                                                 background=False,
                                                 commands=[phase],
                                                 scheduler=PollScheduler().sched_iface,
                                                 settings=self.settings)
                                         ebuild_phase.start()
                                         ebuild_phase.wait()
                                 else:
                                         self._scheduler.dblinkEbuildPhase(
                                                 self, mydbapi, myebuild, phase)
 
-                                elog_process(self.mycpv, self.settings)
+                                self._elog_process()
 
                                 if 'noclean' not in self.settings.features and \
                                         (retval == os.EX_OK or \
                                         'fail-clean' in self.settings.features):
                                         if myebuild is None:
                                                 myebuild = os.path.join(inforoot, self.pkg + ".ebuild")
 
                                         doebuild_environment(myebuild, "clean",
                                                 settings=self.settings, db=mydbapi)
                                         if self._scheduler is None:
@@ skipping 79 matching lines @@
         """
         @param myroot: ignored, settings['EROOT'] is used instead
         """
         myroot = None
         if settings is None:
                 raise TypeError("settings argument is required")
         if not os.access(settings['EROOT'], os.W_OK):
                 writemsg(_("Permission denied: access('%s', W_OK)\n") % settings['EROOT'],
                         noiselevel=-1)
                 return errno.EACCES
-        mylink = dblink(mycat, mypkg, settings=settings, treetype=mytree,
-                vartree=vartree, blockers=blockers, scheduler=scheduler)
-        return mylink.merge(pkgloc, infloc, myebuild=myebuild,
-                mydbapi=mydbapi, prev_mtimes=prev_mtimes)
+        background = (settings.get('PORTAGE_BACKGROUND') == '1')
+        merge_task = MergeProcess(
+                dblink=dblink, mycat=mycat, mypkg=mypkg, settings=settings,
+                treetype=mytree, vartree=vartree, scheduler=scheduler,
+                background=background, blockers=blockers, pkgloc=pkgloc,
+                infloc=infloc, myebuild=myebuild, mydbapi=mydbapi,
+                prev_mtimes=prev_mtimes, logfile=settings.get('PORTAGE_LOG_FILE'))
+        merge_task.start()
+        retcode = merge_task.wait()
+        return retcode
 
 def unmerge(cat, pkg, myroot=None, settings=None,
         mytrimworld=None, vartree=None,
         ldpath_mtimes=None, scheduler=None):
         """
         @param myroot: ignored, settings['EROOT'] is used instead
         @param mytrimworld: ignored
         """
         myroot = None
         if settings is None:
@@ skipping 120 matching lines @@
                                         encoding=object.__getattribute__(os, '_encoding'),
                                         errors='strict'), 'rb')
                                 try:
                                         tar.addfile(tarinfo, f)
                                 finally:
                                         f.close()
                 else:
                         tar.addfile(tarinfo)
                 if onProgress:
                         onProgress(maxval, curval)