FM / python

/usr/lib/python3/dist-packages/twisted/python/

/usr/lib/python3/dist-packages/twisted/python

Name	Size	Mode	Actions
test/	-	0755	rm
_pydoctortemplates/	-	0755	rm
__pycache__/	-	0755	rm
compat.py	16931	0644	edit dl rm
components.py	14199	0644	edit dl rm
constants.py	513	0644	edit dl rm
context.py	4054	0644	edit dl rm
deprecate.py	27668	0644	edit dl rm
failure.py	27007	0644	edit dl rm
fakepwd.py	6732	0644	edit dl rm
filepath.py	54023	0644	edit dl rm
formmethod.py	12107	0644	edit dl rm
htmlizer.py	3626	0644	edit dl rm
lockfile.py	8026	0644	edit dl rm
log.py	22302	0644	edit dl rm
logfile.py	10119	0644	edit dl rm
modules.py	26715	0644	edit dl rm
monkey.py	2164	0644	edit dl rm
procutils.py	1371	0644	edit dl rm
randbytes.py	3459	0644	edit dl rm
rebuild.py	7122	0644	edit dl rm
reflect.py	20480	0644	edit dl rm
release.py	1104	0644	edit dl rm
roots.py	7178	0644	edit dl rm
runtime.py	5924	0644	edit dl rm
sendmsg.py	2682	0644	edit dl rm
shortcut.py	2302	0644	edit dl rm
syslog.py	3652	0644	edit dl rm
systemd.py	2987	0644	edit dl rm
text.py	5417	0644	edit dl rm
threadable.py	3327	0644	edit dl rm
threadpool.py	10203	0644	edit dl rm
twisted-completion.zsh	1371	0644	edit dl rm
url.py	244	0644	edit dl rm
urlpath.py	8447	0644	edit dl rm
usage.py	34577	0644	edit dl rm
util.py	27432	0644	edit dl rm
versions.py	273	0644	edit dl rm
win32.py	4794	0644	edit dl rm
zippath.py	9028	0644	edit dl rm
zipstream.py	9680	0644	edit dl rm
_appdirs.py	820	0644	edit dl rm
_inotify.py	3496	0644	edit dl rm
_pydoctor.py	6735	0644	edit dl rm
_release.py	18879	0644	edit dl rm
_shellcomp.py	25279	0644	edit dl rm
_textattributes.py	9097	0644	edit dl rm
_tzhelper.py	3128	0644	edit dl rm
_url.py	228	0644	edit dl rm
__init__.py	598	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/python/threadpool.py (10203B)

# -*- test-case-name: twisted.test.test_threadpool -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
twisted.python.threadpool: a pool of threads to which we dispatch tasks.

In most cases you can just use C{reactor.callInThread} and friends
instead of creating a thread pool directly.
"""

from threading import Thread, current_thread
from typing import List

from twisted._threads import pool as _pool
from twisted.python import context, log
from twisted.python.deprecate import deprecated
from twisted.python.failure import Failure
from twisted.python.versions import Version

WorkerStop = object()

class ThreadPool:
    """
    This class (hopefully) generalizes the functionality of a pool of threads
    to which work can be dispatched.

L{callInThread} and L{stop} should only be called from a single thread.

@ivar started: Whether or not the thread pool is currently running.
    @type started: L{bool}

@ivar threads: List of workers currently running in this thread pool.
    @type threads: L{list}

@ivar _pool: A hook for testing.
    @type _pool: callable compatible with L{_pool}
    """

min = 5
    max = 20
    joined = False
    started = False
    name = None

threadFactory = Thread
    currentThread = staticmethod(
        deprecated(
            version=Version("Twisted", 22, 1, 0),
            replacement="threading.current_thread",
        )(current_thread)
    )
    _pool = staticmethod(_pool)

def __init__(self, minthreads=5, maxthreads=20, name=None):
        """
        Create a new threadpool.

@param minthreads: minimum number of threads in the pool
        @type minthreads: L{int}

@param maxthreads: maximum number of threads in the pool
        @type maxthreads: L{int}

@param name: The name to give this threadpool; visible in log messages.
        @type name: native L{str}
        """
        assert minthreads >= 0, "minimum is negative"
        assert minthreads <= maxthreads, "minimum is greater than maximum"
        self.min = minthreads
        self.max = maxthreads
        self.name = name
        self.threads: List[Thread] = []

def trackingThreadFactory(*a, **kw):
            thread = self.threadFactory(  # type: ignore[misc]
                *a, name=self._generateName(), **kw
            )
            self.threads.append(thread)
            return thread

def currentLimit():
            if not self.started:
                return 0
            return self.max

self._team = self._pool(currentLimit, trackingThreadFactory)

@property
    def workers(self):
        """
        For legacy compatibility purposes, return a total number of workers.

@return: the current number of workers, both idle and busy (but not
            those that have been quit by L{ThreadPool.adjustPoolsize})
        @rtype: L{int}
        """
        stats = self._team.statistics()
        return stats.idleWorkerCount + stats.busyWorkerCount

@property
    def working(self):
        """
        For legacy compatibility purposes, return the number of busy workers as
        expressed by a list the length of that number.

@return: the number of workers currently processing a work item.
        @rtype: L{list} of L{None}
        """
        return [None] * self._team.statistics().busyWorkerCount

@property
    def waiters(self):
        """
        For legacy compatibility purposes, return the number of idle workers as
        expressed by a list the length of that number.

@return: the number of workers currently alive (with an allocated
            thread) but waiting for new work.
        @rtype: L{list} of L{None}
        """
        return [None] * self._team.statistics().idleWorkerCount

@property
    def _queue(self):
        """
        For legacy compatibility purposes, return an object with a C{qsize}
        method that indicates the amount of work not yet allocated to a worker.

@return: an object with a C{qsize} method.
        """

class NotAQueue:
            def qsize(q):
                """
                Pretend to be a Python threading Queue and return the
                number of as-yet-unconsumed tasks.

@return: the amount of backlogged work not yet dispatched to a
                    worker.
                @rtype: L{int}
                """
                return self._team.statistics().backloggedWorkCount

return NotAQueue()

q = _queue  # Yes, twistedchecker, I want a single-letter
    # attribute name.

def start(self):
        """
        Start the threadpool.
        """
        self.joined = False
        self.started = True
        # Start some threads.
        self.adjustPoolsize()
        backlog = self._team.statistics().backloggedWorkCount
        if backlog:
            self._team.grow(backlog)

def startAWorker(self):
        """
        Increase the number of available workers for the thread pool by 1, up
        to the maximum allowed by L{ThreadPool.max}.
        """
        self._team.grow(1)

def _generateName(self):
        """
        Generate a name for a new pool thread.

@return: A distinctive name for the thread.
        @rtype: native L{str}
        """
        return f"PoolThread-{self.name or id(self)}-{self.workers}"

def stopAWorker(self):
        """
        Decrease the number of available workers by 1, by quitting one as soon
        as it's idle.
        """
        self._team.shrink(1)

def __setstate__(self, state):
        setattr(self, "__dict__", state)
        ThreadPool.__init__(self, self.min, self.max)

def __getstate__(self):
        state = {}
        state["min"] = self.min
        state["max"] = self.max
        return state

def callInThread(self, func, *args, **kw):
        """
        Call a callable object in a separate thread.

@param func: callable object to be called in separate thread

@param args: positional arguments to be passed to C{func}

@param kw: keyword args to be passed to C{func}
        """
        self.callInThreadWithCallback(None, func, *args, **kw)

def callInThreadWithCallback(self, onResult, func, *args, **kw):
        """
        Call a callable object in a separate thread and call C{onResult} with
        the return value, or a L{twisted.python.failure.Failure} if the
        callable raises an exception.

The callable is allowed to block, but the C{onResult} function must not
        block and should perform as little work as possible.

A typical action for C{onResult} for a threadpool used with a Twisted
        reactor would be to schedule a L{twisted.internet.defer.Deferred} to
        fire in the main reactor thread using C{.callFromThread}.  Note that
        C{onResult} is called inside the separate thread, not inside the
        reactor thread.

@param onResult: a callable with the signature C{(success, result)}.
            If the callable returns normally, C{onResult} is called with
            C{(True, result)} where C{result} is the return value of the
            callable.  If the callable throws an exception, C{onResult} is
            called with C{(False, failure)}.

Optionally, C{onResult} may be L{None}, in which case it is not
            called at all.

@param func: callable object to be called in separate thread

@param args: positional arguments to be passed to C{func}

@param kw: keyword arguments to be passed to C{func}
        """
        if self.joined:
            return
        ctx = context.theContextTracker.currentContext().contexts[-1]

def inContext():
            try:
                result = inContext.theWork()  # type: ignore[attr-defined]
                ok = True
            except BaseException:
                result = Failure()
                ok = False

inContext.theWork = None  # type: ignore[attr-defined]
            if inContext.onResult is not None:  # type: ignore[attr-defined]
                inContext.onResult(ok, result)  # type: ignore[attr-defined]
                inContext.onResult = None  # type: ignore[attr-defined]
            elif not ok:
                log.err(result)

# Avoid closing over func, ctx, args, kw so that we can carefully
        # manage their lifecycle.  See
        # test_threadCreationArgumentsCallInThreadWithCallback.
        inContext.theWork = lambda: context.call(  # type: ignore[attr-defined]
            ctx, func, *args, **kw
        )
        inContext.onResult = onResult  # type: ignore[attr-defined]

self._team.do(inContext)

def stop(self):
        """
        Shutdown the threads in the threadpool.
        """
        self.joined = True
        self.started = False
        self._team.quit()
        for thread in self.threads:
            thread.join()

def adjustPoolsize(self, minthreads=None, maxthreads=None):
        """
        Adjust the number of available threads by setting C{min} and C{max} to
        new values.

@param minthreads: The new value for L{ThreadPool.min}.

@param maxthreads: The new value for L{ThreadPool.max}.
        """
        if minthreads is None:
            minthreads = self.min
        if maxthreads is None:
            maxthreads = self.max

assert minthreads >= 0, "minimum is negative"
        assert minthreads <= maxthreads, "minimum is greater than maximum"

self.min = minthreads
        self.max = maxthreads
        if not self.started:
            return

# Kill of some threads if we have too many.
        if self.workers > self.max:
            self._team.shrink(self.workers - self.max)
        # Start some threads if we have too few.
        if self.workers < self.min:
            self._team.grow(self.min - self.workers)

def dumpStats(self):
        """
        Dump some plain-text informational messages to the log about the state
        of this L{ThreadPool}.
        """
        log.msg(f"waiters: {self.waiters}")
        log.msg(f"workers: {self.working}")
        log.msg(f"total: {self.threads}")

cmd: