/usr/share/pyshared/tp/client/threadcheck.py is in python-tp-client 0.3.2-2.
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 | import threading
from decorator import decorator
@decorator
def thread_check_callable(func, self, *args, **kw):
assert self._thread is threading.currentThread(), "%s can only be called by %s not %s" % (func, self._thread, threading.currentThread())
return func(self, *args, **kw)
@decorator
def thread_check_init(func, self, *args, **kw):
self._thread = threading.currentThread()
return func(self, *args, **kw)
def thread_safe(func):
"""
Mark this function as thread safe so can be called accross thread
boundaries.
"""
func.__threadsafe = True
return func
# FIXME: This doesn't also check the classes base class!
import inspect
class thread_checker(type):
"""
This class can be set as the metaclass for an object to make sure that it
is only access from a single thread.
Methods can be marked "thread_safe" with the thread_safe decorator also
found in this module.
"""
def __new__(cls,classname,bases,classdict):
for attr, item in classdict.items():
if attr == "__init__":
classdict[attr] = thread_check_init(item)
elif inspect.isfunction(item) and not hasattr(item, "__threadsafe"):
classdict[attr] = thread_check_callable(item)
t = type.__new__(cls,classname,bases,classdict)
return t
__all__ = ["thread_checker", "thread_safe"]
if __name__ == "__main__":
class B(object):
__metaclass__ = thread_checker
def __init__(self):
pass
test_simple = 2
test_complex = [1, 3]
def test_unsafe(self):
return "test_unsafe worked"
@thread_safe
def test_safe(self):
return "test_safe worked"
class t1(threading.Thread):
def __init__(self, ishouldnotaccess):
self.i = ishouldnotaccess
threading.Thread.__init__(self)
def run(self):
print self.i.test_simple
print self.i.test_complex
print self.i.test_safe()
print self.i.test_unsafe()
ot = t1(B())
print ot.i.test_safe
print inspect.getargspec(ot.i.test_safe)
print ot.i.test_unsafe
print inspect.getargspec(ot.i.test_unsafe)
ot.run()
# this should fail
ot.start()
import time
time.sleep(10)
|