+++ /dev/null
-# Copyright (c) 2007-2009, Linden Research, Inc.
-# Copyright (c) 2007, IBM Corp.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import imp
-import os
-import sys
-import traceback
-
-from eventlet import event, greenio, greenthread, patcher, timeout
-from eventlet.support import six
-
-__all__ = ['execute', 'Proxy', 'killall', 'set_num_threads']
-
-
-EXC_CLASSES = (Exception, timeout.Timeout)
-SYS_EXCS = (GeneratorExit, KeyboardInterrupt, SystemExit)
-
-QUIET = True
-
-socket = patcher.original('socket')
-threading = patcher.original('threading')
-if six.PY2:
- Queue_module = patcher.original('Queue')
-if six.PY3:
- Queue_module = patcher.original('queue')
-
-Empty = Queue_module.Empty
-Queue = Queue_module.Queue
-
-_bytetosend = ' '.encode()
-_coro = None
-_nthreads = int(os.environ.get('EVENTLET_THREADPOOL_SIZE', 20))
-_reqq = _rspq = None
-_rsock = _wsock = None
-_setup_already = False
-_threads = []
-
-
-def tpool_trampoline():
- global _rspq
- while True:
- try:
- _c = _rsock.recv(1)
- assert _c
- except ValueError:
- break # will be raised when pipe is closed
- while not _rspq.empty():
- try:
- (e, rv) = _rspq.get(block=False)
- e.send(rv)
- e = rv = None
- except Empty:
- pass
-
-
-def tworker():
- global _rspq
- while True:
- try:
- msg = _reqq.get()
- except AttributeError:
- return # can't get anything off of a dud queue
- if msg is None:
- return
- (e, meth, args, kwargs) = msg
- rv = None
- try:
- rv = meth(*args, **kwargs)
- except SYS_EXCS:
- raise
- except EXC_CLASSES:
- rv = sys.exc_info()
- # test_leakage_from_tracebacks verifies that the use of
- # exc_info does not lead to memory leaks
- _rspq.put((e, rv))
- msg = meth = args = kwargs = e = rv = None
- _wsock.sendall(_bytetosend)
-
-
-def execute(meth, *args, **kwargs):
- """
- Execute *meth* in a Python thread, blocking the current coroutine/
- greenthread until the method completes.
-
- The primary use case for this is to wrap an object or module that is not
- amenable to monkeypatching or any of the other tricks that Eventlet uses
- to achieve cooperative yielding. With tpool, you can force such objects to
- cooperate with green threads by sticking them in native threads, at the cost
- of some overhead.
- """
- setup()
- # if already in tpool, don't recurse into the tpool
- # also, call functions directly if we're inside an import lock, because
- # if meth does any importing (sadly common), it will hang
- my_thread = threading.currentThread()
- if my_thread in _threads or imp.lock_held() or _nthreads == 0:
- return meth(*args, **kwargs)
-
- e = event.Event()
- _reqq.put((e, meth, args, kwargs))
-
- rv = e.wait()
- if isinstance(rv, tuple) \
- and len(rv) == 3 \
- and isinstance(rv[1], EXC_CLASSES):
- (c, e, tb) = rv
- if not QUIET:
- traceback.print_exception(c, e, tb)
- traceback.print_stack()
- six.reraise(c, e, tb)
- return rv
-
-
-def proxy_call(autowrap, f, *args, **kwargs):
- """
- Call a function *f* and returns the value. If the type of the return value
- is in the *autowrap* collection, then it is wrapped in a :class:`Proxy`
- object before return.
-
- Normally *f* will be called in the threadpool with :func:`execute`; if the
- keyword argument "nonblocking" is set to ``True``, it will simply be
- executed directly. This is useful if you have an object which has methods
- that don't need to be called in a separate thread, but which return objects
- that should be Proxy wrapped.
- """
- if kwargs.pop('nonblocking', False):
- rv = f(*args, **kwargs)
- else:
- rv = execute(f, *args, **kwargs)
- if isinstance(rv, autowrap):
- return Proxy(rv, autowrap)
- else:
- return rv
-
-
-class Proxy(object):
- """
- a simple proxy-wrapper of any object that comes with a
- methods-only interface, in order to forward every method
- invocation onto a thread in the native-thread pool. A key
- restriction is that the object's methods should not switch
- greenlets or use Eventlet primitives, since they are in a
- different thread from the main hub, and therefore might behave
- unexpectedly. This is for running native-threaded code
- only.
-
- It's common to want to have some of the attributes or return
- values also wrapped in Proxy objects (for example, database
- connection objects produce cursor objects which also should be
- wrapped in Proxy objects to remain nonblocking). *autowrap*, if
- supplied, is a collection of types; if an attribute or return
- value matches one of those types (via isinstance), it will be
- wrapped in a Proxy. *autowrap_names* is a collection
- of strings, which represent the names of attributes that should be
- wrapped in Proxy objects when accessed.
- """
-
- def __init__(self, obj, autowrap=(), autowrap_names=()):
- self._obj = obj
- self._autowrap = autowrap
- self._autowrap_names = autowrap_names
-
- def __getattr__(self, attr_name):
- f = getattr(self._obj, attr_name)
- if not hasattr(f, '__call__'):
- if isinstance(f, self._autowrap) or attr_name in self._autowrap_names:
- return Proxy(f, self._autowrap)
- return f
-
- def doit(*args, **kwargs):
- result = proxy_call(self._autowrap, f, *args, **kwargs)
- if attr_name in self._autowrap_names and not isinstance(result, Proxy):
- return Proxy(result)
- return result
- return doit
-
- # the following are a buncha methods that the python interpeter
- # doesn't use getattr to retrieve and therefore have to be defined
- # explicitly
- def __getitem__(self, key):
- return proxy_call(self._autowrap, self._obj.__getitem__, key)
-
- def __setitem__(self, key, value):
- return proxy_call(self._autowrap, self._obj.__setitem__, key, value)
-
- def __deepcopy__(self, memo=None):
- return proxy_call(self._autowrap, self._obj.__deepcopy__, memo)
-
- def __copy__(self, memo=None):
- return proxy_call(self._autowrap, self._obj.__copy__, memo)
-
- def __call__(self, *a, **kw):
- if '__call__' in self._autowrap_names:
- return Proxy(proxy_call(self._autowrap, self._obj, *a, **kw))
- else:
- return proxy_call(self._autowrap, self._obj, *a, **kw)
-
- def __enter__(self):
- return proxy_call(self._autowrap, self._obj.__enter__)
-
- def __exit__(self, *exc):
- return proxy_call(self._autowrap, self._obj.__exit__, *exc)
-
- # these don't go through a proxy call, because they're likely to
- # be called often, and are unlikely to be implemented on the
- # wrapped object in such a way that they would block
- def __eq__(self, rhs):
- return self._obj == rhs
-
- def __hash__(self):
- return self._obj.__hash__()
-
- def __repr__(self):
- return self._obj.__repr__()
-
- def __str__(self):
- return self._obj.__str__()
-
- def __len__(self):
- return len(self._obj)
-
- def __nonzero__(self):
- return bool(self._obj)
- # Python3
- __bool__ = __nonzero__
-
- def __iter__(self):
- it = iter(self._obj)
- if it == self._obj:
- return self
- else:
- return Proxy(it)
-
- def next(self):
- return proxy_call(self._autowrap, next, self._obj)
- # Python3
- __next__ = next
-
-
-def setup():
- global _rsock, _wsock, _threads, _coro, _setup_already, _rspq, _reqq
- if _setup_already:
- return
- else:
- _setup_already = True
-
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.bind(('127.0.0.1', 0))
- sock.listen(1)
- csock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- csock.connect(sock.getsockname())
- _wsock, _addr = sock.accept()
- sock.close()
- _rsock = greenio.GreenSocket(csock)
-
- _reqq = Queue(maxsize=-1)
- _rspq = Queue(maxsize=-1)
- assert _nthreads >= 0, "Can't specify negative number of threads"
- if _nthreads == 0:
- import warnings
- warnings.warn("Zero threads in tpool. All tpool.execute calls will\
- execute in main thread. Check the value of the environment \
- variable EVENTLET_THREADPOOL_SIZE.", RuntimeWarning)
- for i in six.moves.range(_nthreads):
- t = threading.Thread(target=tworker,
- name="tpool_thread_%s" % i)
- t.setDaemon(True)
- t.start()
- _threads.append(t)
-
- _coro = greenthread.spawn_n(tpool_trampoline)
-
-
-def killall():
- global _setup_already, _rspq, _rsock, _wsock
- if not _setup_already:
- return
- for thr in _threads:
- _reqq.put(None)
- for thr in _threads:
- thr.join()
- del _threads[:]
- if _coro is not None:
- greenthread.kill(_coro)
- _rsock.close()
- _wsock.close()
- _rsock = None
- _wsock = None
- _rspq = None
- _setup_already = False
-
-
-def set_num_threads(nthreads):
- global _nthreads
- _nthreads = nthreads
Code Review / packages / trusty / python-eventlet.git / blobdiff