Gecko [ aodai.devttt.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
_pydoctortemplates	[ DIR ]	drwxr-xr-x
test	[ DIR ]	drwxr-xr-x
__init__.py	598 B	-rw-r--r--
_appdirs.py	820 B	-rw-r--r--
_inotify.py	3.41 KB	-rw-r--r--
_pydoctor.py	6.58 KB	-rw-r--r--
_release.py	18.44 KB	-rw-r--r--
_shellcomp.py	24.69 KB	-rw-r--r--
_textattributes.py	8.88 KB	-rw-r--r--
_tzhelper.py	3.05 KB	-rw-r--r--
_url.py	228 B	-rw-r--r--
compat.py	16.53 KB	-rw-r--r--
components.py	13.87 KB	-rw-r--r--
constants.py	513 B	-rw-r--r--
context.py	3.96 KB	-rw-r--r--
deprecate.py	27.02 KB	-rw-r--r--
failure.py	26.37 KB	-rw-r--r--
fakepwd.py	6.57 KB	-rw-r--r--
filepath.py	52.76 KB	-rw-r--r--
formmethod.py	11.82 KB	-rw-r--r--
htmlizer.py	3.54 KB	-rw-r--r--
lockfile.py	7.84 KB	-rw-r--r--
log.py	21.78 KB	-rw-r--r--
logfile.py	9.88 KB	-rw-r--r--
modules.py	26.09 KB	-rw-r--r--
monkey.py	2.11 KB	-rw-r--r--
procutils.py	1.34 KB	-rw-r--r--
randbytes.py	3.38 KB	-rw-r--r--
rebuild.py	6.96 KB	-rw-r--r--
reflect.py	20 KB	-rw-r--r--
release.py	1.08 KB	-rw-r--r--
roots.py	7.01 KB	-rw-r--r--
runtime.py	5.79 KB	-rw-r--r--
sendmsg.py	2.62 KB	-rw-r--r--
shortcut.py	2.25 KB	-rw-r--r--
syslog.py	3.57 KB	-rw-r--r--
systemd.py	2.92 KB	-rw-r--r--
text.py	5.29 KB	-rw-r--r--
threadable.py	3.25 KB	-rw-r--r--
threadpool.py	9.96 KB	-rw-r--r--
twisted-completion.zsh	1.34 KB	-rw-r--r--
url.py	244 B	-rw-r--r--
urlpath.py	8.25 KB	-rw-r--r--
usage.py	33.77 KB	-rw-r--r--
util.py	26.79 KB	-rw-r--r--
versions.py	273 B	-rw-r--r--
win32.py	4.68 KB	-rw-r--r--
zippath.py	8.82 KB	-rw-r--r--
zipstream.py	9.45 KB	-rw-r--r--

Code Editor : threadable.py

# -*- 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 functools import wraps

class DummyLock:
    """
    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  # type: ignore[attr-defined]

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):
                    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.current_thread().ident

def isInIOThread():
    """Are we in the thread responsible for I/O requests (the event loop)?"""
    return ioThread == getThreadID()

def registerAsIOThread():
    """Mark the current thread as responsible for I/O requests."""
    global ioThread
    ioThread = getThreadID()

ioThread = None
threaded = False
# Define these globals which might be overwritten in init().
_synchLockCreator = None
XLock = None

try:
    import threading as _threadingmodule
except ImportError:
    threadingmodule = None
else:
    threadingmodule = _threadingmodule
    init(True)

__all__ = ["isInIOThread", "registerAsIOThread", "getThreadID", "XLock"]

${this.title}

Code Editor : threadable.py