managers.py 36 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210
  1. #
  2. # Module providing the `SyncManager` class for dealing
  3. # with shared objects
  4. #
  5. # multiprocessing/managers.py
  6. #
  7. # Copyright (c) 2006-2008, R Oudkerk
  8. # Licensed to PSF under a Contributor Agreement.
  9. #
  10. from __future__ import absolute_import
  11. #
  12. # Imports
  13. #
  14. import sys
  15. import threading
  16. import array
  17. from traceback import format_exc
  18. from . import connection
  19. from . import context
  20. from . import pool
  21. from . import process
  22. from . import reduction
  23. from . import util
  24. from . import get_context
  25. from .five import Queue, items, monotonic
  26. __all__ = ['BaseManager', 'SyncManager', 'BaseProxy', 'Token']
  27. PY3 = sys.version_info[0] == 3
  28. #
  29. # Register some things for pickling
  30. #
  31. if PY3:
  32. def reduce_array(a):
  33. return array.array, (a.typecode, a.tobytes())
  34. else:
  35. def reduce_array(a): # noqa
  36. return array.array, (a.typecode, a.tostring())
  37. reduction.register(array.array, reduce_array)
  38. view_types = [type(getattr({}, name)())
  39. for name in ('items', 'keys', 'values')]
  40. if view_types[0] is not list: # only needed in Py3.0
  41. def rebuild_as_list(obj):
  42. return list, (list(obj), )
  43. for view_type in view_types:
  44. reduction.register(view_type, rebuild_as_list)
  45. #
  46. # Type for identifying shared objects
  47. #
  48. class Token(object):
  49. '''
  50. Type to uniquely indentify a shared object
  51. '''
  52. __slots__ = ('typeid', 'address', 'id')
  53. def __init__(self, typeid, address, id):
  54. (self.typeid, self.address, self.id) = (typeid, address, id)
  55. def __getstate__(self):
  56. return (self.typeid, self.address, self.id)
  57. def __setstate__(self, state):
  58. (self.typeid, self.address, self.id) = state
  59. def __repr__(self):
  60. return '%s(typeid=%r, address=%r, id=%r)' % \
  61. (self.__class__.__name__, self.typeid, self.address, self.id)
  62. #
  63. # Function for communication with a manager's server process
  64. #
  65. def dispatch(c, id, methodname, args=(), kwds={}):
  66. '''
  67. Send a message to manager using connection `c` and return response
  68. '''
  69. c.send((id, methodname, args, kwds))
  70. kind, result = c.recv()
  71. if kind == '#RETURN':
  72. return result
  73. raise convert_to_error(kind, result)
  74. def convert_to_error(kind, result):
  75. if kind == '#ERROR':
  76. return result
  77. elif kind == '#TRACEBACK':
  78. assert type(result) is str
  79. return RemoteError(result)
  80. elif kind == '#UNSERIALIZABLE':
  81. assert type(result) is str
  82. return RemoteError('Unserializable message: %s\n' % result)
  83. else:
  84. return ValueError('Unrecognized message type')
  85. class RemoteError(Exception):
  86. def __str__(self):
  87. return ('\n' + '-' * 75 + '\n' + str(self.args[0]) + '-' * 75)
  88. #
  89. # Functions for finding the method names of an object
  90. #
  91. def all_methods(obj):
  92. '''
  93. Return a list of names of methods of `obj`
  94. '''
  95. temp = []
  96. for name in dir(obj):
  97. func = getattr(obj, name)
  98. if callable(func):
  99. temp.append(name)
  100. return temp
  101. def public_methods(obj):
  102. '''
  103. Return a list of names of methods of `obj` which do not start with '_'
  104. '''
  105. return [name for name in all_methods(obj) if name[0] != '_']
  106. #
  107. # Server which is run in a process controlled by a manager
  108. #
  109. class Server(object):
  110. '''
  111. Server class which runs in a process controlled by a manager object
  112. '''
  113. public = ['shutdown', 'create', 'accept_connection', 'get_methods',
  114. 'debug_info', 'number_of_objects', 'dummy', 'incref', 'decref']
  115. def __init__(self, registry, address, authkey, serializer):
  116. assert isinstance(authkey, bytes)
  117. self.registry = registry
  118. self.authkey = process.AuthenticationString(authkey)
  119. Listener, Client = listener_client[serializer]
  120. # do authentication later
  121. self.listener = Listener(address=address, backlog=16)
  122. self.address = self.listener.address
  123. self.id_to_obj = {'0': (None, ())}
  124. self.id_to_refcount = {}
  125. self.mutex = threading.RLock()
  126. def serve_forever(self):
  127. '''
  128. Run the server forever
  129. '''
  130. self.stop_event = threading.Event()
  131. process.current_process()._manager_server = self
  132. try:
  133. accepter = threading.Thread(target=self.accepter)
  134. accepter.daemon = True
  135. accepter.start()
  136. try:
  137. while not self.stop_event.is_set():
  138. self.stop_event.wait(1)
  139. except (KeyboardInterrupt, SystemExit):
  140. pass
  141. finally:
  142. if sys.stdout != sys.__stdout__:
  143. util.debug('resetting stdout, stderr')
  144. sys.stdout = sys.__stdout__
  145. sys.stderr = sys.__stderr__
  146. sys.exit(0)
  147. def accepter(self):
  148. while True:
  149. try:
  150. c = self.listener.accept()
  151. except OSError:
  152. continue
  153. t = threading.Thread(target=self.handle_request, args=(c, ))
  154. t.daemon = True
  155. t.start()
  156. def handle_request(self, c):
  157. '''
  158. Handle a new connection
  159. '''
  160. funcname = result = request = None
  161. try:
  162. connection.deliver_challenge(c, self.authkey)
  163. connection.answer_challenge(c, self.authkey)
  164. request = c.recv()
  165. ignore, funcname, args, kwds = request
  166. assert funcname in self.public, '%r unrecognized' % funcname
  167. func = getattr(self, funcname)
  168. except Exception:
  169. msg = ('#TRACEBACK', format_exc())
  170. else:
  171. try:
  172. result = func(c, *args, **kwds)
  173. except Exception:
  174. msg = ('#TRACEBACK', format_exc())
  175. else:
  176. msg = ('#RETURN', result)
  177. try:
  178. c.send(msg)
  179. except Exception as exc:
  180. try:
  181. c.send(('#TRACEBACK', format_exc()))
  182. except Exception:
  183. pass
  184. util.info('Failure to send message: %r', msg)
  185. util.info(' ... request was %r', request)
  186. util.info(' ... exception was %r', exc)
  187. c.close()
  188. def serve_client(self, conn):
  189. '''
  190. Handle requests from the proxies in a particular process/thread
  191. '''
  192. util.debug('starting server thread to service %r',
  193. threading.current_thread().name)
  194. recv = conn.recv
  195. send = conn.send
  196. id_to_obj = self.id_to_obj
  197. while not self.stop_event.is_set():
  198. try:
  199. methodname = obj = None
  200. request = recv()
  201. ident, methodname, args, kwds = request
  202. obj, exposed, gettypeid = id_to_obj[ident]
  203. if methodname not in exposed:
  204. raise AttributeError(
  205. 'method %r of %r object is not in exposed=%r' % (
  206. methodname, type(obj), exposed)
  207. )
  208. function = getattr(obj, methodname)
  209. try:
  210. res = function(*args, **kwds)
  211. except Exception as exc:
  212. msg = ('#ERROR', exc)
  213. else:
  214. typeid = gettypeid and gettypeid.get(methodname, None)
  215. if typeid:
  216. rident, rexposed = self.create(conn, typeid, res)
  217. token = Token(typeid, self.address, rident)
  218. msg = ('#PROXY', (rexposed, token))
  219. else:
  220. msg = ('#RETURN', res)
  221. except AttributeError:
  222. if methodname is None:
  223. msg = ('#TRACEBACK', format_exc())
  224. else:
  225. try:
  226. fallback_func = self.fallback_mapping[methodname]
  227. result = fallback_func(
  228. self, conn, ident, obj, *args, **kwds
  229. )
  230. msg = ('#RETURN', result)
  231. except Exception:
  232. msg = ('#TRACEBACK', format_exc())
  233. except EOFError:
  234. util.debug('got EOF -- exiting thread serving %r',
  235. threading.current_thread().name)
  236. sys.exit(0)
  237. except Exception:
  238. msg = ('#TRACEBACK', format_exc())
  239. try:
  240. try:
  241. send(msg)
  242. except Exception:
  243. send(('#UNSERIALIZABLE', repr(msg)))
  244. except Exception as exc:
  245. util.info('exception in thread serving %r',
  246. threading.current_thread().name)
  247. util.info(' ... message was %r', msg)
  248. util.info(' ... exception was %r', exc)
  249. conn.close()
  250. sys.exit(1)
  251. def fallback_getvalue(self, conn, ident, obj):
  252. return obj
  253. def fallback_str(self, conn, ident, obj):
  254. return str(obj)
  255. def fallback_repr(self, conn, ident, obj):
  256. return repr(obj)
  257. fallback_mapping = {
  258. '__str__': fallback_str,
  259. '__repr__': fallback_repr,
  260. '#GETVALUE': fallback_getvalue,
  261. }
  262. def dummy(self, c):
  263. pass
  264. def debug_info(self, c):
  265. '''
  266. Return some info --- useful to spot problems with refcounting
  267. '''
  268. with self.mutex:
  269. result = []
  270. keys = list(self.id_to_obj.keys())
  271. keys.sort()
  272. for ident in keys:
  273. if ident != '0':
  274. result.append(' %s: refcount=%s\n %s' %
  275. (ident, self.id_to_refcount[ident],
  276. str(self.id_to_obj[ident][0])[:75]))
  277. return '\n'.join(result)
  278. def number_of_objects(self, c):
  279. '''
  280. Number of shared objects
  281. '''
  282. return len(self.id_to_obj) - 1 # don't count ident='0'
  283. def shutdown(self, c):
  284. '''
  285. Shutdown this process
  286. '''
  287. try:
  288. util.debug('Manager received shutdown message')
  289. c.send(('#RETURN', None))
  290. except:
  291. import traceback
  292. traceback.print_exc()
  293. finally:
  294. self.stop_event.set()
  295. def create(self, c, typeid, *args, **kwds):
  296. '''
  297. Create a new shared object and return its id
  298. '''
  299. with self.mutex:
  300. callable, exposed, method_to_typeid, proxytype = \
  301. self.registry[typeid]
  302. if callable is None:
  303. assert len(args) == 1 and not kwds
  304. obj = args[0]
  305. else:
  306. obj = callable(*args, **kwds)
  307. if exposed is None:
  308. exposed = public_methods(obj)
  309. if method_to_typeid is not None:
  310. assert type(method_to_typeid) is dict
  311. exposed = list(exposed) + list(method_to_typeid)
  312. # convert to string because xmlrpclib
  313. # only has 32 bit signed integers
  314. ident = '%x' % id(obj)
  315. util.debug('%r callable returned object with id %r', typeid, ident)
  316. self.id_to_obj[ident] = (obj, set(exposed), method_to_typeid)
  317. if ident not in self.id_to_refcount:
  318. self.id_to_refcount[ident] = 0
  319. # increment the reference count immediately, to avoid
  320. # this object being garbage collected before a Proxy
  321. # object for it can be created. The caller of create()
  322. # is responsible for doing a decref once the Proxy object
  323. # has been created.
  324. self.incref(c, ident)
  325. return ident, tuple(exposed)
  326. def get_methods(self, c, token):
  327. '''
  328. Return the methods of the shared object indicated by token
  329. '''
  330. return tuple(self.id_to_obj[token.id][1])
  331. def accept_connection(self, c, name):
  332. '''
  333. Spawn a new thread to serve this connection
  334. '''
  335. threading.current_thread().name = name
  336. c.send(('#RETURN', None))
  337. self.serve_client(c)
  338. def incref(self, c, ident):
  339. with self.mutex:
  340. self.id_to_refcount[ident] += 1
  341. def decref(self, c, ident):
  342. with self.mutex:
  343. assert self.id_to_refcount[ident] >= 1
  344. self.id_to_refcount[ident] -= 1
  345. if self.id_to_refcount[ident] == 0:
  346. del self.id_to_obj[ident], self.id_to_refcount[ident]
  347. util.debug('disposing of obj with id %r', ident)
  348. #
  349. # Class to represent state of a manager
  350. #
  351. class State(object):
  352. __slots__ = ['value']
  353. INITIAL = 0
  354. STARTED = 1
  355. SHUTDOWN = 2
  356. #
  357. # Mapping from serializer name to Listener and Client types
  358. #
  359. listener_client = {
  360. 'pickle': (connection.Listener, connection.Client),
  361. 'xmlrpclib': (connection.XmlListener, connection.XmlClient),
  362. }
  363. #
  364. # Definition of BaseManager
  365. #
  366. class BaseManager(object):
  367. '''
  368. Base class for managers
  369. '''
  370. _registry = {}
  371. _Server = Server
  372. def __init__(self, address=None, authkey=None, serializer='pickle',
  373. ctx=None):
  374. if authkey is None:
  375. authkey = process.current_process().authkey
  376. self._address = address # XXX not final address if eg ('', 0)
  377. self._authkey = process.AuthenticationString(authkey)
  378. self._state = State()
  379. self._state.value = State.INITIAL
  380. self._serializer = serializer
  381. self._Listener, self._Client = listener_client[serializer]
  382. self._ctx = ctx or get_context()
  383. def __reduce__(self):
  384. return (type(self).from_address,
  385. (self._address, self._authkey, self._serializer))
  386. def get_server(self):
  387. '''
  388. Return server object with serve_forever() method and address attribute
  389. '''
  390. assert self._state.value == State.INITIAL
  391. return Server(self._registry, self._address,
  392. self._authkey, self._serializer)
  393. def connect(self):
  394. '''
  395. Connect manager object to the server process
  396. '''
  397. Listener, Client = listener_client[self._serializer]
  398. conn = Client(self._address, authkey=self._authkey)
  399. dispatch(conn, None, 'dummy')
  400. self._state.value = State.STARTED
  401. def start(self, initializer=None, initargs=()):
  402. '''
  403. Spawn a server process for this manager object
  404. '''
  405. assert self._state.value == State.INITIAL
  406. if initializer is not None and not callable(initializer):
  407. raise TypeError('initializer must be a callable')
  408. # pipe over which we will retrieve address of server
  409. reader, writer = connection.Pipe(duplex=False)
  410. # spawn process which runs a server
  411. self._process = self._ctx.Process(
  412. target=type(self)._run_server,
  413. args=(self._registry, self._address, self._authkey,
  414. self._serializer, writer, initializer, initargs),
  415. )
  416. ident = ':'.join(str(i) for i in self._process._identity)
  417. self._process.name = type(self).__name__ + '-' + ident
  418. self._process.start()
  419. # get address of server
  420. writer.close()
  421. self._address = reader.recv()
  422. reader.close()
  423. # register a finalizer
  424. self._state.value = State.STARTED
  425. self.shutdown = util.Finalize(
  426. self, type(self)._finalize_manager,
  427. args=(self._process, self._address, self._authkey,
  428. self._state, self._Client),
  429. exitpriority=0
  430. )
  431. @classmethod
  432. def _run_server(cls, registry, address, authkey, serializer, writer,
  433. initializer=None, initargs=()):
  434. '''
  435. Create a server, report its address and run it
  436. '''
  437. if initializer is not None:
  438. initializer(*initargs)
  439. # create server
  440. server = cls._Server(registry, address, authkey, serializer)
  441. # inform parent process of the server's address
  442. writer.send(server.address)
  443. writer.close()
  444. # run the manager
  445. util.info('manager serving at %r', server.address)
  446. server.serve_forever()
  447. def _create(self, typeid, *args, **kwds):
  448. '''
  449. Create a new shared object; return the token and exposed tuple
  450. '''
  451. assert self._state.value == State.STARTED, 'server not yet started'
  452. conn = self._Client(self._address, authkey=self._authkey)
  453. try:
  454. id, exposed = dispatch(conn, None, 'create',
  455. (typeid,) + args, kwds)
  456. finally:
  457. conn.close()
  458. return Token(typeid, self._address, id), exposed
  459. def join(self, timeout=None):
  460. '''
  461. Join the manager process (if it has been spawned)
  462. '''
  463. if self._process is not None:
  464. self._process.join(timeout)
  465. if not self._process.is_alive():
  466. self._process = None
  467. def _debug_info(self):
  468. '''
  469. Return some info about the servers shared objects and connections
  470. '''
  471. conn = self._Client(self._address, authkey=self._authkey)
  472. try:
  473. return dispatch(conn, None, 'debug_info')
  474. finally:
  475. conn.close()
  476. def _number_of_objects(self):
  477. '''
  478. Return the number of shared objects
  479. '''
  480. conn = self._Client(self._address, authkey=self._authkey)
  481. try:
  482. return dispatch(conn, None, 'number_of_objects')
  483. finally:
  484. conn.close()
  485. def __enter__(self):
  486. if self._state.value == State.INITIAL:
  487. self.start()
  488. assert self._state.value == State.STARTED
  489. return self
  490. def __exit__(self, exc_type, exc_val, exc_tb):
  491. self.shutdown()
  492. @staticmethod
  493. def _finalize_manager(process, address, authkey, state, _Client):
  494. '''
  495. Shutdown the manager process; will be registered as a finalizer
  496. '''
  497. if process.is_alive():
  498. util.info('sending shutdown message to manager')
  499. try:
  500. conn = _Client(address, authkey=authkey)
  501. try:
  502. dispatch(conn, None, 'shutdown')
  503. finally:
  504. conn.close()
  505. except Exception:
  506. pass
  507. process.join(timeout=1.0)
  508. if process.is_alive():
  509. util.info('manager still alive')
  510. if hasattr(process, 'terminate'):
  511. util.info('trying to `terminate()` manager process')
  512. process.terminate()
  513. process.join(timeout=0.1)
  514. if process.is_alive():
  515. util.info('manager still alive after terminate')
  516. state.value = State.SHUTDOWN
  517. try:
  518. del BaseProxy._address_to_local[address]
  519. except KeyError:
  520. pass
  521. address = property(lambda self: self._address)
  522. @classmethod
  523. def register(cls, typeid, callable=None, proxytype=None, exposed=None,
  524. method_to_typeid=None, create_method=True):
  525. '''
  526. Register a typeid with the manager type
  527. '''
  528. if '_registry' not in cls.__dict__:
  529. cls._registry = cls._registry.copy()
  530. if proxytype is None:
  531. proxytype = AutoProxy
  532. exposed = exposed or getattr(proxytype, '_exposed_', None)
  533. method_to_typeid = (
  534. method_to_typeid or
  535. getattr(proxytype, '_method_to_typeid_', None)
  536. )
  537. if method_to_typeid:
  538. for key, value in items(method_to_typeid):
  539. assert type(key) is str, '%r is not a string' % key
  540. assert type(value) is str, '%r is not a string' % value
  541. cls._registry[typeid] = (
  542. callable, exposed, method_to_typeid, proxytype
  543. )
  544. if create_method:
  545. def temp(self, *args, **kwds):
  546. util.debug('requesting creation of a shared %r object', typeid)
  547. token, exp = self._create(typeid, *args, **kwds)
  548. proxy = proxytype(
  549. token, self._serializer, manager=self,
  550. authkey=self._authkey, exposed=exp
  551. )
  552. conn = self._Client(token.address, authkey=self._authkey)
  553. dispatch(conn, None, 'decref', (token.id,))
  554. return proxy
  555. temp.__name__ = typeid
  556. setattr(cls, typeid, temp)
  557. #
  558. # Subclass of set which get cleared after a fork
  559. #
  560. class ProcessLocalSet(set):
  561. def __init__(self):
  562. util.register_after_fork(self, lambda obj: obj.clear())
  563. def __reduce__(self):
  564. return type(self), ()
  565. #
  566. # Definition of BaseProxy
  567. #
  568. class BaseProxy(object):
  569. '''
  570. A base for proxies of shared objects
  571. '''
  572. _address_to_local = {}
  573. _mutex = util.ForkAwareThreadLock()
  574. def __init__(self, token, serializer, manager=None,
  575. authkey=None, exposed=None, incref=True):
  576. with BaseProxy._mutex:
  577. tls_idset = BaseProxy._address_to_local.get(token.address, None)
  578. if tls_idset is None:
  579. tls_idset = util.ForkAwareLocal(), ProcessLocalSet()
  580. BaseProxy._address_to_local[token.address] = tls_idset
  581. # self._tls is used to record the connection used by this
  582. # thread to communicate with the manager at token.address
  583. self._tls = tls_idset[0]
  584. # self._idset is used to record the identities of all shared
  585. # objects for which the current process owns references and
  586. # which are in the manager at token.address
  587. self._idset = tls_idset[1]
  588. self._token = token
  589. self._id = self._token.id
  590. self._manager = manager
  591. self._serializer = serializer
  592. self._Client = listener_client[serializer][1]
  593. if authkey is not None:
  594. self._authkey = process.AuthenticationString(authkey)
  595. elif self._manager is not None:
  596. self._authkey = self._manager._authkey
  597. else:
  598. self._authkey = process.current_process().authkey
  599. if incref:
  600. self._incref()
  601. util.register_after_fork(self, BaseProxy._after_fork)
  602. def _connect(self):
  603. util.debug('making connection to manager')
  604. name = process.current_process().name
  605. if threading.current_thread().name != 'MainThread':
  606. name += '|' + threading.current_thread().name
  607. conn = self._Client(self._token.address, authkey=self._authkey)
  608. dispatch(conn, None, 'accept_connection', (name,))
  609. self._tls.connection = conn
  610. def _callmethod(self, methodname, args=(), kwds={}):
  611. '''
  612. Try to call a method of the referrent and return a copy of the result
  613. '''
  614. try:
  615. conn = self._tls.connection
  616. except AttributeError:
  617. util.debug('thread %r does not own a connection',
  618. threading.current_thread().name)
  619. self._connect()
  620. conn = self._tls.connection
  621. conn.send((self._id, methodname, args, kwds))
  622. kind, result = conn.recv()
  623. if kind == '#RETURN':
  624. return result
  625. elif kind == '#PROXY':
  626. exposed, token = result
  627. proxytype = self._manager._registry[token.typeid][-1]
  628. token.address = self._token.address
  629. proxy = proxytype(
  630. token, self._serializer, manager=self._manager,
  631. authkey=self._authkey, exposed=exposed
  632. )
  633. conn = self._Client(token.address, authkey=self._authkey)
  634. dispatch(conn, None, 'decref', (token.id,))
  635. return proxy
  636. raise convert_to_error(kind, result)
  637. def _getvalue(self):
  638. '''
  639. Get a copy of the value of the referent
  640. '''
  641. return self._callmethod('#GETVALUE')
  642. def _incref(self):
  643. conn = self._Client(self._token.address, authkey=self._authkey)
  644. dispatch(conn, None, 'incref', (self._id,))
  645. util.debug('INCREF %r', self._token.id)
  646. self._idset.add(self._id)
  647. state = self._manager and self._manager._state
  648. self._close = util.Finalize(
  649. self, BaseProxy._decref,
  650. args=(self._token, self._authkey, state,
  651. self._tls, self._idset, self._Client),
  652. exitpriority=10
  653. )
  654. @staticmethod
  655. def _decref(token, authkey, state, tls, idset, _Client):
  656. idset.discard(token.id)
  657. # check whether manager is still alive
  658. if state is None or state.value == State.STARTED:
  659. # tell manager this process no longer cares about referent
  660. try:
  661. util.debug('DECREF %r', token.id)
  662. conn = _Client(token.address, authkey=authkey)
  663. dispatch(conn, None, 'decref', (token.id,))
  664. except Exception as exc:
  665. util.debug('... decref failed %s', exc)
  666. else:
  667. util.debug('DECREF %r -- manager already shutdown', token.id)
  668. # check whether we can close this thread's connection because
  669. # the process owns no more references to objects for this manager
  670. if not idset and hasattr(tls, 'connection'):
  671. util.debug('thread %r has no more proxies so closing conn',
  672. threading.current_thread().name)
  673. tls.connection.close()
  674. del tls.connection
  675. def _after_fork(self):
  676. self._manager = None
  677. try:
  678. self._incref()
  679. except Exception as exc:
  680. # the proxy may just be for a manager which has shutdown
  681. util.info('incref failed: %s', exc)
  682. def __reduce__(self):
  683. kwds = {}
  684. if context.get_spawning_popen() is not None:
  685. kwds['authkey'] = self._authkey
  686. if getattr(self, '_isauto', False):
  687. kwds['exposed'] = self._exposed_
  688. return (RebuildProxy,
  689. (AutoProxy, self._token, self._serializer, kwds))
  690. else:
  691. return (RebuildProxy,
  692. (type(self), self._token, self._serializer, kwds))
  693. def __deepcopy__(self, memo):
  694. return self._getvalue()
  695. def __repr__(self):
  696. return '<%s object, typeid %r at %#x>' % \
  697. (type(self).__name__, self._token.typeid, id(self))
  698. def __str__(self):
  699. '''
  700. Return representation of the referent (or a fall-back if that fails)
  701. '''
  702. try:
  703. return self._callmethod('__repr__')
  704. except Exception:
  705. return repr(self)[:-1] + "; '__str__()' failed>"
  706. #
  707. # Function used for unpickling
  708. #
  709. def RebuildProxy(func, token, serializer, kwds):
  710. '''
  711. Function used for unpickling proxy objects.
  712. If possible the shared object is returned, or otherwise a proxy for it.
  713. '''
  714. server = getattr(process.current_process(), '_manager_server', None)
  715. if server and server.address == token.address:
  716. return server.id_to_obj[token.id][0]
  717. else:
  718. incref = (
  719. kwds.pop('incref', True) and
  720. not getattr(process.current_process(), '_inheriting', False)
  721. )
  722. return func(token, serializer, incref=incref, **kwds)
  723. #
  724. # Functions to create proxies and proxy types
  725. #
  726. def MakeProxyType(name, exposed, _cache={}):
  727. '''
  728. Return an proxy type whose methods are given by `exposed`
  729. '''
  730. exposed = tuple(exposed)
  731. try:
  732. return _cache[(name, exposed)]
  733. except KeyError:
  734. pass
  735. dic = {}
  736. for meth in exposed:
  737. exec('''def %s(self, *args, **kwds):
  738. return self._callmethod(%r, args, kwds)''' % (meth, meth), dic)
  739. ProxyType = type(name, (BaseProxy,), dic)
  740. ProxyType._exposed_ = exposed
  741. _cache[(name, exposed)] = ProxyType
  742. return ProxyType
  743. def AutoProxy(token, serializer, manager=None, authkey=None,
  744. exposed=None, incref=True):
  745. '''
  746. Return an auto-proxy for `token`
  747. '''
  748. _Client = listener_client[serializer][1]
  749. if exposed is None:
  750. conn = _Client(token.address, authkey=authkey)
  751. try:
  752. exposed = dispatch(conn, None, 'get_methods', (token,))
  753. finally:
  754. conn.close()
  755. if authkey is None and manager is not None:
  756. authkey = manager._authkey
  757. if authkey is None:
  758. authkey = process.current_process().authkey
  759. ProxyType = MakeProxyType('AutoProxy[%s]' % token.typeid, exposed)
  760. proxy = ProxyType(token, serializer, manager=manager, authkey=authkey,
  761. incref=incref)
  762. proxy._isauto = True
  763. return proxy
  764. #
  765. # Types/callables which we will register with SyncManager
  766. #
  767. class Namespace(object):
  768. def __init__(self, **kwds):
  769. self.__dict__.update(kwds)
  770. def __repr__(self):
  771. items = list(self.__dict__.items())
  772. temp = []
  773. for name, value in items:
  774. if not name.startswith('_'):
  775. temp.append('%s=%r' % (name, value))
  776. temp.sort()
  777. return '%s(%s)' % (self.__class__.__name__, ', '.join(temp))
  778. class Value(object):
  779. def __init__(self, typecode, value, lock=True):
  780. self._typecode = typecode
  781. self._value = value
  782. def get(self):
  783. return self._value
  784. def set(self, value):
  785. self._value = value
  786. def __repr__(self):
  787. return '%s(%r, %r)' % (type(self).__name__,
  788. self._typecode, self._value)
  789. value = property(get, set)
  790. def Array(typecode, sequence, lock=True):
  791. return array.array(typecode, sequence)
  792. #
  793. # Proxy types used by SyncManager
  794. #
  795. class IteratorProxy(BaseProxy):
  796. if sys.version_info[0] == 3:
  797. _exposed = ('__next__', 'send', 'throw', 'close')
  798. else:
  799. _exposed_ = ('__next__', 'next', 'send', 'throw', 'close')
  800. def next(self, *args):
  801. return self._callmethod('next', args)
  802. def __iter__(self):
  803. return self
  804. def __next__(self, *args):
  805. return self._callmethod('__next__', args)
  806. def send(self, *args):
  807. return self._callmethod('send', args)
  808. def throw(self, *args):
  809. return self._callmethod('throw', args)
  810. def close(self, *args):
  811. return self._callmethod('close', args)
  812. class AcquirerProxy(BaseProxy):
  813. _exposed_ = ('acquire', 'release')
  814. def acquire(self, blocking=True, timeout=None):
  815. args = (blocking, ) if timeout is None else (blocking, timeout)
  816. return self._callmethod('acquire', args)
  817. def release(self):
  818. return self._callmethod('release')
  819. def __enter__(self):
  820. return self._callmethod('acquire')
  821. def __exit__(self, exc_type, exc_val, exc_tb):
  822. return self._callmethod('release')
  823. class ConditionProxy(AcquirerProxy):
  824. _exposed_ = ('acquire', 'release', 'wait', 'notify', 'notify_all')
  825. def wait(self, timeout=None):
  826. return self._callmethod('wait', (timeout,))
  827. def notify(self):
  828. return self._callmethod('notify')
  829. def notify_all(self):
  830. return self._callmethod('notify_all')
  831. def wait_for(self, predicate, timeout=None):
  832. result = predicate()
  833. if result:
  834. return result
  835. if timeout is not None:
  836. endtime = monotonic() + timeout
  837. else:
  838. endtime = None
  839. waittime = None
  840. while not result:
  841. if endtime is not None:
  842. waittime = endtime - monotonic()
  843. if waittime <= 0:
  844. break
  845. self.wait(waittime)
  846. result = predicate()
  847. return result
  848. class EventProxy(BaseProxy):
  849. _exposed_ = ('is_set', 'set', 'clear', 'wait')
  850. def is_set(self):
  851. return self._callmethod('is_set')
  852. def set(self):
  853. return self._callmethod('set')
  854. def clear(self):
  855. return self._callmethod('clear')
  856. def wait(self, timeout=None):
  857. return self._callmethod('wait', (timeout,))
  858. class BarrierProxy(BaseProxy):
  859. _exposed_ = ('__getattribute__', 'wait', 'abort', 'reset')
  860. def wait(self, timeout=None):
  861. return self._callmethod('wait', (timeout, ))
  862. def abort(self):
  863. return self._callmethod('abort')
  864. def reset(self):
  865. return self._callmethod('reset')
  866. @property
  867. def parties(self):
  868. return self._callmethod('__getattribute__', ('parties', ))
  869. @property
  870. def n_waiting(self):
  871. return self._callmethod('__getattribute__', ('n_waiting', ))
  872. @property
  873. def broken(self):
  874. return self._callmethod('__getattribute__', ('broken', ))
  875. class NamespaceProxy(BaseProxy):
  876. _exposed_ = ('__getattribute__', '__setattr__', '__delattr__')
  877. def __getattr__(self, key):
  878. if key[0] == '_':
  879. return object.__getattribute__(self, key)
  880. callmethod = object.__getattribute__(self, '_callmethod')
  881. return callmethod('__getattribute__', (key,))
  882. def __setattr__(self, key, value):
  883. if key[0] == '_':
  884. return object.__setattr__(self, key, value)
  885. callmethod = object.__getattribute__(self, '_callmethod')
  886. return callmethod('__setattr__', (key, value))
  887. def __delattr__(self, key):
  888. if key[0] == '_':
  889. return object.__delattr__(self, key)
  890. callmethod = object.__getattribute__(self, '_callmethod')
  891. return callmethod('__delattr__', (key,))
  892. class ValueProxy(BaseProxy):
  893. _exposed_ = ('get', 'set')
  894. def get(self):
  895. return self._callmethod('get')
  896. def set(self, value):
  897. return self._callmethod('set', (value,))
  898. value = property(get, set)
  899. _ListProxy_Attributes = (
  900. '__add__', '__contains__', '__delitem__', '__getitem__', '__len__',
  901. '__mul__', '__reversed__', '__rmul__', '__setitem__',
  902. 'append', 'count', 'extend', 'index', 'insert', 'pop', 'remove',
  903. 'reverse', 'sort', '__imul__',
  904. )
  905. if not PY3:
  906. _ListProxy_Attributes += ('__getslice__', '__setslice__', '__delslice__')
  907. BaseListProxy = MakeProxyType('BaseListProxy', _ListProxy_Attributes)
  908. class ListProxy(BaseListProxy):
  909. def __iadd__(self, value):
  910. self._callmethod('extend', (value,))
  911. return self
  912. def __imul__(self, value):
  913. self._callmethod('__imul__', (value,))
  914. return self
  915. DictProxy = MakeProxyType('DictProxy', (
  916. '__contains__', '__delitem__', '__getitem__', '__len__',
  917. '__setitem__', 'clear', 'copy', 'get', 'has_key', 'items',
  918. 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values',
  919. ))
  920. _ArrayProxy_Attributes = (
  921. '__len__', '__getitem__', '__setitem__',
  922. )
  923. if not PY3:
  924. _ArrayProxy_Attributes += ('__getslice__', '__setslice__')
  925. ArrayProxy = MakeProxyType('ArrayProxy', _ArrayProxy_Attributes)
  926. BasePoolProxy = MakeProxyType('PoolProxy', (
  927. 'apply', 'apply_async', 'close', 'imap', 'imap_unordered', 'join',
  928. 'map', 'map_async', 'starmap', 'starmap_async', 'terminate',
  929. ))
  930. BasePoolProxy._method_to_typeid_ = {
  931. 'apply_async': 'AsyncResult',
  932. 'map_async': 'AsyncResult',
  933. 'starmap_async': 'AsyncResult',
  934. 'imap': 'Iterator',
  935. 'imap_unordered': 'Iterator',
  936. }
  937. class PoolProxy(BasePoolProxy):
  938. def __enter__(self):
  939. return self
  940. def __exit__(self, *exc_info):
  941. self.terminate()
  942. #
  943. # Definition of SyncManager
  944. #
  945. class SyncManager(BaseManager):
  946. '''
  947. Subclass of `BaseManager` which supports a number of shared object types.
  948. The types registered are those intended for the synchronization
  949. of threads, plus `dict`, `list` and `Namespace`.
  950. The `billiard.Manager()` function creates started instances of
  951. this class.
  952. '''
  953. SyncManager.register('Queue', Queue)
  954. SyncManager.register('JoinableQueue', Queue)
  955. SyncManager.register('Event', threading.Event, EventProxy)
  956. SyncManager.register('Lock', threading.Lock, AcquirerProxy)
  957. SyncManager.register('RLock', threading.RLock, AcquirerProxy)
  958. SyncManager.register('Semaphore', threading.Semaphore, AcquirerProxy)
  959. SyncManager.register('BoundedSemaphore', threading.BoundedSemaphore,
  960. AcquirerProxy)
  961. SyncManager.register('Condition', threading.Condition, ConditionProxy)
  962. if hasattr(threading, 'Barrier'): # PY3
  963. SyncManager.register('Barrier', threading.Barrier, BarrierProxy)
  964. SyncManager.register('Pool', pool.Pool, PoolProxy)
  965. SyncManager.register('list', list, ListProxy)
  966. SyncManager.register('dict', dict, DictProxy)
  967. SyncManager.register('Value', Value, ValueProxy)
  968. SyncManager.register('Array', Array, ArrayProxy)
  969. SyncManager.register('Namespace', Namespace, NamespaceProxy)
  970. # types returned by methods of PoolProxy
  971. SyncManager.register('Iterator', proxytype=IteratorProxy, create_method=False)
  972. SyncManager.register('AsyncResult', create_method=False)