logo

oasis-root

Compiled tree of Oasis Linux based on own branch at <https://hacktivis.me/git/oasis/> git clone https://anongit.hacktivis.me/git/oasis-root.git

managers.py (47502B)


  1. #
  2. # Module providing manager classes 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. __all__ = [ 'BaseManager', 'SyncManager', 'BaseProxy', 'Token' ]
  11. #
  12. # Imports
  13. #
  14. import sys
  15. import threading
  16. import signal
  17. import array
  18. import queue
  19. import time
  20. import types
  21. import os
  22. from os import getpid
  23. from traceback import format_exc
  24. from . import connection
  25. from .context import reduction, get_spawning_popen, ProcessError
  26. from . import pool
  27. from . import process
  28. from . import util
  29. from . import get_context
  30. try:
  31. from . import shared_memory
  32. except ImportError:
  33. HAS_SHMEM = False
  34. else:
  35. HAS_SHMEM = True
  36. __all__.append('SharedMemoryManager')
  37. #
  38. # Register some things for pickling
  39. #
  40. def reduce_array(a):
  41. return array.array, (a.typecode, a.tobytes())
  42. reduction.register(array.array, reduce_array)
  43. view_types = [type(getattr({}, name)()) for name in ('items','keys','values')]
  44. if view_types[0] is not list: # only needed in Py3.0
  45. def rebuild_as_list(obj):
  46. return list, (list(obj),)
  47. for view_type in view_types:
  48. reduction.register(view_type, rebuild_as_list)
  49. #
  50. # Type for identifying shared objects
  51. #
  52. class Token(object):
  53. '''
  54. Type to uniquely identify a shared object
  55. '''
  56. __slots__ = ('typeid', 'address', 'id')
  57. def __init__(self, typeid, address, id):
  58. (self.typeid, self.address, self.id) = (typeid, address, id)
  59. def __getstate__(self):
  60. return (self.typeid, self.address, self.id)
  61. def __setstate__(self, state):
  62. (self.typeid, self.address, self.id) = state
  63. def __repr__(self):
  64. return '%s(typeid=%r, address=%r, id=%r)' % \
  65. (self.__class__.__name__, self.typeid, self.address, self.id)
  66. #
  67. # Function for communication with a manager's server process
  68. #
  69. def dispatch(c, id, methodname, args=(), kwds={}):
  70. '''
  71. Send a message to manager using connection `c` and return response
  72. '''
  73. c.send((id, methodname, args, kwds))
  74. kind, result = c.recv()
  75. if kind == '#RETURN':
  76. return result
  77. raise convert_to_error(kind, result)
  78. def convert_to_error(kind, result):
  79. if kind == '#ERROR':
  80. return result
  81. elif kind in ('#TRACEBACK', '#UNSERIALIZABLE'):
  82. if not isinstance(result, str):
  83. raise TypeError(
  84. "Result {0!r} (kind '{1}') type is {2}, not str".format(
  85. result, kind, type(result)))
  86. if kind == '#UNSERIALIZABLE':
  87. return RemoteError('Unserializable message: %s\n' % result)
  88. else:
  89. return RemoteError(result)
  90. else:
  91. return ValueError('Unrecognized message type {!r}'.format(kind))
  92. class RemoteError(Exception):
  93. def __str__(self):
  94. return ('\n' + '-'*75 + '\n' + str(self.args[0]) + '-'*75)
  95. #
  96. # Functions for finding the method names of an object
  97. #
  98. def all_methods(obj):
  99. '''
  100. Return a list of names of methods of `obj`
  101. '''
  102. temp = []
  103. for name in dir(obj):
  104. func = getattr(obj, name)
  105. if callable(func):
  106. temp.append(name)
  107. return temp
  108. def public_methods(obj):
  109. '''
  110. Return a list of names of methods of `obj` which do not start with '_'
  111. '''
  112. return [name for name in all_methods(obj) if name[0] != '_']
  113. #
  114. # Server which is run in a process controlled by a manager
  115. #
  116. class Server(object):
  117. '''
  118. Server class which runs in a process controlled by a manager object
  119. '''
  120. public = ['shutdown', 'create', 'accept_connection', 'get_methods',
  121. 'debug_info', 'number_of_objects', 'dummy', 'incref', 'decref']
  122. def __init__(self, registry, address, authkey, serializer):
  123. if not isinstance(authkey, bytes):
  124. raise TypeError(
  125. "Authkey {0!r} is type {1!s}, not bytes".format(
  126. authkey, type(authkey)))
  127. self.registry = registry
  128. self.authkey = process.AuthenticationString(authkey)
  129. Listener, Client = listener_client[serializer]
  130. # do authentication later
  131. self.listener = Listener(address=address, backlog=16)
  132. self.address = self.listener.address
  133. self.id_to_obj = {'0': (None, ())}
  134. self.id_to_refcount = {}
  135. self.id_to_local_proxy_obj = {}
  136. self.mutex = threading.Lock()
  137. def serve_forever(self):
  138. '''
  139. Run the server forever
  140. '''
  141. self.stop_event = threading.Event()
  142. process.current_process()._manager_server = self
  143. try:
  144. accepter = threading.Thread(target=self.accepter)
  145. accepter.daemon = True
  146. accepter.start()
  147. try:
  148. while not self.stop_event.is_set():
  149. self.stop_event.wait(1)
  150. except (KeyboardInterrupt, SystemExit):
  151. pass
  152. finally:
  153. if sys.stdout != sys.__stdout__: # what about stderr?
  154. util.debug('resetting stdout, stderr')
  155. sys.stdout = sys.__stdout__
  156. sys.stderr = sys.__stderr__
  157. sys.exit(0)
  158. def accepter(self):
  159. while True:
  160. try:
  161. c = self.listener.accept()
  162. except OSError:
  163. continue
  164. t = threading.Thread(target=self.handle_request, args=(c,))
  165. t.daemon = True
  166. t.start()
  167. def _handle_request(self, c):
  168. request = None
  169. try:
  170. connection.deliver_challenge(c, self.authkey)
  171. connection.answer_challenge(c, self.authkey)
  172. request = c.recv()
  173. ignore, funcname, args, kwds = request
  174. assert funcname in self.public, '%r unrecognized' % funcname
  175. func = getattr(self, funcname)
  176. except Exception:
  177. msg = ('#TRACEBACK', format_exc())
  178. else:
  179. try:
  180. result = func(c, *args, **kwds)
  181. except Exception:
  182. msg = ('#TRACEBACK', format_exc())
  183. else:
  184. msg = ('#RETURN', result)
  185. try:
  186. c.send(msg)
  187. except Exception as e:
  188. try:
  189. c.send(('#TRACEBACK', format_exc()))
  190. except Exception:
  191. pass
  192. util.info('Failure to send message: %r', msg)
  193. util.info(' ... request was %r', request)
  194. util.info(' ... exception was %r', e)
  195. def handle_request(self, conn):
  196. '''
  197. Handle a new connection
  198. '''
  199. try:
  200. self._handle_request(conn)
  201. except SystemExit:
  202. # Server.serve_client() calls sys.exit(0) on EOF
  203. pass
  204. finally:
  205. conn.close()
  206. def serve_client(self, conn):
  207. '''
  208. Handle requests from the proxies in a particular process/thread
  209. '''
  210. util.debug('starting server thread to service %r',
  211. threading.current_thread().name)
  212. recv = conn.recv
  213. send = conn.send
  214. id_to_obj = self.id_to_obj
  215. while not self.stop_event.is_set():
  216. try:
  217. methodname = obj = None
  218. request = recv()
  219. ident, methodname, args, kwds = request
  220. try:
  221. obj, exposed, gettypeid = id_to_obj[ident]
  222. except KeyError as ke:
  223. try:
  224. obj, exposed, gettypeid = \
  225. self.id_to_local_proxy_obj[ident]
  226. except KeyError:
  227. raise ke
  228. if methodname not in exposed:
  229. raise AttributeError(
  230. 'method %r of %r object is not in exposed=%r' %
  231. (methodname, type(obj), exposed)
  232. )
  233. function = getattr(obj, methodname)
  234. try:
  235. res = function(*args, **kwds)
  236. except Exception as e:
  237. msg = ('#ERROR', e)
  238. else:
  239. typeid = gettypeid and gettypeid.get(methodname, None)
  240. if typeid:
  241. rident, rexposed = self.create(conn, typeid, res)
  242. token = Token(typeid, self.address, rident)
  243. msg = ('#PROXY', (rexposed, token))
  244. else:
  245. msg = ('#RETURN', res)
  246. except AttributeError:
  247. if methodname is None:
  248. msg = ('#TRACEBACK', format_exc())
  249. else:
  250. try:
  251. fallback_func = self.fallback_mapping[methodname]
  252. result = fallback_func(
  253. self, conn, ident, obj, *args, **kwds
  254. )
  255. msg = ('#RETURN', result)
  256. except Exception:
  257. msg = ('#TRACEBACK', format_exc())
  258. except EOFError:
  259. util.debug('got EOF -- exiting thread serving %r',
  260. threading.current_thread().name)
  261. sys.exit(0)
  262. except Exception:
  263. msg = ('#TRACEBACK', format_exc())
  264. try:
  265. try:
  266. send(msg)
  267. except Exception:
  268. send(('#UNSERIALIZABLE', format_exc()))
  269. except Exception as e:
  270. util.info('exception in thread serving %r',
  271. threading.current_thread().name)
  272. util.info(' ... message was %r', msg)
  273. util.info(' ... exception was %r', e)
  274. conn.close()
  275. sys.exit(1)
  276. def fallback_getvalue(self, conn, ident, obj):
  277. return obj
  278. def fallback_str(self, conn, ident, obj):
  279. return str(obj)
  280. def fallback_repr(self, conn, ident, obj):
  281. return repr(obj)
  282. fallback_mapping = {
  283. '__str__':fallback_str,
  284. '__repr__':fallback_repr,
  285. '#GETVALUE':fallback_getvalue
  286. }
  287. def dummy(self, c):
  288. pass
  289. def debug_info(self, c):
  290. '''
  291. Return some info --- useful to spot problems with refcounting
  292. '''
  293. # Perhaps include debug info about 'c'?
  294. with self.mutex:
  295. result = []
  296. keys = list(self.id_to_refcount.keys())
  297. keys.sort()
  298. for ident in keys:
  299. if ident != '0':
  300. result.append(' %s: refcount=%s\n %s' %
  301. (ident, self.id_to_refcount[ident],
  302. str(self.id_to_obj[ident][0])[:75]))
  303. return '\n'.join(result)
  304. def number_of_objects(self, c):
  305. '''
  306. Number of shared objects
  307. '''
  308. # Doesn't use (len(self.id_to_obj) - 1) as we shouldn't count ident='0'
  309. return len(self.id_to_refcount)
  310. def shutdown(self, c):
  311. '''
  312. Shutdown this process
  313. '''
  314. try:
  315. util.debug('manager received shutdown message')
  316. c.send(('#RETURN', None))
  317. except:
  318. import traceback
  319. traceback.print_exc()
  320. finally:
  321. self.stop_event.set()
  322. def create(self, c, typeid, /, *args, **kwds):
  323. '''
  324. Create a new shared object and return its id
  325. '''
  326. with self.mutex:
  327. callable, exposed, method_to_typeid, proxytype = \
  328. self.registry[typeid]
  329. if callable is None:
  330. if kwds or (len(args) != 1):
  331. raise ValueError(
  332. "Without callable, must have one non-keyword argument")
  333. obj = args[0]
  334. else:
  335. obj = callable(*args, **kwds)
  336. if exposed is None:
  337. exposed = public_methods(obj)
  338. if method_to_typeid is not None:
  339. if not isinstance(method_to_typeid, dict):
  340. raise TypeError(
  341. "Method_to_typeid {0!r}: type {1!s}, not dict".format(
  342. method_to_typeid, type(method_to_typeid)))
  343. exposed = list(exposed) + list(method_to_typeid)
  344. ident = '%x' % id(obj) # convert to string because xmlrpclib
  345. # only has 32 bit signed integers
  346. util.debug('%r callable returned object with id %r', typeid, ident)
  347. self.id_to_obj[ident] = (obj, set(exposed), method_to_typeid)
  348. if ident not in self.id_to_refcount:
  349. self.id_to_refcount[ident] = 0
  350. self.incref(c, ident)
  351. return ident, tuple(exposed)
  352. def get_methods(self, c, token):
  353. '''
  354. Return the methods of the shared object indicated by token
  355. '''
  356. return tuple(self.id_to_obj[token.id][1])
  357. def accept_connection(self, c, name):
  358. '''
  359. Spawn a new thread to serve this connection
  360. '''
  361. threading.current_thread().name = name
  362. c.send(('#RETURN', None))
  363. self.serve_client(c)
  364. def incref(self, c, ident):
  365. with self.mutex:
  366. try:
  367. self.id_to_refcount[ident] += 1
  368. except KeyError as ke:
  369. # If no external references exist but an internal (to the
  370. # manager) still does and a new external reference is created
  371. # from it, restore the manager's tracking of it from the
  372. # previously stashed internal ref.
  373. if ident in self.id_to_local_proxy_obj:
  374. self.id_to_refcount[ident] = 1
  375. self.id_to_obj[ident] = \
  376. self.id_to_local_proxy_obj[ident]
  377. obj, exposed, gettypeid = self.id_to_obj[ident]
  378. util.debug('Server re-enabled tracking & INCREF %r', ident)
  379. else:
  380. raise ke
  381. def decref(self, c, ident):
  382. if ident not in self.id_to_refcount and \
  383. ident in self.id_to_local_proxy_obj:
  384. util.debug('Server DECREF skipping %r', ident)
  385. return
  386. with self.mutex:
  387. if self.id_to_refcount[ident] <= 0:
  388. raise AssertionError(
  389. "Id {0!s} ({1!r}) has refcount {2:n}, not 1+".format(
  390. ident, self.id_to_obj[ident],
  391. self.id_to_refcount[ident]))
  392. self.id_to_refcount[ident] -= 1
  393. if self.id_to_refcount[ident] == 0:
  394. del self.id_to_refcount[ident]
  395. if ident not in self.id_to_refcount:
  396. # Two-step process in case the object turns out to contain other
  397. # proxy objects (e.g. a managed list of managed lists).
  398. # Otherwise, deleting self.id_to_obj[ident] would trigger the
  399. # deleting of the stored value (another managed object) which would
  400. # in turn attempt to acquire the mutex that is already held here.
  401. self.id_to_obj[ident] = (None, (), None) # thread-safe
  402. util.debug('disposing of obj with id %r', ident)
  403. with self.mutex:
  404. del self.id_to_obj[ident]
  405. #
  406. # Class to represent state of a manager
  407. #
  408. class State(object):
  409. __slots__ = ['value']
  410. INITIAL = 0
  411. STARTED = 1
  412. SHUTDOWN = 2
  413. #
  414. # Mapping from serializer name to Listener and Client types
  415. #
  416. listener_client = {
  417. 'pickle' : (connection.Listener, connection.Client),
  418. 'xmlrpclib' : (connection.XmlListener, connection.XmlClient)
  419. }
  420. #
  421. # Definition of BaseManager
  422. #
  423. class BaseManager(object):
  424. '''
  425. Base class for managers
  426. '''
  427. _registry = {}
  428. _Server = Server
  429. def __init__(self, address=None, authkey=None, serializer='pickle',
  430. ctx=None):
  431. if authkey is None:
  432. authkey = process.current_process().authkey
  433. self._address = address # XXX not final address if eg ('', 0)
  434. self._authkey = process.AuthenticationString(authkey)
  435. self._state = State()
  436. self._state.value = State.INITIAL
  437. self._serializer = serializer
  438. self._Listener, self._Client = listener_client[serializer]
  439. self._ctx = ctx or get_context()
  440. def get_server(self):
  441. '''
  442. Return server object with serve_forever() method and address attribute
  443. '''
  444. if self._state.value != State.INITIAL:
  445. if self._state.value == State.STARTED:
  446. raise ProcessError("Already started server")
  447. elif self._state.value == State.SHUTDOWN:
  448. raise ProcessError("Manager has shut down")
  449. else:
  450. raise ProcessError(
  451. "Unknown state {!r}".format(self._state.value))
  452. return Server(self._registry, self._address,
  453. self._authkey, self._serializer)
  454. def connect(self):
  455. '''
  456. Connect manager object to the server process
  457. '''
  458. Listener, Client = listener_client[self._serializer]
  459. conn = Client(self._address, authkey=self._authkey)
  460. dispatch(conn, None, 'dummy')
  461. self._state.value = State.STARTED
  462. def start(self, initializer=None, initargs=()):
  463. '''
  464. Spawn a server process for this manager object
  465. '''
  466. if self._state.value != State.INITIAL:
  467. if self._state.value == State.STARTED:
  468. raise ProcessError("Already started server")
  469. elif self._state.value == State.SHUTDOWN:
  470. raise ProcessError("Manager has shut down")
  471. else:
  472. raise ProcessError(
  473. "Unknown state {!r}".format(self._state.value))
  474. if initializer is not None and not callable(initializer):
  475. raise TypeError('initializer must be a callable')
  476. # pipe over which we will retrieve address of server
  477. reader, writer = connection.Pipe(duplex=False)
  478. # spawn process which runs a server
  479. self._process = self._ctx.Process(
  480. target=type(self)._run_server,
  481. args=(self._registry, self._address, self._authkey,
  482. self._serializer, writer, initializer, initargs),
  483. )
  484. ident = ':'.join(str(i) for i in self._process._identity)
  485. self._process.name = type(self).__name__ + '-' + ident
  486. self._process.start()
  487. # get address of server
  488. writer.close()
  489. self._address = reader.recv()
  490. reader.close()
  491. # register a finalizer
  492. self._state.value = State.STARTED
  493. self.shutdown = util.Finalize(
  494. self, type(self)._finalize_manager,
  495. args=(self._process, self._address, self._authkey,
  496. self._state, self._Client),
  497. exitpriority=0
  498. )
  499. @classmethod
  500. def _run_server(cls, registry, address, authkey, serializer, writer,
  501. initializer=None, initargs=()):
  502. '''
  503. Create a server, report its address and run it
  504. '''
  505. # bpo-36368: protect server process from KeyboardInterrupt signals
  506. signal.signal(signal.SIGINT, signal.SIG_IGN)
  507. if initializer is not None:
  508. initializer(*initargs)
  509. # create server
  510. server = cls._Server(registry, address, authkey, serializer)
  511. # inform parent process of the server's address
  512. writer.send(server.address)
  513. writer.close()
  514. # run the manager
  515. util.info('manager serving at %r', server.address)
  516. server.serve_forever()
  517. def _create(self, typeid, /, *args, **kwds):
  518. '''
  519. Create a new shared object; return the token and exposed tuple
  520. '''
  521. assert self._state.value == State.STARTED, 'server not yet started'
  522. conn = self._Client(self._address, authkey=self._authkey)
  523. try:
  524. id, exposed = dispatch(conn, None, 'create', (typeid,)+args, kwds)
  525. finally:
  526. conn.close()
  527. return Token(typeid, self._address, id), exposed
  528. def join(self, timeout=None):
  529. '''
  530. Join the manager process (if it has been spawned)
  531. '''
  532. if self._process is not None:
  533. self._process.join(timeout)
  534. if not self._process.is_alive():
  535. self._process = None
  536. def _debug_info(self):
  537. '''
  538. Return some info about the servers shared objects and connections
  539. '''
  540. conn = self._Client(self._address, authkey=self._authkey)
  541. try:
  542. return dispatch(conn, None, 'debug_info')
  543. finally:
  544. conn.close()
  545. def _number_of_objects(self):
  546. '''
  547. Return the number of shared objects
  548. '''
  549. conn = self._Client(self._address, authkey=self._authkey)
  550. try:
  551. return dispatch(conn, None, 'number_of_objects')
  552. finally:
  553. conn.close()
  554. def __enter__(self):
  555. if self._state.value == State.INITIAL:
  556. self.start()
  557. if self._state.value != State.STARTED:
  558. if self._state.value == State.INITIAL:
  559. raise ProcessError("Unable to start server")
  560. elif self._state.value == State.SHUTDOWN:
  561. raise ProcessError("Manager has shut down")
  562. else:
  563. raise ProcessError(
  564. "Unknown state {!r}".format(self._state.value))
  565. return self
  566. def __exit__(self, exc_type, exc_val, exc_tb):
  567. self.shutdown()
  568. @staticmethod
  569. def _finalize_manager(process, address, authkey, state, _Client):
  570. '''
  571. Shutdown the manager process; will be registered as a finalizer
  572. '''
  573. if process.is_alive():
  574. util.info('sending shutdown message to manager')
  575. try:
  576. conn = _Client(address, authkey=authkey)
  577. try:
  578. dispatch(conn, None, 'shutdown')
  579. finally:
  580. conn.close()
  581. except Exception:
  582. pass
  583. process.join(timeout=1.0)
  584. if process.is_alive():
  585. util.info('manager still alive')
  586. if hasattr(process, 'terminate'):
  587. util.info('trying to `terminate()` manager process')
  588. process.terminate()
  589. process.join(timeout=0.1)
  590. if process.is_alive():
  591. util.info('manager still alive after terminate')
  592. state.value = State.SHUTDOWN
  593. try:
  594. del BaseProxy._address_to_local[address]
  595. except KeyError:
  596. pass
  597. @property
  598. def address(self):
  599. return self._address
  600. @classmethod
  601. def register(cls, typeid, callable=None, proxytype=None, exposed=None,
  602. method_to_typeid=None, create_method=True):
  603. '''
  604. Register a typeid with the manager type
  605. '''
  606. if '_registry' not in cls.__dict__:
  607. cls._registry = cls._registry.copy()
  608. if proxytype is None:
  609. proxytype = AutoProxy
  610. exposed = exposed or getattr(proxytype, '_exposed_', None)
  611. method_to_typeid = method_to_typeid or \
  612. getattr(proxytype, '_method_to_typeid_', None)
  613. if method_to_typeid:
  614. for key, value in list(method_to_typeid.items()): # isinstance?
  615. assert type(key) is str, '%r is not a string' % key
  616. assert type(value) is str, '%r is not a string' % value
  617. cls._registry[typeid] = (
  618. callable, exposed, method_to_typeid, proxytype
  619. )
  620. if create_method:
  621. def temp(self, /, *args, **kwds):
  622. util.debug('requesting creation of a shared %r object', typeid)
  623. token, exp = self._create(typeid, *args, **kwds)
  624. proxy = proxytype(
  625. token, self._serializer, manager=self,
  626. authkey=self._authkey, exposed=exp
  627. )
  628. conn = self._Client(token.address, authkey=self._authkey)
  629. dispatch(conn, None, 'decref', (token.id,))
  630. return proxy
  631. temp.__name__ = typeid
  632. setattr(cls, typeid, temp)
  633. #
  634. # Subclass of set which get cleared after a fork
  635. #
  636. class ProcessLocalSet(set):
  637. def __init__(self):
  638. util.register_after_fork(self, lambda obj: obj.clear())
  639. def __reduce__(self):
  640. return type(self), ()
  641. #
  642. # Definition of BaseProxy
  643. #
  644. class BaseProxy(object):
  645. '''
  646. A base for proxies of shared objects
  647. '''
  648. _address_to_local = {}
  649. _mutex = util.ForkAwareThreadLock()
  650. def __init__(self, token, serializer, manager=None,
  651. authkey=None, exposed=None, incref=True, manager_owned=False):
  652. with BaseProxy._mutex:
  653. tls_idset = BaseProxy._address_to_local.get(token.address, None)
  654. if tls_idset is None:
  655. tls_idset = util.ForkAwareLocal(), ProcessLocalSet()
  656. BaseProxy._address_to_local[token.address] = tls_idset
  657. # self._tls is used to record the connection used by this
  658. # thread to communicate with the manager at token.address
  659. self._tls = tls_idset[0]
  660. # self._idset is used to record the identities of all shared
  661. # objects for which the current process owns references and
  662. # which are in the manager at token.address
  663. self._idset = tls_idset[1]
  664. self._token = token
  665. self._id = self._token.id
  666. self._manager = manager
  667. self._serializer = serializer
  668. self._Client = listener_client[serializer][1]
  669. # Should be set to True only when a proxy object is being created
  670. # on the manager server; primary use case: nested proxy objects.
  671. # RebuildProxy detects when a proxy is being created on the manager
  672. # and sets this value appropriately.
  673. self._owned_by_manager = manager_owned
  674. if authkey is not None:
  675. self._authkey = process.AuthenticationString(authkey)
  676. elif self._manager is not None:
  677. self._authkey = self._manager._authkey
  678. else:
  679. self._authkey = process.current_process().authkey
  680. if incref:
  681. self._incref()
  682. util.register_after_fork(self, BaseProxy._after_fork)
  683. def _connect(self):
  684. util.debug('making connection to manager')
  685. name = process.current_process().name
  686. if threading.current_thread().name != 'MainThread':
  687. name += '|' + threading.current_thread().name
  688. conn = self._Client(self._token.address, authkey=self._authkey)
  689. dispatch(conn, None, 'accept_connection', (name,))
  690. self._tls.connection = conn
  691. def _callmethod(self, methodname, args=(), kwds={}):
  692. '''
  693. Try to call a method of the referent and return a copy of the result
  694. '''
  695. try:
  696. conn = self._tls.connection
  697. except AttributeError:
  698. util.debug('thread %r does not own a connection',
  699. threading.current_thread().name)
  700. self._connect()
  701. conn = self._tls.connection
  702. conn.send((self._id, methodname, args, kwds))
  703. kind, result = conn.recv()
  704. if kind == '#RETURN':
  705. return result
  706. elif kind == '#PROXY':
  707. exposed, token = result
  708. proxytype = self._manager._registry[token.typeid][-1]
  709. token.address = self._token.address
  710. proxy = proxytype(
  711. token, self._serializer, manager=self._manager,
  712. authkey=self._authkey, exposed=exposed
  713. )
  714. conn = self._Client(token.address, authkey=self._authkey)
  715. dispatch(conn, None, 'decref', (token.id,))
  716. return proxy
  717. raise convert_to_error(kind, result)
  718. def _getvalue(self):
  719. '''
  720. Get a copy of the value of the referent
  721. '''
  722. return self._callmethod('#GETVALUE')
  723. def _incref(self):
  724. if self._owned_by_manager:
  725. util.debug('owned_by_manager skipped INCREF of %r', self._token.id)
  726. return
  727. conn = self._Client(self._token.address, authkey=self._authkey)
  728. dispatch(conn, None, 'incref', (self._id,))
  729. util.debug('INCREF %r', self._token.id)
  730. self._idset.add(self._id)
  731. state = self._manager and self._manager._state
  732. self._close = util.Finalize(
  733. self, BaseProxy._decref,
  734. args=(self._token, self._authkey, state,
  735. self._tls, self._idset, self._Client),
  736. exitpriority=10
  737. )
  738. @staticmethod
  739. def _decref(token, authkey, state, tls, idset, _Client):
  740. idset.discard(token.id)
  741. # check whether manager is still alive
  742. if state is None or state.value == State.STARTED:
  743. # tell manager this process no longer cares about referent
  744. try:
  745. util.debug('DECREF %r', token.id)
  746. conn = _Client(token.address, authkey=authkey)
  747. dispatch(conn, None, 'decref', (token.id,))
  748. except Exception as e:
  749. util.debug('... decref failed %s', e)
  750. else:
  751. util.debug('DECREF %r -- manager already shutdown', token.id)
  752. # check whether we can close this thread's connection because
  753. # the process owns no more references to objects for this manager
  754. if not idset and hasattr(tls, 'connection'):
  755. util.debug('thread %r has no more proxies so closing conn',
  756. threading.current_thread().name)
  757. tls.connection.close()
  758. del tls.connection
  759. def _after_fork(self):
  760. self._manager = None
  761. try:
  762. self._incref()
  763. except Exception as e:
  764. # the proxy may just be for a manager which has shutdown
  765. util.info('incref failed: %s' % e)
  766. def __reduce__(self):
  767. kwds = {}
  768. if get_spawning_popen() is not None:
  769. kwds['authkey'] = self._authkey
  770. if getattr(self, '_isauto', False):
  771. kwds['exposed'] = self._exposed_
  772. return (RebuildProxy,
  773. (AutoProxy, self._token, self._serializer, kwds))
  774. else:
  775. return (RebuildProxy,
  776. (type(self), self._token, self._serializer, kwds))
  777. def __deepcopy__(self, memo):
  778. return self._getvalue()
  779. def __repr__(self):
  780. return '<%s object, typeid %r at %#x>' % \
  781. (type(self).__name__, self._token.typeid, id(self))
  782. def __str__(self):
  783. '''
  784. Return representation of the referent (or a fall-back if that fails)
  785. '''
  786. try:
  787. return self._callmethod('__repr__')
  788. except Exception:
  789. return repr(self)[:-1] + "; '__str__()' failed>"
  790. #
  791. # Function used for unpickling
  792. #
  793. def RebuildProxy(func, token, serializer, kwds):
  794. '''
  795. Function used for unpickling proxy objects.
  796. '''
  797. server = getattr(process.current_process(), '_manager_server', None)
  798. if server and server.address == token.address:
  799. util.debug('Rebuild a proxy owned by manager, token=%r', token)
  800. kwds['manager_owned'] = True
  801. if token.id not in server.id_to_local_proxy_obj:
  802. server.id_to_local_proxy_obj[token.id] = \
  803. server.id_to_obj[token.id]
  804. incref = (
  805. kwds.pop('incref', True) and
  806. not getattr(process.current_process(), '_inheriting', False)
  807. )
  808. return func(token, serializer, incref=incref, **kwds)
  809. #
  810. # Functions to create proxies and proxy types
  811. #
  812. def MakeProxyType(name, exposed, _cache={}):
  813. '''
  814. Return a proxy type whose methods are given by `exposed`
  815. '''
  816. exposed = tuple(exposed)
  817. try:
  818. return _cache[(name, exposed)]
  819. except KeyError:
  820. pass
  821. dic = {}
  822. for meth in exposed:
  823. exec('''def %s(self, /, *args, **kwds):
  824. return self._callmethod(%r, args, kwds)''' % (meth, meth), dic)
  825. ProxyType = type(name, (BaseProxy,), dic)
  826. ProxyType._exposed_ = exposed
  827. _cache[(name, exposed)] = ProxyType
  828. return ProxyType
  829. def AutoProxy(token, serializer, manager=None, authkey=None,
  830. exposed=None, incref=True, manager_owned=False):
  831. '''
  832. Return an auto-proxy for `token`
  833. '''
  834. _Client = listener_client[serializer][1]
  835. if exposed is None:
  836. conn = _Client(token.address, authkey=authkey)
  837. try:
  838. exposed = dispatch(conn, None, 'get_methods', (token,))
  839. finally:
  840. conn.close()
  841. if authkey is None and manager is not None:
  842. authkey = manager._authkey
  843. if authkey is None:
  844. authkey = process.current_process().authkey
  845. ProxyType = MakeProxyType('AutoProxy[%s]' % token.typeid, exposed)
  846. proxy = ProxyType(token, serializer, manager=manager, authkey=authkey,
  847. incref=incref, manager_owned=manager_owned)
  848. proxy._isauto = True
  849. return proxy
  850. #
  851. # Types/callables which we will register with SyncManager
  852. #
  853. class Namespace(object):
  854. def __init__(self, /, **kwds):
  855. self.__dict__.update(kwds)
  856. def __repr__(self):
  857. items = list(self.__dict__.items())
  858. temp = []
  859. for name, value in items:
  860. if not name.startswith('_'):
  861. temp.append('%s=%r' % (name, value))
  862. temp.sort()
  863. return '%s(%s)' % (self.__class__.__name__, ', '.join(temp))
  864. class Value(object):
  865. def __init__(self, typecode, value, lock=True):
  866. self._typecode = typecode
  867. self._value = value
  868. def get(self):
  869. return self._value
  870. def set(self, value):
  871. self._value = value
  872. def __repr__(self):
  873. return '%s(%r, %r)'%(type(self).__name__, self._typecode, self._value)
  874. value = property(get, set)
  875. def Array(typecode, sequence, lock=True):
  876. return array.array(typecode, sequence)
  877. #
  878. # Proxy types used by SyncManager
  879. #
  880. class IteratorProxy(BaseProxy):
  881. _exposed_ = ('__next__', 'send', 'throw', 'close')
  882. def __iter__(self):
  883. return self
  884. def __next__(self, *args):
  885. return self._callmethod('__next__', args)
  886. def send(self, *args):
  887. return self._callmethod('send', args)
  888. def throw(self, *args):
  889. return self._callmethod('throw', args)
  890. def close(self, *args):
  891. return self._callmethod('close', args)
  892. class AcquirerProxy(BaseProxy):
  893. _exposed_ = ('acquire', 'release')
  894. def acquire(self, blocking=True, timeout=None):
  895. args = (blocking,) if timeout is None else (blocking, timeout)
  896. return self._callmethod('acquire', args)
  897. def release(self):
  898. return self._callmethod('release')
  899. def __enter__(self):
  900. return self._callmethod('acquire')
  901. def __exit__(self, exc_type, exc_val, exc_tb):
  902. return self._callmethod('release')
  903. class ConditionProxy(AcquirerProxy):
  904. _exposed_ = ('acquire', 'release', 'wait', 'notify', 'notify_all')
  905. def wait(self, timeout=None):
  906. return self._callmethod('wait', (timeout,))
  907. def notify(self, n=1):
  908. return self._callmethod('notify', (n,))
  909. def notify_all(self):
  910. return self._callmethod('notify_all')
  911. def wait_for(self, predicate, timeout=None):
  912. result = predicate()
  913. if result:
  914. return result
  915. if timeout is not None:
  916. endtime = time.monotonic() + timeout
  917. else:
  918. endtime = None
  919. waittime = None
  920. while not result:
  921. if endtime is not None:
  922. waittime = endtime - time.monotonic()
  923. if waittime <= 0:
  924. break
  925. self.wait(waittime)
  926. result = predicate()
  927. return result
  928. class EventProxy(BaseProxy):
  929. _exposed_ = ('is_set', 'set', 'clear', 'wait')
  930. def is_set(self):
  931. return self._callmethod('is_set')
  932. def set(self):
  933. return self._callmethod('set')
  934. def clear(self):
  935. return self._callmethod('clear')
  936. def wait(self, timeout=None):
  937. return self._callmethod('wait', (timeout,))
  938. class BarrierProxy(BaseProxy):
  939. _exposed_ = ('__getattribute__', 'wait', 'abort', 'reset')
  940. def wait(self, timeout=None):
  941. return self._callmethod('wait', (timeout,))
  942. def abort(self):
  943. return self._callmethod('abort')
  944. def reset(self):
  945. return self._callmethod('reset')
  946. @property
  947. def parties(self):
  948. return self._callmethod('__getattribute__', ('parties',))
  949. @property
  950. def n_waiting(self):
  951. return self._callmethod('__getattribute__', ('n_waiting',))
  952. @property
  953. def broken(self):
  954. return self._callmethod('__getattribute__', ('broken',))
  955. class NamespaceProxy(BaseProxy):
  956. _exposed_ = ('__getattribute__', '__setattr__', '__delattr__')
  957. def __getattr__(self, key):
  958. if key[0] == '_':
  959. return object.__getattribute__(self, key)
  960. callmethod = object.__getattribute__(self, '_callmethod')
  961. return callmethod('__getattribute__', (key,))
  962. def __setattr__(self, key, value):
  963. if key[0] == '_':
  964. return object.__setattr__(self, key, value)
  965. callmethod = object.__getattribute__(self, '_callmethod')
  966. return callmethod('__setattr__', (key, value))
  967. def __delattr__(self, key):
  968. if key[0] == '_':
  969. return object.__delattr__(self, key)
  970. callmethod = object.__getattribute__(self, '_callmethod')
  971. return callmethod('__delattr__', (key,))
  972. class ValueProxy(BaseProxy):
  973. _exposed_ = ('get', 'set')
  974. def get(self):
  975. return self._callmethod('get')
  976. def set(self, value):
  977. return self._callmethod('set', (value,))
  978. value = property(get, set)
  979. __class_getitem__ = classmethod(types.GenericAlias)
  980. BaseListProxy = MakeProxyType('BaseListProxy', (
  981. '__add__', '__contains__', '__delitem__', '__getitem__', '__len__',
  982. '__mul__', '__reversed__', '__rmul__', '__setitem__',
  983. 'append', 'count', 'extend', 'index', 'insert', 'pop', 'remove',
  984. 'reverse', 'sort', '__imul__'
  985. ))
  986. class ListProxy(BaseListProxy):
  987. def __iadd__(self, value):
  988. self._callmethod('extend', (value,))
  989. return self
  990. def __imul__(self, value):
  991. self._callmethod('__imul__', (value,))
  992. return self
  993. DictProxy = MakeProxyType('DictProxy', (
  994. '__contains__', '__delitem__', '__getitem__', '__iter__', '__len__',
  995. '__setitem__', 'clear', 'copy', 'get', 'items',
  996. 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values'
  997. ))
  998. DictProxy._method_to_typeid_ = {
  999. '__iter__': 'Iterator',
  1000. }
  1001. ArrayProxy = MakeProxyType('ArrayProxy', (
  1002. '__len__', '__getitem__', '__setitem__'
  1003. ))
  1004. BasePoolProxy = MakeProxyType('PoolProxy', (
  1005. 'apply', 'apply_async', 'close', 'imap', 'imap_unordered', 'join',
  1006. 'map', 'map_async', 'starmap', 'starmap_async', 'terminate',
  1007. ))
  1008. BasePoolProxy._method_to_typeid_ = {
  1009. 'apply_async': 'AsyncResult',
  1010. 'map_async': 'AsyncResult',
  1011. 'starmap_async': 'AsyncResult',
  1012. 'imap': 'Iterator',
  1013. 'imap_unordered': 'Iterator'
  1014. }
  1015. class PoolProxy(BasePoolProxy):
  1016. def __enter__(self):
  1017. return self
  1018. def __exit__(self, exc_type, exc_val, exc_tb):
  1019. self.terminate()
  1020. #
  1021. # Definition of SyncManager
  1022. #
  1023. class SyncManager(BaseManager):
  1024. '''
  1025. Subclass of `BaseManager` which supports a number of shared object types.
  1026. The types registered are those intended for the synchronization
  1027. of threads, plus `dict`, `list` and `Namespace`.
  1028. The `multiprocessing.Manager()` function creates started instances of
  1029. this class.
  1030. '''
  1031. SyncManager.register('Queue', queue.Queue)
  1032. SyncManager.register('JoinableQueue', queue.Queue)
  1033. SyncManager.register('Event', threading.Event, EventProxy)
  1034. SyncManager.register('Lock', threading.Lock, AcquirerProxy)
  1035. SyncManager.register('RLock', threading.RLock, AcquirerProxy)
  1036. SyncManager.register('Semaphore', threading.Semaphore, AcquirerProxy)
  1037. SyncManager.register('BoundedSemaphore', threading.BoundedSemaphore,
  1038. AcquirerProxy)
  1039. SyncManager.register('Condition', threading.Condition, ConditionProxy)
  1040. SyncManager.register('Barrier', threading.Barrier, BarrierProxy)
  1041. SyncManager.register('Pool', pool.Pool, PoolProxy)
  1042. SyncManager.register('list', list, ListProxy)
  1043. SyncManager.register('dict', dict, DictProxy)
  1044. SyncManager.register('Value', Value, ValueProxy)
  1045. SyncManager.register('Array', Array, ArrayProxy)
  1046. SyncManager.register('Namespace', Namespace, NamespaceProxy)
  1047. # types returned by methods of PoolProxy
  1048. SyncManager.register('Iterator', proxytype=IteratorProxy, create_method=False)
  1049. SyncManager.register('AsyncResult', create_method=False)
  1050. #
  1051. # Definition of SharedMemoryManager and SharedMemoryServer
  1052. #
  1053. if HAS_SHMEM:
  1054. class _SharedMemoryTracker:
  1055. "Manages one or more shared memory segments."
  1056. def __init__(self, name, segment_names=[]):
  1057. self.shared_memory_context_name = name
  1058. self.segment_names = segment_names
  1059. def register_segment(self, segment_name):
  1060. "Adds the supplied shared memory block name to tracker."
  1061. util.debug(f"Register segment {segment_name!r} in pid {getpid()}")
  1062. self.segment_names.append(segment_name)
  1063. def destroy_segment(self, segment_name):
  1064. """Calls unlink() on the shared memory block with the supplied name
  1065. and removes it from the list of blocks being tracked."""
  1066. util.debug(f"Destroy segment {segment_name!r} in pid {getpid()}")
  1067. self.segment_names.remove(segment_name)
  1068. segment = shared_memory.SharedMemory(segment_name)
  1069. segment.close()
  1070. segment.unlink()
  1071. def unlink(self):
  1072. "Calls destroy_segment() on all tracked shared memory blocks."
  1073. for segment_name in self.segment_names[:]:
  1074. self.destroy_segment(segment_name)
  1075. def __del__(self):
  1076. util.debug(f"Call {self.__class__.__name__}.__del__ in {getpid()}")
  1077. self.unlink()
  1078. def __getstate__(self):
  1079. return (self.shared_memory_context_name, self.segment_names)
  1080. def __setstate__(self, state):
  1081. self.__init__(*state)
  1082. class SharedMemoryServer(Server):
  1083. public = Server.public + \
  1084. ['track_segment', 'release_segment', 'list_segments']
  1085. def __init__(self, *args, **kwargs):
  1086. Server.__init__(self, *args, **kwargs)
  1087. address = self.address
  1088. # The address of Linux abstract namespaces can be bytes
  1089. if isinstance(address, bytes):
  1090. address = os.fsdecode(address)
  1091. self.shared_memory_context = \
  1092. _SharedMemoryTracker(f"shm_{address}_{getpid()}")
  1093. util.debug(f"SharedMemoryServer started by pid {getpid()}")
  1094. def create(self, c, typeid, /, *args, **kwargs):
  1095. """Create a new distributed-shared object (not backed by a shared
  1096. memory block) and return its id to be used in a Proxy Object."""
  1097. # Unless set up as a shared proxy, don't make shared_memory_context
  1098. # a standard part of kwargs. This makes things easier for supplying
  1099. # simple functions.
  1100. if hasattr(self.registry[typeid][-1], "_shared_memory_proxy"):
  1101. kwargs['shared_memory_context'] = self.shared_memory_context
  1102. return Server.create(self, c, typeid, *args, **kwargs)
  1103. def shutdown(self, c):
  1104. "Call unlink() on all tracked shared memory, terminate the Server."
  1105. self.shared_memory_context.unlink()
  1106. return Server.shutdown(self, c)
  1107. def track_segment(self, c, segment_name):
  1108. "Adds the supplied shared memory block name to Server's tracker."
  1109. self.shared_memory_context.register_segment(segment_name)
  1110. def release_segment(self, c, segment_name):
  1111. """Calls unlink() on the shared memory block with the supplied name
  1112. and removes it from the tracker instance inside the Server."""
  1113. self.shared_memory_context.destroy_segment(segment_name)
  1114. def list_segments(self, c):
  1115. """Returns a list of names of shared memory blocks that the Server
  1116. is currently tracking."""
  1117. return self.shared_memory_context.segment_names
  1118. class SharedMemoryManager(BaseManager):
  1119. """Like SyncManager but uses SharedMemoryServer instead of Server.
  1120. It provides methods for creating and returning SharedMemory instances
  1121. and for creating a list-like object (ShareableList) backed by shared
  1122. memory. It also provides methods that create and return Proxy Objects
  1123. that support synchronization across processes (i.e. multi-process-safe
  1124. locks and semaphores).
  1125. """
  1126. _Server = SharedMemoryServer
  1127. def __init__(self, *args, **kwargs):
  1128. if os.name == "posix":
  1129. # bpo-36867: Ensure the resource_tracker is running before
  1130. # launching the manager process, so that concurrent
  1131. # shared_memory manipulation both in the manager and in the
  1132. # current process does not create two resource_tracker
  1133. # processes.
  1134. from . import resource_tracker
  1135. resource_tracker.ensure_running()
  1136. BaseManager.__init__(self, *args, **kwargs)
  1137. util.debug(f"{self.__class__.__name__} created by pid {getpid()}")
  1138. def __del__(self):
  1139. util.debug(f"{self.__class__.__name__}.__del__ by pid {getpid()}")
  1140. pass
  1141. def get_server(self):
  1142. 'Better than monkeypatching for now; merge into Server ultimately'
  1143. if self._state.value != State.INITIAL:
  1144. if self._state.value == State.STARTED:
  1145. raise ProcessError("Already started SharedMemoryServer")
  1146. elif self._state.value == State.SHUTDOWN:
  1147. raise ProcessError("SharedMemoryManager has shut down")
  1148. else:
  1149. raise ProcessError(
  1150. "Unknown state {!r}".format(self._state.value))
  1151. return self._Server(self._registry, self._address,
  1152. self._authkey, self._serializer)
  1153. def SharedMemory(self, size):
  1154. """Returns a new SharedMemory instance with the specified size in
  1155. bytes, to be tracked by the manager."""
  1156. with self._Client(self._address, authkey=self._authkey) as conn:
  1157. sms = shared_memory.SharedMemory(None, create=True, size=size)
  1158. try:
  1159. dispatch(conn, None, 'track_segment', (sms.name,))
  1160. except BaseException as e:
  1161. sms.unlink()
  1162. raise e
  1163. return sms
  1164. def ShareableList(self, sequence):
  1165. """Returns a new ShareableList instance populated with the values
  1166. from the input sequence, to be tracked by the manager."""
  1167. with self._Client(self._address, authkey=self._authkey) as conn:
  1168. sl = shared_memory.ShareableList(sequence)
  1169. try:
  1170. dispatch(conn, None, 'track_segment', (sl.shm.name,))
  1171. except BaseException as e:
  1172. sl.shm.unlink()
  1173. raise e
  1174. return sl