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	23194	0644	edit dl rm
components.py	14295	0644	edit dl rm
constants.py	544	0644	edit dl rm
context.py	4024	0644	edit dl rm
deprecate.py	26775	0644	edit dl rm
failure.py	26638	0644	edit dl rm
fakepwd.py	6138	0644	edit dl rm
filepath.py	58887	0644	edit dl rm
formmethod.py	11460	0644	edit dl rm
htmlizer.py	3541	0644	edit dl rm
lockfile.py	7718	0644	edit dl rm
log.py	22472	0644	edit dl rm
logfile.py	10083	0644	edit dl rm
modules.py	27141	0644	edit dl rm
monkey.py	2227	0644	edit dl rm
procutils.py	1419	0644	edit dl rm
randbytes.py	3959	0644	edit dl rm
rebuild.py	9268	0644	edit dl rm
reflect.py	19477	0644	edit dl rm
release.py	1192	0644	edit dl rm
roots.py	7404	0644	edit dl rm
runtime.py	6270	0644	edit dl rm
sendmsg.py	3420	0644	edit dl rm
shortcut.py	2253	0644	edit dl rm
syslog.py	3730	0644	edit dl rm
systemd.py	2834	0644	edit dl rm
text.py	5482	0644	edit dl rm
threadable.py	3297	0644	edit dl rm
threadpool.py	9840	0644	edit dl rm
twisted-completion.zsh	1371	0644	edit dl rm
url.py	244	0644	edit dl rm
urlpath.py	9084	0644	edit dl rm
usage.py	35010	0644	edit dl rm
util.py	27931	0644	edit dl rm
versions.py	322	0644	edit dl rm
win32.py	4317	0644	edit dl rm
zippath.py	9237	0644	edit dl rm
zipstream.py	9757	0644	edit dl rm
_appdirs.py	788	0644	edit dl rm
_inotify.py	3455	0644	edit dl rm
_oldstyle.py	2593	0644	edit dl rm
_release.py	18548	0644	edit dl rm
_setup.py	12953	0644	edit dl rm
_shellcomp.py	24830	0644	edit dl rm
_textattributes.py	9079	0644	edit dl rm
_tzhelper.py	3192	0644	edit dl rm
_url.py	253	0644	edit dl rm
__init__.py	674	0644	edit dl rm

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

# -*- 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 __future__ import division, absolute_import

import threading

from twisted._threads import pool as _pool
from twisted.python import log, context
from twisted.python.failure import Failure
from twisted.python._oldstyle import _oldStyle

WorkerStop = object()

@_oldStyle
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
    workers = 0
    name = None

threadFactory = threading.Thread
    currentThread = staticmethod(threading.currentThread)
    _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 = []

def trackingThreadFactory(*a, **kw):
            thread = self.threadFactory(*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(object):
            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 "PoolThread-%s-%s" % (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()
                ok = True
            except:
                result = Failure()
                ok = False

inContext.theWork = None
            if inContext.onResult is not None:
                inContext.onResult(ok, result)
                inContext.onResult = None
            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(ctx, func, *args, **kw)
        inContext.onResult = onResult

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('waiters: %s' % (self.waiters,))
        log.msg('workers: %s' % (self.working,))
        log.msg('total: %s'   % (self.threads,))

cmd: