"""
__slots__ = ("input_files", "scheduler",) + \
- ("pid", "registered", "_reg_ids", "_read_data")
+ ("pid", "_read_data", "_registered", "_reg_ids")
_bufsize = 4096
fcntl.fcntl(f.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK)
self._reg_ids.add(self.scheduler.register(f.fileno(),
PollConstants.POLLIN, self._output_handler))
- self.registered = True
+ self._registered = True
def isAlive(self):
- return self.registered
+ return self._registered
def _wait(self):
if self.returncode is not None:
return self.returncode
- if self.registered:
+
+ if self._registered:
self.scheduler.schedule(self._reg_ids)
+ self._unregister()
+
self.returncode = os.EX_OK
return self.returncode
if buf:
self._read_data.append(buf.tostring())
else:
- self.registered = False
+ self._unregister()
+ self.wait()
+
+ return self._registered
+
+ def _unregister(self):
+ """
+ Unregister from the scheduler and close open files.
+ """
+
+ self._registered = False
+
+ if self._reg_ids is not None:
for reg_id in self._reg_ids:
self.scheduler.unregister(reg_id)
+ self._reg_ids = None
- for f in files.values():
+ if self.input_files is not None:
+ for f in self.input_files.itervalues():
f.close()
-
- self.wait()
-
- return self.registered
+ self.input_files = None
class CompositeTask(AsynchronousTask):
class SubProcess(AsynchronousTask):
- __slots__ = ("scheduler",) + ("pid", "registered", "_reg_id")
+ __slots__ = ("scheduler",) + ("pid", "_files", "_registered", "_reg_id")
# A file descriptor is required for the scheduler to monitor changes from
# inside a poll() loop. When logging is not enabled, create a pipe just to
return self.returncode
if self.pid is None:
return self.returncode
+ if self._registered:
+ return self.returncode
try:
retval = os.waitpid(self.pid, os.WNOHANG)
self.returncode is None
def _wait(self):
- if self.registered:
- self.scheduler.schedule(self._reg_id)
+
if self.returncode is not None:
return self.returncode
+
+ if self._registered:
+ self.scheduler.schedule(self._reg_id)
+ self._unregister()
+ if self.returncode is not None:
+ return self.returncode
+
try:
wait_retval = os.waitpid(self.pid, 0)
except OSError, e:
self._set_returncode((self.pid, 1))
else:
self._set_returncode(wait_retval)
+
return self.returncode
+ def _unregister(self):
+ """
+ Unregister from the scheduler and close open files.
+ """
+
+ self._registered = False
+
+ if self._reg_id is not None:
+ self.scheduler.unregister(self._reg_id)
+ self._reg_id = None
+
+ if self._files is not None:
+ for f in self._files.itervalues():
+ f.close()
+ self._files = None
+
def _set_returncode(self, wait_retval):
retval = wait_retval[1]
"uid", "gid", "groups", "umask", "logfile",
"path_lookup", "pre_exec")
- __slots__ = ("args", "files") + \
+ __slots__ = ("args",) + \
_spawn_kwarg_names
_file_names = ("process", "out")
sys.stderr.flush()
logfile = self.logfile
- self.files = self._files_dict()
- files = self.files
+ self._files = self._files_dict()
+ files = self._files
master_fd, slave_fd = os.pipe()
fcntl.fcntl(master_fd, fcntl.F_SETFL,
files.process = os.fdopen(master_fd, 'r')
self._reg_id = self.scheduler.register(files.process.fileno(),
PollConstants.POLLIN, output_handler)
- self.registered = True
+ self._registered = True
def _output_handler(self, fd, event):
- files = self.files
+ files = self._files
buf = array.array('B')
try:
buf.fromfile(files.process, self._bufsize)
buf.tofile(files.out)
files.out.flush()
else:
- self.registered = False
- self.scheduler.unregister(self._reg_id)
-
- for f in files.values():
- f.close()
-
+ self._unregister()
self.wait()
- return self.registered
+ return self._registered
def _dummy_handler(self, fd, event):
"""
the only purpose of the pipe is to allow the scheduler to
monitor the process from inside a poll() loop.
"""
- files = self.files
+ files = self._files
buf = array.array('B')
try:
buf.fromfile(files.process, self._bufsize)
if buf:
pass
else:
- self.registered = False
- self.scheduler.unregister(self._reg_id)
-
- for f in files.values():
- f.close()
-
+ self._unregister()
self.wait()
- return self.registered
+ return self._registered
class EbuildFetcher(SpawnProcess):
__slots__ = ("cpv", "ebuild_path", "fd_pipes", "metadata_callback",
"ebuild_mtime", "portdb", "repo_path", "settings") + \
- ("files", "_raw_metadata")
+ ("_raw_metadata",)
_file_names = ("ebuild",)
_files_dict = slot_dict_class(_file_names, prefix="")
sys.stderr.flush()
fd_pipes_orig = fd_pipes.copy()
- self.files = self._files_dict()
- files = self.files
+ self._files = self._files_dict()
+ files = self._files
master_fd, slave_fd = os.pipe()
fcntl.fcntl(master_fd, fcntl.F_SETFL,
files.ebuild = os.fdopen(master_fd, 'r')
self._reg_id = self.scheduler.register(files.ebuild.fileno(),
PollConstants.POLLIN, self._output_handler)
- self.registered = True
+ self._registered = True
def _output_handler(self, fd, event):
- files = self.files
+ files = self._files
self._raw_metadata.append(files.ebuild.read())
if not self._raw_metadata[-1]:
- self.registered = False
- self.scheduler.unregister(self._reg_id)
-
- for f in files.values():
- f.close()
-
+ self._unregister()
self.wait()
if self.returncode == os.EX_OK:
self.metadata_callback(self.cpv, self.ebuild_path,
self.repo_path, metadata, self.ebuild_mtime)
- return self.registered
+ return self._registered
class EbuildPhase(SubProcess):
__slots__ = ("fd_pipes", "phase", "pkg",
- "settings", "tree",
- "files")
+ "settings", "tree")
_file_names = ("log", "stdout", "ebuild")
_files_dict = slot_dict_class(_file_names, prefix="")
sys.stderr.flush()
fd_pipes_orig = fd_pipes.copy()
- self.files = self._files_dict()
- files = self.files
+ self._files = self._files_dict()
+ files = self._files
got_pty = False
portage._doebuild_exit_status_unlink(
files.ebuild = os.fdopen(master_fd, 'r')
self._reg_id = self.scheduler.register(files.ebuild.fileno(),
PollConstants.POLLIN, output_handler)
- self.registered = True
+ self._registered = True
def _output_handler(self, fd, event):
- files = self.files
+ files = self._files
buf = array.array('B')
try:
buf.fromfile(files.ebuild, self._bufsize)
buf.tofile(files.log)
files.log.flush()
else:
- self.registered = False
- self.scheduler.unregister(self._reg_id)
-
- for f in files.values():
- f.close()
-
+ self._unregister()
self.wait()
- return self.registered
+ return self._registered
def _dummy_handler(self, fd, event):
"""
the only purpose of the pipe is to allow the scheduler to
monitor the process from inside a poll() loop.
"""
- files = self.files
+ files = self._files
buf = array.array('B')
try:
buf.fromfile(files.ebuild, self._bufsize)
if buf:
pass
else:
- self.registered = False
- self.scheduler.unregister(self._reg_id)
-
- for f in files.values():
- f.close()
-
+ self._unregister()
self.wait()
- return self.registered
+ return self._registered
def _set_returncode(self, wait_retval):
SubProcess._set_returncode(self, wait_retval)
self.returncode = portage._post_src_install_checks(settings)
elif self.phase == "preinst":
-
if self.returncode == os.EX_OK:
portage._doebuild_exit_status_unlink(
settings.get("EBUILD_EXIT_STATUS_FILE"))
state_changed = False
for task in list(running_tasks):
- if hasattr(task, "registered") and task.registered:
- continue
if task.poll() is not None:
state_changed = True
self._max_jobs = 1
self._max_load = None
self._jobs = 0
+ self._poll_event_queue = []
self._poll_event_handlers = {}
self._poll_event_handler_ids = {}
# Increment id for each new handler.
return True
+ def _next_poll_event(self):
+ """
+ Since the _schedule_wait() loop is called by event
+ handlers from _poll_loop(), maintain a central event
+ queue for both of them to share events from a single
+ poll() call.
+ """
+ if not self._poll_event_queue:
+ self._poll_event_queue.extend(self._poll.poll())
+ return self._poll_event_queue.pop()
+
def _poll_loop(self):
event_handlers = self._poll_event_handlers
- poll = self._poll.poll
- state_change = 0
+ event_handled = False
while event_handlers:
- for f, event in poll():
- handler, reg_id = event_handlers[f]
- if not handler(f, event):
- state_change += 1
+ f, event = self._next_poll_event()
+ handler, reg_id = event_handlers[f]
+ handler(f, event)
+ event_handled = True
- if not state_change:
+ if not event_handled:
raise AssertionError("tight loop")
def _register(self, f, eventmask, handler):
"""
event_handlers = self._poll_event_handlers
handler_ids = self._poll_event_handler_ids
- poll = self._poll.poll
+ event_handled = False
if isinstance(wait_ids, int):
wait_ids = frozenset([wait_ids])
while wait_ids.intersection(handler_ids):
- for f, event in poll():
- handler, reg_id = event_handlers[f]
- handler(f, event)
+ f, event = self._next_poll_event()
+ handler, reg_id = event_handlers[f]
+ handler(f, event)
+ event_handled = True
+
+ return event_handled
class QueueScheduler(PollScheduler):
self._pkg_queue = []
self._completed_tasks = set()
self._failed_pkgs = []
+ self._failed_pkgs_all = []
+ self._failed_pkgs_die_msgs = []
self._failed_fetches = []
self._parallel_fetch = False
merge_count = len([x for x in mergelist \
if self._pkg_count.curval >= self._pkg_count.maxval:
return
+ self._main_loop_cleanup()
+
logger = self._logger
pkg_count = self._pkg_count
mtimedb = self._mtimedb
keep_going = "--keep-going" in self.myopts
mtimedb = self._mtimedb
+ failed_pkgs = self._failed_pkgs
while True:
rval = self._merge()
mergelist = self._mtimedb["resume"].get("mergelist")
if not mergelist:
break
- if mergelist[0][-1] != "merge":
- break
- # Skip the first one because it failed to build or install.
- pkg_key = tuple(mergelist[0])
- del mergelist[0]
- failed_pkg = None
- for task in self._mergelist:
- if task == pkg_key:
- failed_pkg = task
- break
- if failed_pkg is None:
+ if not failed_pkgs:
break
+
+ for failed_pkg, returncode in failed_pkgs:
+ mergelist.remove(list(failed_pkg))
+
+ self._failed_pkgs_all.extend(failed_pkgs)
+ del failed_pkgs[:]
+
if not mergelist:
break
self._logger.log(" *** Finished. Cleaning up...")
+ background = self._max_jobs > 1
+ if self._failed_pkgs_all and background and \
+ self._failed_pkgs_die_msgs and \
+ not _flush_elog_mod_echo():
+
+ printer = portage.output.EOutput()
+ for mysettings, key, logentries in self._failed_pkgs_die_msgs:
+ root_msg = ""
+ if mysettings["ROOT"] != "/":
+ root_msg = " merged to %s" % mysettings["ROOT"]
+ print
+ printer.einfo("Error messages for package %s%s:" % \
+ (colorize("INFORM", key), root_msg))
+ print
+ for phase in portage.const.EBUILD_PHASES:
+ if phase not in logentries:
+ continue
+ for msgtype, msgcontent in logentries[phase]:
+ if isinstance(msgcontent, basestring):
+ msgcontent = [msgcontent]
+ for line in msgcontent:
+ printer.eerror(line.strip("\n"))
+
+ if len(self._failed_pkgs_all) > 1:
+ _flush_elog_mod_echo()
+ msg = "The following packages have " + \
+ "failed to build or install:"
+ prefix = bad(" * ")
+ writemsg(prefix + "\n", noiselevel=-1)
+ from textwrap import wrap
+ for line in wrap(msg, 72):
+ writemsg("%s%s\n" % (prefix, line), noiselevel=-1)
+ writemsg(prefix + "\n", noiselevel=-1)
+ for pkg, returncode in self._failed_pkgs_all:
+ writemsg("%s\t%s\n" % (prefix,
+ colorize("INFORM", str(pkg))),
+ noiselevel=-1)
+ writemsg(prefix + "\n", noiselevel=-1)
+
return rval
+ def _elog_listener(self, mysettings, key, logentries, fulltext):
+ errors = portage.elog.filter_loglevels(logentries, ["ERROR"])
+ if errors:
+ self._failed_pkgs_die_msgs.append(
+ (mysettings, key, errors))
+
def _add_packages(self):
pkg_queue = self._pkg_queue
for pkg in self._mergelist:
# --resume still works after being interrupted
# by reboot, sigkill or similar.
mtimedb = self._mtimedb
- del mtimedb["resume"]["mergelist"][0]
+ mtimedb["resume"]["mergelist"].remove(list(pkg))
if not mtimedb["resume"]["mergelist"]:
del mtimedb["resume"]
mtimedb.commit()
self._add_packages()
pkg_queue = self._pkg_queue
failed_pkgs = self._failed_pkgs
+ portage.elog._emerge_elog_listener = self._elog_listener
rval = os.EX_OK
try:
self._main_loop()
finally:
- # discard remaining packages if necessary
- del pkg_queue[:]
- self._completed_tasks.clear()
- self._digraph = None
- self._task_queues.fetch.clear()
-
- # discard any failures and return the
- # exist status of the last one
+ self._main_loop_cleanup()
+ portage.elog._emerge_elog_listener = None
if failed_pkgs:
pkg, rval = failed_pkgs[-1]
- del failed_pkgs[:]
-
return rval
+ def _main_loop_cleanup(self):
+ del self._pkg_queue[:]
+ self._completed_tasks.clear()
+ self._digraph = None
+ self._task_queues.fetch.clear()
+
def _choose_pkg(self):
"""
Choose a task that has all it's dependencies satisfied.
print "Use " + colorize("GOOD", "eselect news") + " to read news items."
print
+def _flush_elog_mod_echo():
+ """
+ Dump the mod_echo output now so that our other
+ notifications are shown last.
+ @rtype: bool
+ @returns: True if messages were shown, False otherwise.
+ """
+ messages_shown = False
+ try:
+ from portage.elog import mod_echo
+ except ImportError:
+ pass # happens during downgrade to a version without the module
+ else:
+ messages_shown = bool(mod_echo._items)
+ mod_echo.finalize()
+ return messages_shown
+
def post_emerge(trees, mtimedb, retval):
"""
Misc. things to run at the end of a merge session.
exit_msg = " *** exiting unsuccessfully with status '%s'." % retval
emergelog("notitles" not in settings.features, exit_msg)
- from portage.util import normalize_path
- # Dump the mod_echo output now so that our other notifications are shown
- # last.
- try:
- from portage.elog import mod_echo
- except ImportError:
- pass # happens during downgrade to a version without the module
- else:
- mod_echo.finalize()
+ _flush_elog_mod_echo()
vdb_path = os.path.join(target_root, portage.VDB_PATH)
portage.util.ensure_dirs(vdb_path)
# will be infinite. Therefore, if that case ever
# occurs for some reason, raise the exception to
# break out of the loop.
- if not pruned_mergelist or \
- len(pruned_mergelist) == len(mergelist):
- raise
+ if len(pruned_mergelist) == len(mergelist):
+ raise AssertionError("tight loop")
mergelist[:] = pruned_mergelist
dropped_tasks.update(unsatisfied_parents)
del e, graph, traversed_nodes, \
projects / portage.git / commitdiff