/usr/share/pyshared/wimpiggy/error.py is in python-wimpiggy 0.0.7.36+dfsg-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | # This file is part of Parti.
# Copyright (C) 2008, 2009 Nathaniel Smith <njs@pobox.com>
# Parti is released under the terms of the GNU GPL v2, or, at your option, any
# later version. See the file COPYING for details.
# Goal: make it as easy and efficient as possible to manage the X errors that
# a WM is inevitably susceptible to. (E.g., if a window goes away while we
# are working on it.) On the one hand, we want to parcel operations into as
# broad chunks as possible that at treated as succeeding or failing as a whole
# (e.g., "setting up a new window", we don't really care how much was
# accomplished before the failure occurred). On the other, we do want to
# check for X errors often, for use in debugging (esp., this makes it more
# useful to run with -sync).
#
# The solution is to keep a stack of how deep we are in "transaction-like"
# operations -- a transaction is a series of operations where we don't care if
# we don't find about the failures until the end. We only sync when exiting a
# top-level transaction.
#
# The _synced and _unsynced variants differ in whether they assume the X
# connection was left in a synchronized state by the code they called (e.g.,
# if the last operation was an XGetProperty, then there is no need for us to
# do another XSync).
#
# (In this modern world, with WM's either on the same machine or over
# super-fast connections to the X server, everything running on fast
# computers... does being this careful to avoid sync's actually matter?)
__all__ = ["XError", "trap"]
import sys
import gtk.gdk
from wimpiggy.log import Logger
log = Logger()
class XError(Exception):
pass
xerror_to_name = None
def XErrorToName(xerror):
global xerror_to_name
try:
if xerror_to_name is None:
xerror_to_name = {}
from wimpiggy.lowlevel import const #@UnresolvedImport
for name,code in const.items():
if name=="Success" or name.startswith("Bad"):
xerror_to_name[code] = name
log("XErrorToName(..) initialized error names: %s", xerror_to_name)
if xerror.message in xerror_to_name:
return xerror_to_name.get(xerror.message)
except:
log.error("XErrorToName", exc_info=True)
return xerror
_exc_for_error = {}
# for error in _all_errors:
# exc_name = "X%s" % error
# exc_class = type(exc_name, (XError,), {})
# locals()[exc_name] = exc_class
# _exc_for_error[_lowlevel.const[error]] = exc_class
# gdk has its own depth tracking stuff, but we have to duplicate it here to
# minimize calls to XSync.
class _ErrorManager(object):
def __init__(self):
self.depth = 0
def _enter(self):
assert self.depth >= 0
gtk.gdk.error_trap_push()
self.depth += 1
def _exit(self, need_sync):
assert self.depth >= 0
self.depth -= 1
if self.depth == 0 and need_sync:
gtk.gdk.flush()
# This is a Xlib error constant (Success == 0)
error = gtk.gdk.error_trap_pop()
if error:
if error in _exc_for_error:
raise _exc_for_error[error](error)
else:
raise XError, error
def _call(self, need_sync, fun, args, kwargs):
# Goal: call the function. In all conditions, call _exit exactly once
# on the way out. However, if we are exiting because of an exception,
# then probably that exception is more informative than any XError
# that might also be raised, so suppress the XError in that case.
value = None
try:
self._enter()
value = fun(*args, **kwargs)
except:
exc_type, exc_value, exc_traceback = sys.exc_info()
try:
self._exit(need_sync)
except XError:
log("XError detected while already in unwind; discarding")
raise exc_type, exc_value, exc_traceback
self._exit(need_sync)
return value
def call_unsynced(self, fun, *args, **kwargs):
return self._call(True, fun, args, kwargs)
def call_synced(self, fun, *args, **kwargs):
return self._call(False, fun, args, kwargs)
call = call_unsynced
def swallow_unsynced(self, fun, *args, **kwargs):
try:
self.call_unsynced(fun, *args, **kwargs)
return True
except XError, e:
log("Ignoring X error: %s on %s", XErrorToName(e), fun)
return False
def swallow_synced(self, fun, *args, **kwargs):
try:
self.call_synced(fun, *args, **kwargs)
return True
except XError, e:
log("Ignoring X error: %s on %s", XErrorToName(e), fun)
return False
swallow = swallow_unsynced
def assert_out(self):
assert self.depth == 0
trap = _ErrorManager()
|