/usr/lib/python3/dist-packages/twisted/python/threadable.py is in python3-twisted-experimental 13.2.0-0ubuntu1.
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 139 | # -*- test-case-name: twisted.python.test_threadable -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
A module to provide some very basic threading primitives, such as
synchronization.
"""
from __future__ import division, absolute_import
from functools import wraps
class DummyLock(object):
"""
Hack to allow locks to be unpickled on an unthreaded system.
"""
def __reduce__(self):
return (unpickle_lock, ())
def unpickle_lock():
if threadingmodule is not None:
return XLock()
else:
return DummyLock()
unpickle_lock.__safe_for_unpickling__ = True
def _synchPre(self):
if '_threadable_lock' not in self.__dict__:
_synchLockCreator.acquire()
if '_threadable_lock' not in self.__dict__:
self.__dict__['_threadable_lock'] = XLock()
_synchLockCreator.release()
self._threadable_lock.acquire()
def _synchPost(self):
self._threadable_lock.release()
def _sync(klass, function):
@wraps(function)
def sync(self, *args, **kwargs):
_synchPre(self)
try:
return function(self, *args, **kwargs)
finally:
_synchPost(self)
return sync
def synchronize(*klasses):
"""
Make all methods listed in each class' synchronized attribute synchronized.
The synchronized attribute should be a list of strings, consisting of the
names of methods that must be synchronized. If we are running in threaded
mode these methods will be wrapped with a lock.
"""
if threadingmodule is not None:
for klass in klasses:
for methodName in klass.synchronized:
sync = _sync(klass, klass.__dict__[methodName])
setattr(klass, methodName, sync)
def init(with_threads=1):
"""Initialize threading.
Don't bother calling this. If it needs to happen, it will happen.
"""
global threaded, _synchLockCreator, XLock
if with_threads:
if not threaded:
if threadingmodule is not None:
threaded = True
class XLock(threadingmodule._RLock, object):
def __reduce__(self):
return (unpickle_lock, ())
_synchLockCreator = XLock()
else:
raise RuntimeError("Cannot initialize threading, platform lacks thread support")
else:
if threaded:
raise RuntimeError("Cannot uninitialize threads")
else:
pass
_dummyID = object()
def getThreadID():
if threadingmodule is None:
return _dummyID
return threadingmodule.currentThread().ident
def isInIOThread():
"""Are we in the thread responsable for I/O requests (the event loop)?
"""
return ioThread == getThreadID()
def registerAsIOThread():
"""Mark the current thread as responsable for I/O requests.
"""
global ioThread
ioThread = getThreadID()
ioThread = None
threaded = False
try:
import threading as threadingmodule
except ImportError:
threadingmodule = None
else:
init(True)
__all__ = ['isInIOThread', 'registerAsIOThread', 'getThreadID', 'XLock']
|