tpool.py 9.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295
  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 imp
  16. import os
  17. import sys
  18. from eventlet import event
  19. from eventlet import greenio
  20. from eventlet import greenthread
  21. from eventlet import patcher
  22. threading = patcher.original('threading')
  23. Queue_module = patcher.original('Queue')
  24. Queue = Queue_module.Queue
  25. Empty = Queue_module.Empty
  26. __all__ = ['execute', 'Proxy', 'killall']
  27. QUIET=True
  28. _rfile = _wfile = None
  29. _bytetosend = ' '.encode()
  30. def _signal_t2e():
  31. _wfile.write(_bytetosend)
  32. _wfile.flush()
  33. _rspq = None
  34. def tpool_trampoline():
  35. global _rspq
  36. while(True):
  37. try:
  38. _c = _rfile.read(1)
  39. assert _c
  40. except ValueError:
  41. break # will be raised when pipe is closed
  42. while not _rspq.empty():
  43. try:
  44. (e,rv) = _rspq.get(block=False)
  45. e.send(rv)
  46. rv = None
  47. except Empty:
  48. pass
  49. SYS_EXCS = (KeyboardInterrupt, SystemExit)
  50. def tworker(reqq):
  51. global _rspq
  52. while(True):
  53. try:
  54. msg = reqq.get()
  55. except AttributeError:
  56. return # can't get anything off of a dud queue
  57. if msg is None:
  58. return
  59. (e,meth,args,kwargs) = msg
  60. rv = None
  61. try:
  62. rv = meth(*args,**kwargs)
  63. except SYS_EXCS:
  64. raise
  65. except Exception:
  66. rv = sys.exc_info()
  67. # test_leakage_from_tracebacks verifies that the use of
  68. # exc_info does not lead to memory leaks
  69. _rspq.put((e,rv))
  70. meth = args = kwargs = e = rv = None
  71. _signal_t2e()
  72. def execute(meth,*args, **kwargs):
  73. """
  74. Execute *meth* in a Python thread, blocking the current coroutine/
  75. greenthread until the method completes.
  76. The primary use case for this is to wrap an object or module that is not
  77. amenable to monkeypatching or any of the other tricks that Eventlet uses
  78. to achieve cooperative yielding. With tpool, you can force such objects to
  79. cooperate with green threads by sticking them in native threads, at the cost
  80. of some overhead.
  81. """
  82. setup()
  83. # if already in tpool, don't recurse into the tpool
  84. # also, call functions directly if we're inside an import lock, because
  85. # if meth does any importing (sadly common), it will hang
  86. my_thread = threading.currentThread()
  87. if my_thread in _threads or imp.lock_held() or _nthreads == 0:
  88. return meth(*args, **kwargs)
  89. cur = greenthread.getcurrent()
  90. # a mini mixing function to make up for the fact that hash(greenlet) doesn't
  91. # have much variability in the lower bits
  92. k = hash(cur)
  93. k = k + 0x2c865fd + (k >> 5)
  94. k = k ^ 0xc84d1b7 ^ (k >> 7)
  95. thread_index = k % _nthreads
  96. reqq, _thread = _threads[thread_index]
  97. e = event.Event()
  98. reqq.put((e,meth,args,kwargs))
  99. rv = e.wait()
  100. if isinstance(rv,tuple) and len(rv) == 3 and isinstance(rv[1],Exception):
  101. import traceback
  102. (c,e,tb) = rv
  103. if not QUIET:
  104. traceback.print_exception(c,e,tb)
  105. traceback.print_stack()
  106. raise c,e,tb
  107. return rv
  108. def proxy_call(autowrap, f, *args, **kwargs):
  109. """
  110. Call a function *f* and returns the value. If the type of the return value
  111. is in the *autowrap* collection, then it is wrapped in a :class:`Proxy`
  112. object before return.
  113. Normally *f* will be called in the threadpool with :func:`execute`; if the
  114. keyword argument "nonblocking" is set to ``True``, it will simply be
  115. executed directly. This is useful if you have an object which has methods
  116. that don't need to be called in a separate thread, but which return objects
  117. that should be Proxy wrapped.
  118. """
  119. if kwargs.pop('nonblocking',False):
  120. rv = f(*args, **kwargs)
  121. else:
  122. rv = execute(f,*args,**kwargs)
  123. if isinstance(rv, autowrap):
  124. return Proxy(rv, autowrap)
  125. else:
  126. return rv
  127. class Proxy(object):
  128. """
  129. a simple proxy-wrapper of any object that comes with a
  130. methods-only interface, in order to forward every method
  131. invocation onto a thread in the native-thread pool. A key
  132. restriction is that the object's methods should not switch
  133. greenlets or use Eventlet primitives, since they are in a
  134. different thread from the main hub, and therefore might behave
  135. unexpectedly. This is for running native-threaded code
  136. only.
  137. It's common to want to have some of the attributes or return
  138. values also wrapped in Proxy objects (for example, database
  139. connection objects produce cursor objects which also should be
  140. wrapped in Proxy objects to remain nonblocking). *autowrap*, if
  141. supplied, is a collection of types; if an attribute or return
  142. value matches one of those types (via isinstance), it will be
  143. wrapped in a Proxy. *autowrap_names* is a collection
  144. of strings, which represent the names of attributes that should be
  145. wrapped in Proxy objects when accessed.
  146. """
  147. def __init__(self, obj,autowrap=(), autowrap_names=()):
  148. self._obj = obj
  149. self._autowrap = autowrap
  150. self._autowrap_names = autowrap_names
  151. def __getattr__(self,attr_name):
  152. f = getattr(self._obj,attr_name)
  153. if not hasattr(f, '__call__'):
  154. if (isinstance(f, self._autowrap) or
  155. attr_name in self._autowrap_names):
  156. return Proxy(f, self._autowrap)
  157. return f
  158. def doit(*args, **kwargs):
  159. result = proxy_call(self._autowrap, f, *args, **kwargs)
  160. if attr_name in self._autowrap_names and not isinstance(result, Proxy):
  161. return Proxy(result)
  162. return result
  163. return doit
  164. # the following are a buncha methods that the python interpeter
  165. # doesn't use getattr to retrieve and therefore have to be defined
  166. # explicitly
  167. def __getitem__(self, key):
  168. return proxy_call(self._autowrap, self._obj.__getitem__, key)
  169. def __setitem__(self, key, value):
  170. return proxy_call(self._autowrap, self._obj.__setitem__, key, value)
  171. def __deepcopy__(self, memo=None):
  172. return proxy_call(self._autowrap, self._obj.__deepcopy__, memo)
  173. def __copy__(self, memo=None):
  174. return proxy_call(self._autowrap, self._obj.__copy__, memo)
  175. def __call__(self, *a, **kw):
  176. if '__call__' in self._autowrap_names:
  177. return Proxy(proxy_call(self._autowrap, self._obj, *a, **kw))
  178. else:
  179. return proxy_call(self._autowrap, self._obj, *a, **kw)
  180. # these don't go through a proxy call, because they're likely to
  181. # be called often, and are unlikely to be implemented on the
  182. # wrapped object in such a way that they would block
  183. def __eq__(self, rhs):
  184. return self._obj == rhs
  185. def __hash__(self):
  186. return self._obj.__hash__()
  187. def __repr__(self):
  188. return self._obj.__repr__()
  189. def __str__(self):
  190. return self._obj.__str__()
  191. def __len__(self):
  192. return len(self._obj)
  193. def __nonzero__(self):
  194. return bool(self._obj)
  195. def __iter__(self):
  196. it = iter(self._obj)
  197. if it == self._obj:
  198. return self
  199. else:
  200. return Proxy(it)
  201. def next(self):
  202. return proxy_call(self._autowrap, self._obj.next)
  203. _nthreads = int(os.environ.get('EVENTLET_THREADPOOL_SIZE', 20))
  204. _threads = []
  205. _coro = None
  206. _setup_already = False
  207. def setup():
  208. global _rfile, _wfile, _threads, _coro, _setup_already, _rspq
  209. if _setup_already:
  210. return
  211. else:
  212. _setup_already = True
  213. try:
  214. _rpipe, _wpipe = os.pipe()
  215. _wfile = greenio.GreenPipe(_wpipe, 'wb', 0)
  216. _rfile = greenio.GreenPipe(_rpipe, 'rb', 0)
  217. except (ImportError, NotImplementedError):
  218. # This is Windows compatibility -- use a socket instead of a pipe because
  219. # pipes don't really exist on Windows.
  220. import socket
  221. from eventlet import util
  222. sock = util.__original_socket__(socket.AF_INET, socket.SOCK_STREAM)
  223. sock.bind(('localhost', 0))
  224. sock.listen(50)
  225. csock = util.__original_socket__(socket.AF_INET, socket.SOCK_STREAM)
  226. csock.connect(('localhost', sock.getsockname()[1]))
  227. nsock, addr = sock.accept()
  228. _rfile = greenio.GreenSocket(csock).makefile('rb', 0)
  229. _wfile = nsock.makefile('wb',0)
  230. _rspq = Queue(maxsize=-1)
  231. assert _nthreads >= 0, "Can't specify negative number of threads"
  232. if _nthreads == 0:
  233. import warnings
  234. warnings.warn("Zero threads in tpool. All tpool.execute calls will\
  235. execute in main thread. Check the value of the environment \
  236. variable EVENTLET_THREADPOOL_SIZE.", RuntimeWarning)
  237. for i in xrange(_nthreads):
  238. reqq = Queue(maxsize=-1)
  239. t = threading.Thread(target=tworker,
  240. name="tpool_thread_%s" % i,
  241. args=(reqq,))
  242. t.setDaemon(True)
  243. t.start()
  244. _threads.append((reqq, t))
  245. _coro = greenthread.spawn_n(tpool_trampoline)
  246. def killall():
  247. global _setup_already, _rspq, _rfile, _wfile
  248. if not _setup_already:
  249. return
  250. for reqq, _ in _threads:
  251. reqq.put(None)
  252. for _, thr in _threads:
  253. thr.join()
  254. del _threads[:]
  255. if _coro is not None:
  256. greenthread.kill(_coro)
  257. _rfile.close()
  258. _wfile.close()
  259. _rfile = None
  260. _wfile = None
  261. _rspq = None
  262. _setup_already = False