tpool.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340
  1. # Copyright (c) 2007-2009, Linden Research, Inc.
  2. # Copyright (c) 2007, IBM Corp.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import atexit
  16. import imp
  17. import os
  18. import sys
  19. import traceback
  20. import eventlet
  21. from eventlet import event, greenio, greenthread, patcher, timeout
  22. import six
  23. __all__ = ['execute', 'Proxy', 'killall', 'set_num_threads']
  24. EXC_CLASSES = (Exception, timeout.Timeout)
  25. SYS_EXCS = (GeneratorExit, KeyboardInterrupt, SystemExit)
  26. QUIET = True
  27. socket = patcher.original('socket')
  28. threading = patcher.original('threading')
  29. if six.PY2:
  30. Queue_module = patcher.original('Queue')
  31. if six.PY3:
  32. Queue_module = patcher.original('queue')
  33. Empty = Queue_module.Empty
  34. Queue = Queue_module.Queue
  35. _bytetosend = b' '
  36. _coro = None
  37. _nthreads = int(os.environ.get('EVENTLET_THREADPOOL_SIZE', 20))
  38. _reqq = _rspq = None
  39. _rsock = _wsock = None
  40. _setup_already = False
  41. _threads = []
  42. def tpool_trampoline():
  43. global _rspq
  44. while True:
  45. try:
  46. _c = _rsock.recv(1)
  47. assert _c
  48. # FIXME: this is probably redundant since using sockets instead of pipe now
  49. except ValueError:
  50. break # will be raised when pipe is closed
  51. while not _rspq.empty():
  52. try:
  53. (e, rv) = _rspq.get(block=False)
  54. e.send(rv)
  55. e = rv = None
  56. except Empty:
  57. pass
  58. def tworker():
  59. global _rspq
  60. while True:
  61. try:
  62. msg = _reqq.get()
  63. except AttributeError:
  64. return # can't get anything off of a dud queue
  65. if msg is None:
  66. return
  67. (e, meth, args, kwargs) = msg
  68. rv = None
  69. try:
  70. rv = meth(*args, **kwargs)
  71. except SYS_EXCS:
  72. raise
  73. except EXC_CLASSES:
  74. rv = sys.exc_info()
  75. if sys.version_info >= (3, 4):
  76. traceback.clear_frames(rv[1].__traceback__)
  77. if six.PY2:
  78. sys.exc_clear()
  79. # test_leakage_from_tracebacks verifies that the use of
  80. # exc_info does not lead to memory leaks
  81. _rspq.put((e, rv))
  82. msg = meth = args = kwargs = e = rv = None
  83. _wsock.sendall(_bytetosend)
  84. def execute(meth, *args, **kwargs):
  85. """
  86. Execute *meth* in a Python thread, blocking the current coroutine/
  87. greenthread until the method completes.
  88. The primary use case for this is to wrap an object or module that is not
  89. amenable to monkeypatching or any of the other tricks that Eventlet uses
  90. to achieve cooperative yielding. With tpool, you can force such objects to
  91. cooperate with green threads by sticking them in native threads, at the cost
  92. of some overhead.
  93. """
  94. setup()
  95. # if already in tpool, don't recurse into the tpool
  96. # also, call functions directly if we're inside an import lock, because
  97. # if meth does any importing (sadly common), it will hang
  98. my_thread = threading.currentThread()
  99. if my_thread in _threads or imp.lock_held() or _nthreads == 0:
  100. return meth(*args, **kwargs)
  101. e = event.Event()
  102. _reqq.put((e, meth, args, kwargs))
  103. rv = e.wait()
  104. if isinstance(rv, tuple) \
  105. and len(rv) == 3 \
  106. and isinstance(rv[1], EXC_CLASSES):
  107. (c, e, tb) = rv
  108. if not QUIET:
  109. traceback.print_exception(c, e, tb)
  110. traceback.print_stack()
  111. six.reraise(c, e, tb)
  112. return rv
  113. def proxy_call(autowrap, f, *args, **kwargs):
  114. """
  115. Call a function *f* and returns the value. If the type of the return value
  116. is in the *autowrap* collection, then it is wrapped in a :class:`Proxy`
  117. object before return.
  118. Normally *f* will be called in the threadpool with :func:`execute`; if the
  119. keyword argument "nonblocking" is set to ``True``, it will simply be
  120. executed directly. This is useful if you have an object which has methods
  121. that don't need to be called in a separate thread, but which return objects
  122. that should be Proxy wrapped.
  123. """
  124. if kwargs.pop('nonblocking', False):
  125. rv = f(*args, **kwargs)
  126. else:
  127. rv = execute(f, *args, **kwargs)
  128. if isinstance(rv, autowrap):
  129. return Proxy(rv, autowrap)
  130. else:
  131. return rv
  132. class Proxy(object):
  133. """
  134. a simple proxy-wrapper of any object that comes with a
  135. methods-only interface, in order to forward every method
  136. invocation onto a thread in the native-thread pool. A key
  137. restriction is that the object's methods should not switch
  138. greenlets or use Eventlet primitives, since they are in a
  139. different thread from the main hub, and therefore might behave
  140. unexpectedly. This is for running native-threaded code
  141. only.
  142. It's common to want to have some of the attributes or return
  143. values also wrapped in Proxy objects (for example, database
  144. connection objects produce cursor objects which also should be
  145. wrapped in Proxy objects to remain nonblocking). *autowrap*, if
  146. supplied, is a collection of types; if an attribute or return
  147. value matches one of those types (via isinstance), it will be
  148. wrapped in a Proxy. *autowrap_names* is a collection
  149. of strings, which represent the names of attributes that should be
  150. wrapped in Proxy objects when accessed.
  151. """
  152. def __init__(self, obj, autowrap=(), autowrap_names=()):
  153. self._obj = obj
  154. self._autowrap = autowrap
  155. self._autowrap_names = autowrap_names
  156. def __getattr__(self, attr_name):
  157. f = getattr(self._obj, attr_name)
  158. if not hasattr(f, '__call__'):
  159. if isinstance(f, self._autowrap) or attr_name in self._autowrap_names:
  160. return Proxy(f, self._autowrap)
  161. return f
  162. def doit(*args, **kwargs):
  163. result = proxy_call(self._autowrap, f, *args, **kwargs)
  164. if attr_name in self._autowrap_names and not isinstance(result, Proxy):
  165. return Proxy(result)
  166. return result
  167. return doit
  168. # the following are a buncha methods that the python interpeter
  169. # doesn't use getattr to retrieve and therefore have to be defined
  170. # explicitly
  171. def __getitem__(self, key):
  172. return proxy_call(self._autowrap, self._obj.__getitem__, key)
  173. def __setitem__(self, key, value):
  174. return proxy_call(self._autowrap, self._obj.__setitem__, key, value)
  175. def __deepcopy__(self, memo=None):
  176. return proxy_call(self._autowrap, self._obj.__deepcopy__, memo)
  177. def __copy__(self, memo=None):
  178. return proxy_call(self._autowrap, self._obj.__copy__, memo)
  179. def __call__(self, *a, **kw):
  180. if '__call__' in self._autowrap_names:
  181. return Proxy(proxy_call(self._autowrap, self._obj, *a, **kw))
  182. else:
  183. return proxy_call(self._autowrap, self._obj, *a, **kw)
  184. def __enter__(self):
  185. return proxy_call(self._autowrap, self._obj.__enter__)
  186. def __exit__(self, *exc):
  187. return proxy_call(self._autowrap, self._obj.__exit__, *exc)
  188. # these don't go through a proxy call, because they're likely to
  189. # be called often, and are unlikely to be implemented on the
  190. # wrapped object in such a way that they would block
  191. def __eq__(self, rhs):
  192. return self._obj == rhs
  193. def __hash__(self):
  194. return self._obj.__hash__()
  195. def __repr__(self):
  196. return self._obj.__repr__()
  197. def __str__(self):
  198. return self._obj.__str__()
  199. def __len__(self):
  200. return len(self._obj)
  201. def __nonzero__(self):
  202. return bool(self._obj)
  203. # Python3
  204. __bool__ = __nonzero__
  205. def __iter__(self):
  206. it = iter(self._obj)
  207. if it == self._obj:
  208. return self
  209. else:
  210. return Proxy(it)
  211. def next(self):
  212. return proxy_call(self._autowrap, next, self._obj)
  213. # Python3
  214. __next__ = next
  215. def setup():
  216. global _rsock, _wsock, _coro, _setup_already, _rspq, _reqq
  217. if _setup_already:
  218. return
  219. else:
  220. _setup_already = True
  221. assert _nthreads >= 0, "Can't specify negative number of threads"
  222. if _nthreads == 0:
  223. import warnings
  224. warnings.warn("Zero threads in tpool. All tpool.execute calls will\
  225. execute in main thread. Check the value of the environment \
  226. variable EVENTLET_THREADPOOL_SIZE.", RuntimeWarning)
  227. _reqq = Queue(maxsize=-1)
  228. _rspq = Queue(maxsize=-1)
  229. # connected socket pair
  230. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  231. sock.bind(('127.0.0.1', 0))
  232. sock.listen(1)
  233. csock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  234. csock.connect(sock.getsockname())
  235. csock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
  236. _wsock, _addr = sock.accept()
  237. _wsock.settimeout(None)
  238. _wsock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
  239. sock.close()
  240. _rsock = greenio.GreenSocket(csock)
  241. _rsock.settimeout(None)
  242. for i in six.moves.range(_nthreads):
  243. t = threading.Thread(target=tworker,
  244. name="tpool_thread_%s" % i)
  245. t.setDaemon(True)
  246. t.start()
  247. _threads.append(t)
  248. _coro = greenthread.spawn_n(tpool_trampoline)
  249. # This yield fixes subtle error with GreenSocket.__del__
  250. eventlet.sleep(0)
  251. # Avoid ResourceWarning unclosed socket on Python3.2+
  252. @atexit.register
  253. def killall():
  254. global _setup_already, _rspq, _rsock, _wsock
  255. if not _setup_already:
  256. return
  257. # This yield fixes freeze in some scenarios
  258. eventlet.sleep(0)
  259. for thr in _threads:
  260. _reqq.put(None)
  261. for thr in _threads:
  262. thr.join()
  263. del _threads[:]
  264. # return any remaining results
  265. while (_rspq is not None) and not _rspq.empty():
  266. try:
  267. (e, rv) = _rspq.get(block=False)
  268. e.send(rv)
  269. e = rv = None
  270. except Empty:
  271. pass
  272. if _coro is not None:
  273. greenthread.kill(_coro)
  274. if _rsock is not None:
  275. _rsock.close()
  276. _rsock = None
  277. if _wsock is not None:
  278. _wsock.close()
  279. _wsock = None
  280. _rspq = None
  281. _setup_already = False
  282. def set_num_threads(nthreads):
  283. global _nthreads
  284. _nthreads = nthreads