managers.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092
  1. #
  2. # Module providing the `SyncManager` class for dealing
  3. # with shared objects
  4. #
  5. # processing/managers.py
  6. #
  7. # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
  8. #
  9. __all__ = [ 'BaseManager', 'SyncManager', 'BaseProxy',
  10. 'CreatorMethod', 'Token' ]
  11. #
  12. # Imports
  13. #
  14. import os
  15. import sys
  16. import socket
  17. import weakref
  18. import threading
  19. import traceback
  20. import array
  21. import copy_reg
  22. import cPickle
  23. from processing.connection import Listener, Client, Pipe, AuthenticationError
  24. from processing.connection import deliverChallenge, answerChallenge
  25. from processing.process import Process, currentProcess
  26. from processing.process import activeChildren, _registerAfterFork
  27. from processing.logger import debug, info, subWarning
  28. from processing.finalize import Finalize, _runFinalizers
  29. from processing.forking import exit
  30. #
  31. # Register `array.array` for pickling
  32. #
  33. def reduceArray(a):
  34. return array.array, (a.typecode, a.tostring())
  35. copy_reg.pickle(array.array, reduceArray)
  36. #
  37. # Exception class
  38. #
  39. class RemoteError(Exception):
  40. '''
  41. Exception type raised by managers
  42. '''
  43. def __init__(self, *args):
  44. if args:
  45. self.args = args
  46. else:
  47. info = sys.exc_info()
  48. self.args = (info[1], ''.join(traceback.format_exception(*info)))
  49. def __str__(self):
  50. return ('\n' + '-'*75 + '\nRemote ' + self.args[1] + '-'*75)
  51. #
  52. # Type for identifying shared objects
  53. #
  54. class Token(object):
  55. '''
  56. Type to uniquely indentify a shared object
  57. '''
  58. def __init__(self, typeid, address, id):
  59. self.typeid = typeid
  60. self.address = address
  61. self.id = id
  62. def __repr__(self):
  63. return 'Token(typeid=%r, address=%r, id=%r)' % \
  64. (self.typeid, self.address, self.id)
  65. #
  66. # Functions for communication with a manager's server process
  67. #
  68. def dispatch(c, id, methodname, args=(), kwds={}):
  69. '''
  70. Send a message to manager using connection `c` and return response
  71. '''
  72. c.send((id, methodname, args, kwds))
  73. kind, result = c.recv()
  74. if kind == '#RETURN':
  75. return result
  76. elif kind == '#ERROR':
  77. raise result
  78. else:
  79. raise ValueError
  80. def transact(address, authkey, methodname, args=(), kwds={}):
  81. '''
  82. Create connection then send a message to manager and return response
  83. '''
  84. conn = Client(address, authkey=authkey)
  85. try:
  86. return dispatch(conn, None, methodname, args, kwds)
  87. finally:
  88. conn.close()
  89. #
  90. # Functions for finding the method names of an object
  91. #
  92. def allMethods(obj):
  93. '''
  94. Return a list of names of methods of `obj`
  95. '''
  96. temp = []
  97. for name in dir(obj):
  98. func = getattr(obj, name)
  99. if hasattr(func, '__call__'):
  100. temp.append(name)
  101. return temp
  102. def publicMethods(obj):
  103. '''
  104. Return a list of names of methods of `obj` which do not start with '_'
  105. '''
  106. return filter(lambda name: name[0] != '_', allMethods(obj))
  107. #
  108. # Server which is run in a process controlled by a manager
  109. #
  110. class Server(object):
  111. '''
  112. Server class which runs in a process controlled by a manager object
  113. '''
  114. public = ['shutdown', 'create', 'acceptConnection',
  115. 'getMethods', 'debugInfo', 'dummy', 'incref', 'decref']
  116. def __init__(self, registry, address, authkey):
  117. assert type(authkey) is str
  118. self.registry = registry
  119. self.authkey = authkey
  120. # do authentication later
  121. self.listener = Listener(address=address, backlog=5)
  122. self.address = self.listener.address
  123. if type(self.address) is tuple:
  124. self.address = (socket.getfqdn(self.address[0]), self.address[1])
  125. self.id_to_obj = {0: (None, ())}
  126. self.id_to_refcount = {}
  127. self.mutex = threading.RLock()
  128. self.stop = 0
  129. def serveForever(self):
  130. '''
  131. Run the server forever
  132. '''
  133. try:
  134. try:
  135. while 1:
  136. c = self.listener.accept()
  137. t = threading.Thread(target=self.handleRequest, args=(c,))
  138. t.setDaemon(True)
  139. t.start()
  140. except (KeyboardInterrupt, SystemExit):
  141. return
  142. finally:
  143. self.stop = 999
  144. self.listener.close()
  145. def handleRequest(self, c):
  146. '''
  147. Handle a new connection
  148. '''
  149. funcname = result = request = None
  150. try:
  151. deliverChallenge(c, self.authkey)
  152. answerChallenge(c, self.authkey)
  153. request = c.recv()
  154. ignore, funcname, args, kwds = request
  155. assert funcname in self.public, '%r unrecognized' % funcname
  156. func = getattr(self, funcname)
  157. except (SystemExit, KeyboardInterrupt):
  158. raise
  159. except Exception:
  160. msg = ('#ERROR', RemoteError())
  161. else:
  162. try:
  163. result = func(c, *args, **kwds)
  164. msg = ('#RETURN', result)
  165. except (SystemExit, KeyboardInterrupt):
  166. raise
  167. except Exception:
  168. msg = ('#ERROR', RemoteError())
  169. try:
  170. c.send(msg)
  171. except (SystemExit, KeyboardInterrupt):
  172. raise
  173. except Exception, e:
  174. if msg[0] == '#ERROR':
  175. subWarning('Failure to send exception: %r', msg[1])
  176. else:
  177. subWarning('Failure to send result: %r', msg[1])
  178. subWarning(' ... request was %r', request)
  179. subWarning(' ... exception was %r', e)
  180. c.close()
  181. def serveClient(self, connection):
  182. '''
  183. Handle requests from the proxies in a particular process/thread
  184. '''
  185. debug('starting server thread to service %r',
  186. threading.currentThread().getName())
  187. recv = connection.recv
  188. send = connection.send
  189. id_to_obj = self.id_to_obj
  190. while not self.stop:
  191. try:
  192. methodname = obj = None
  193. request = recv()
  194. ident, methodname, args, kwds = request
  195. obj, exposed = id_to_obj[ident]
  196. if methodname not in exposed:
  197. raise AttributeError, (
  198. 'method %r of %r object is not in exposed=%r' %
  199. (methodname, type(obj), exposed)
  200. )
  201. function = getattr(obj, methodname)
  202. try:
  203. result = function(*args, **kwds)
  204. msg = ('#RETURN', result)
  205. except (SystemExit, KeyboardInterrupt):
  206. raise
  207. except Exception, e:
  208. msg = ('#ERROR', e)
  209. except AttributeError, e:
  210. if methodname is None:
  211. msg = ('#ERROR', RemoteError())
  212. else:
  213. try:
  214. fallback_func = self.fallback_mapping[methodname]
  215. result = fallback_func(
  216. self, connection, ident, obj, *args, **kwds
  217. )
  218. msg = ('#RETURN', result)
  219. except (SystemExit, KeyboardInterrupt):
  220. raise
  221. except Exception:
  222. msg = ('#ERROR', RemoteError())
  223. except EOFError:
  224. debug('got EOF -- exiting thread serving %r',
  225. threading.currentThread().getName())
  226. sys.exit(0)
  227. except (SystemExit, KeyboardInterrupt):
  228. raise
  229. except:
  230. msg = ('#ERROR', RemoteError())
  231. try:
  232. try:
  233. send(msg)
  234. except cPickle.PicklingError:
  235. result = msg[1]
  236. if hasattr(result, '__iter__') and hasattr(result, 'next'):
  237. try:
  238. # send a proxy for this iterator
  239. res_ident, _ = self.create(
  240. connection, 'iter', result
  241. )
  242. res_obj, res_exposed = self.id_to_obj[res_ident]
  243. token = Token('iter', self.address, res_ident)
  244. result = IteratorProxy(token, incref=False)
  245. msg = ('#RETURN', result)
  246. except (SystemExit, KeyboardInterrupt):
  247. raise
  248. except Exception:
  249. msg = ('#ERROR', RemoteError())
  250. else:
  251. msg = ('#ERROR', RemoteError())
  252. send(msg)
  253. except (SystemExit, KeyboardInterrupt):
  254. raise
  255. except Exception, e:
  256. subWarning('exception in thread serving %r',
  257. threading.currentThread().getName())
  258. subWarning(' ... message was %r', msg)
  259. subWarning(' ... exception was %r', e)
  260. connection.close()
  261. sys.exit(1)
  262. def fallbackGetValue(self, connection, ident, obj):
  263. return obj
  264. def fallbackStr(self, connection, ident, obj):
  265. return str(obj)
  266. def fallbackRepr(self, connection, ident, obj):
  267. return repr(obj)
  268. def fallbackCmp(self, connection, ident, obj, *args):
  269. return cmp(obj, *args)
  270. fallback_mapping = {
  271. '__str__':fallbackStr, '__repr__':fallbackRepr,
  272. '__cmp__':fallbackCmp, '#GETVALUE':fallbackGetValue
  273. }
  274. def dummy(self, c):
  275. pass
  276. def debugInfo(self, c):
  277. '''
  278. Return some info --- useful to spot problems with refcounting
  279. '''
  280. self.mutex.acquire()
  281. try:
  282. result = []
  283. keys = self.id_to_obj.keys()
  284. keys.sort()
  285. for ident in keys:
  286. if ident != 0:
  287. result.append(' %s: refcount=%s\n %s' %
  288. (hex(ident), self.id_to_refcount[ident],
  289. str(self.id_to_obj[ident][0])[:75]))
  290. return '\n'.join(result)
  291. finally:
  292. self.mutex.release()
  293. def shutdown(self, c):
  294. '''
  295. Shutdown this process
  296. '''
  297. c.send(('#RETURN', None))
  298. info('manager received shutdown message')
  299. # do some cleaning up
  300. _runFinalizers(0)
  301. for p in activeChildren():
  302. debug('terminating a child process of manager')
  303. p.terminate()
  304. for p in activeChildren():
  305. debug('terminating a child process of manager')
  306. p.join()
  307. _runFinalizers()
  308. info('manager exiting with exitcode 0')
  309. # now exit without waiting for other threads to finish
  310. exit(0)
  311. def create(self, c, typeid, *args, **kwds):
  312. '''
  313. Create a new shared object and return its id
  314. '''
  315. self.mutex.acquire()
  316. try:
  317. callable, exposed = self.registry[typeid]
  318. obj = callable(*args, **kwds)
  319. if exposed is None:
  320. exposed = publicMethods(obj)
  321. ident = id(obj)
  322. debug('have created %r object with id %r', typeid, ident)
  323. self.id_to_obj[ident] = (obj, set(exposed))
  324. if ident not in self.id_to_refcount:
  325. self.id_to_refcount[ident] = None
  326. return ident, tuple(exposed)
  327. finally:
  328. self.mutex.release()
  329. def getMethods(self, c, token):
  330. '''
  331. Return the methods of the shared object indicated by token
  332. '''
  333. return tuple(self.id_to_obj[token.id][1])
  334. def acceptConnection(self, c, name):
  335. '''
  336. Spawn a new thread to serve this connection
  337. '''
  338. threading.currentThread().setName(name)
  339. c.send(('#RETURN', None))
  340. self.serveClient(c)
  341. def incref(self, c, ident):
  342. self.mutex.acquire()
  343. try:
  344. try:
  345. self.id_to_refcount[ident] += 1
  346. except TypeError:
  347. assert self.id_to_refcount[ident] is None
  348. self.id_to_refcount[ident] = 1
  349. finally:
  350. self.mutex.release()
  351. def decref(self, c, ident):
  352. self.mutex.acquire()
  353. try:
  354. assert self.id_to_refcount[ident] >= 1
  355. self.id_to_refcount[ident] -= 1
  356. if self.id_to_refcount[ident] == 0:
  357. del self.id_to_obj[ident], self.id_to_refcount[ident]
  358. debug('disposing of obj with id %d', ident)
  359. finally:
  360. self.mutex.release()
  361. #
  362. # Definition of BaseManager
  363. #
  364. class BaseManager(object):
  365. '''
  366. Base class for managers
  367. '''
  368. def __init__(self, address=None, authkey=None):
  369. '''
  370. `address`:
  371. The address on which manager should listen for new
  372. connections. If `address` is None then an arbitrary one
  373. is chosen (which will be available as `self.address`).
  374. `authkey`:
  375. Only connections from clients which are using `authkey` as an
  376. authentication key will be accepted. If `authkey` is `None`
  377. then `currentProcess().getAuthKey()` is used.
  378. '''
  379. self._address = address # XXX not necessarily the final address
  380. if authkey is None:
  381. self._authkey = authkey = currentProcess().getAuthKey()
  382. else:
  383. self._authkey = authkey
  384. assert type(authkey) is str
  385. self._started = False
  386. def start(self):
  387. '''
  388. Spawn a server process for this manager object
  389. '''
  390. assert not self._started
  391. self._started = True
  392. self._registry, _ = BaseManager._getRegistryCreators(self)
  393. # pipe over which we will retreive address of server
  394. reader, writer = Pipe(duplex=False)
  395. # spawn process which runs a server
  396. self._process = Process(
  397. target=self._runServer,
  398. args=(self._registry, self._address, self._authkey, writer),
  399. )
  400. ident = ':'.join(map(str, self._process._identity))
  401. self._process.setName(type(self).__name__ + '-' + ident)
  402. self._process.setAuthKey(self._authkey)
  403. self._process.start()
  404. # get address of server
  405. writer.close()
  406. self._address = reader.recv()
  407. reader.close()
  408. # register a finalizer
  409. self.shutdown = Finalize(
  410. self, BaseManager._finalizeManager,
  411. args=(self._process, self._address, self._authkey),
  412. exitpriority=0
  413. )
  414. @classmethod
  415. def _runServer(cls, registry, address, authkey, writer):
  416. '''
  417. Create a server, report its address and run it
  418. '''
  419. # create server
  420. server = Server(registry, address, authkey)
  421. currentProcess()._server = server
  422. # inform parent process of the server's address
  423. writer.send(server.address)
  424. writer.close()
  425. # run the manager
  426. info('manager serving at %r', server.address)
  427. server.serveForever()
  428. def serveForever(self, verbose=True):
  429. '''
  430. Start server in the current process
  431. '''
  432. assert not self._started
  433. self._started = True
  434. registry, _ = BaseManager._getRegistryCreators(self)
  435. server = Server(registry, self._address, self._authkey)
  436. currentProcess()._server = server
  437. if verbose:
  438. print >>sys.stderr, '%s serving at address %s' % \
  439. (type(self).__name__, server.address)
  440. server.serveForever()
  441. @classmethod
  442. def fromAddress(cls, address, authkey):
  443. '''
  444. Create a new manager object for a pre-existing server process
  445. '''
  446. manager = cls(address, authkey)
  447. transact(address, authkey, 'dummy')
  448. manager._started = True
  449. return manager
  450. def _create(self, typeid, *args, **kwds):
  451. '''
  452. Create a new shared object; return the token and exposed tuple
  453. '''
  454. assert self._started
  455. id, exposed = transact(
  456. self._address, self._authkey, 'create', (typeid,) + args, kwds
  457. )
  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. self._process.join(timeout)
  464. def _debugInfo(self):
  465. '''
  466. Return some info about the servers shared objects and connections
  467. '''
  468. return transact(self._address, self._authkey, 'debugInfo')
  469. def _proxyFromToken(self, token):
  470. '''
  471. Create a proxy for a token
  472. '''
  473. assert token.address == self.address
  474. _, creators = BaseManager._getRegistryCreators(self)
  475. proxytype = creators[token.typeid]._proxytype
  476. return proxytype(token, authkey=self._authkey)
  477. @staticmethod
  478. def _getRegistryCreators(self_or_cls):
  479. registry = {}
  480. creators = {}
  481. for name in dir(self_or_cls):
  482. obj = getattr(self_or_cls, name)
  483. info = getattr(obj, '_manager_info', None)
  484. if info is not None and hasattr(obj, '__call__'):
  485. creators[name] = obj
  486. typeid, callable, exposed = info
  487. assert typeid not in registry, 'typeids must be unique'
  488. registry[typeid] = (callable, exposed)
  489. return registry, creators
  490. def __enter__(self):
  491. return self
  492. def __exit__(self, exc_type, exc_val, exc_tb):
  493. self.shutdown()
  494. @staticmethod
  495. def _finalizeManager(process, address, authkey):
  496. '''
  497. Shutdown the manager process; will be registered as a finalizer
  498. '''
  499. if process.isAlive():
  500. info('sending shutdown message to manager')
  501. try:
  502. transact(address, authkey, 'shutdown')
  503. except (SystemExit, KeyboardInterrupt):
  504. raise
  505. except Exception:
  506. pass
  507. process.join(timeout=0.2)
  508. if process.isAlive():
  509. info('manager still alive')
  510. if hasattr(process, 'terminate'):
  511. info('trying to `terminate()` manager process')
  512. process.terminate()
  513. process.join(timeout=0.1)
  514. if process.isAlive():
  515. info('manager still alive after terminate')
  516. try:
  517. del BaseProxy._address_to_local[address]
  518. except KeyError:
  519. pass
  520. address = property(lambda self: self._address)
  521. # deprecated
  522. from_address = fromAddress
  523. serve_forever = serveForever
  524. #
  525. # Function for adding methods to managers
  526. #
  527. def CreatorMethod(callable=None, proxytype=None, exposed=None, typeid=None):
  528. '''
  529. Returns a method for a manager class which will create
  530. a shared object using `callable` and return a proxy for it.
  531. '''
  532. if exposed is None and hasattr(callable, '__exposed__'):
  533. exposed = callable.__exposed__
  534. if proxytype is None:
  535. proxytype = MakeAutoProxy
  536. typeid = typeid or _uniqueLabel(callable.__name__)
  537. def temp(self, *args, **kwds):
  538. debug('requesting creation of a shared %r object', typeid)
  539. token, exp = self._create(typeid, *args, **kwds)
  540. proxy = proxytype(
  541. token, manager=self, authkey=self._authkey, exposed=exp
  542. )
  543. return proxy
  544. try:
  545. temp.__name__ = typeid
  546. except TypeError:
  547. pass
  548. temp._manager_info = (typeid, callable, exposed)
  549. temp._proxytype = proxytype
  550. return temp
  551. def _uniqueLabel(prefix, _count={}):
  552. '''
  553. Return a string beginning with 'prefix' which has not already been used.
  554. '''
  555. try:
  556. _count[prefix] += 1
  557. return prefix + '-' + str(_count[prefix])
  558. except KeyError:
  559. _count[prefix] = 0
  560. return prefix
  561. #
  562. # Subclasses of threading.local and set which get cleared after a fork
  563. #
  564. class ProcessLocalSet(set):
  565. def __init__(self):
  566. _registerAfterFork(self, set.clear)
  567. def __reduce__(self):
  568. return type(self), ()
  569. class ThreadLocalStorage(threading.local):
  570. def __init__(self):
  571. _registerAfterFork(self, _clearNamespace)
  572. def __reduce__(self):
  573. return type(self), ()
  574. def _clearNamespace(obj):
  575. obj.__dict__.clear()
  576. #
  577. # Definition of BaseProxy
  578. #
  579. class BaseProxy(object):
  580. '''
  581. A base for proxies of shared objects
  582. '''
  583. _address_to_local = {}
  584. _mutex = threading.Lock()
  585. def __init__(self, token, manager=None,
  586. authkey=None, exposed=None, incref=True):
  587. BaseProxy._mutex.acquire()
  588. try:
  589. tls_idset = BaseProxy._address_to_local.get(token.address, None)
  590. if tls_idset is None:
  591. tls_idset = ThreadLocalStorage(), ProcessLocalSet()
  592. BaseProxy._address_to_local[token.address] = tls_idset
  593. finally:
  594. BaseProxy._mutex.release()
  595. # self._tls is used to record the connection used by this
  596. # thread to communicate with the manager at token.address
  597. self._tls = tls_idset[0]
  598. # self._idset is used to record the identities of all shared
  599. # objects for which the current process owns references and
  600. # which are in the manager at token.address
  601. self._idset = tls_idset[1]
  602. self._token = token
  603. self._id = self._token.id
  604. self._manager = manager
  605. if authkey:
  606. self._authkey = authkey
  607. elif self._manager:
  608. self._authkey = self._manager
  609. else:
  610. self._authkey = currentProcess().getAuthKey()
  611. if incref:
  612. self._incref()
  613. _registerAfterFork(self, BaseProxy._afterFork)
  614. def _connect(self):
  615. debug('making connection to manager')
  616. name = currentProcess().getName()
  617. if threading.currentThread().getName() != 'MainThread':
  618. name += '|' + threading.currentThread().getName()
  619. connection = Client(self._token.address, authkey=self._authkey)
  620. dispatch(connection, None, 'acceptConnection', (name,))
  621. self._tls.connection = connection
  622. def _callMethod(self, methodname, args=(), kwds={}):
  623. '''
  624. Try to call a method of the referrent and return a copy of the result
  625. '''
  626. try:
  627. conn = self._tls.connection
  628. except AttributeError:
  629. debug('thread %r does not own a connection',
  630. threading.currentThread().getName())
  631. self._connect()
  632. conn = self._tls.connection
  633. conn.send((self._id, methodname, args, kwds))
  634. kind, result = conn.recv()
  635. if kind == '#RETURN':
  636. return result
  637. elif kind == '#ERROR':
  638. raise result
  639. else:
  640. raise ValueError
  641. def _getValue(self):
  642. '''
  643. Get a copy of the value of the referent
  644. '''
  645. return self._callMethod('#GETVALUE')
  646. def _incref(self):
  647. connection = Client(self._token.address, authkey=self._authkey)
  648. dispatch(connection, None, 'incref', (self._id,))
  649. debug('INCREF %r', self._token.id)
  650. assert self._id not in self._idset
  651. self._idset.add(self._id)
  652. shutdown = getattr(self._manager, 'shutdown', None)
  653. self._close = Finalize(
  654. self, BaseProxy._decref,
  655. args=(self._token, self._authkey, shutdown,
  656. self._tls, self._idset),
  657. exitpriority=10
  658. )
  659. @staticmethod
  660. def _decref(token, authkey, shutdown, tls, idset):
  661. idset.remove(token.id)
  662. # check whether manager is still alive
  663. manager_still_alive = shutdown is None or shutdown.stillActive()
  664. if manager_still_alive:
  665. # tell manager this process no longer cares about referent
  666. try:
  667. debug('DECREF %r', token.id)
  668. connection = Client(token.address, authkey=authkey)
  669. dispatch(connection, None, 'decref', (token.id,))
  670. except (SystemExit, KeyboardInterrupt):
  671. raise
  672. except Exception, e:
  673. debug('... decref failed %s', e)
  674. else:
  675. debug('DECREF %r -- manager already shutdown',
  676. token.id)
  677. # check whether we can close this thread's connection because
  678. # the process owns no more references to objects for this manager
  679. if not idset and hasattr(tls, 'connection'):
  680. debug('thread %r has no more proxies so closing conn',
  681. threading.currentThread().getName())
  682. tls.connection.close()
  683. del tls.connection
  684. def _afterFork(self):
  685. self._manager = None
  686. self._incref()
  687. def __reduce__(self):
  688. if hasattr(self, '_exposed'):
  689. return (RebuildProxy, (MakeAutoProxy, self._token,
  690. {'exposed': self._exposed}))
  691. else:
  692. return (RebuildProxy, (type(self), self._token, {}))
  693. def __deepcopy__(self, memo):
  694. return self._getValue()
  695. def __hash__(self):
  696. raise NotImplementedError, 'proxies are unhashable'
  697. def __repr__(self):
  698. return '<Proxy[%s] object at %s>' % (self._token.typeid,
  699. '0x%x' % id(self))
  700. def __str__(self):
  701. '''
  702. Return representation of the referent (or a fall-back if that fails)
  703. '''
  704. try:
  705. return self._callMethod('__repr__')
  706. except (SystemExit, KeyboardInterrupt):
  707. raise
  708. except Exception:
  709. return repr(self)[:-1] + "; '__str__()' failed>"
  710. # deprecated
  711. _callmethod = _callMethod
  712. _getvalue = _getValue
  713. #
  714. # Since BaseProxy._mutex might be locked at time of fork we reset it
  715. #
  716. def _resetMutex(obj):
  717. obj._mutex = threading.Lock()
  718. _registerAfterFork(BaseProxy, _resetMutex)
  719. #
  720. # Function used for unpickling
  721. #
  722. def RebuildProxy(func, token, kwds={}):
  723. '''
  724. Function used for unpickling proxy objects.
  725. If possible the shared object is returned, or otherwise a proxy for it.
  726. '''
  727. server = getattr(currentProcess(), '_server', None)
  728. if server and server.address == token.address:
  729. return server.id_to_obj[token.id][0]
  730. else:
  731. incref = (
  732. kwds.pop('incref', True) and
  733. not getattr(currentProcess(), '_inheriting', False)
  734. )
  735. try:
  736. return func(token, manager=None, authkey=None,
  737. incref=incref, **kwds)
  738. except AuthenticationError:
  739. raise AuthenticationError, 'cannot rebuild proxy without authkey'
  740. #
  741. # Functions to create proxies and proxy types
  742. #
  743. def MakeAutoProxyType(exposed, typeid='unnamed', _cache={}):
  744. '''
  745. Return an auto-proxy type whose methods are given by `exposed`
  746. '''
  747. exposed = tuple(exposed)
  748. try:
  749. return _cache[(typeid, exposed)]
  750. except KeyError:
  751. pass
  752. dic = {}
  753. for name in exposed:
  754. exec '''def %s(self, *args, **kwds):
  755. return self._callMethod(%r, args, kwds)''' % (name, name) in dic
  756. ProxyType = type('AutoProxy[%s]' % typeid, (BaseProxy,), dic)
  757. ProxyType._exposed = exposed
  758. _cache[(typeid, exposed)] = ProxyType
  759. return ProxyType
  760. def MakeAutoProxy(token, manager=None, authkey=None,
  761. exposed=None, incref=True):
  762. '''
  763. Return an auto-proxy for `token`
  764. '''
  765. if exposed is None:
  766. exposed = transact(token.address, authkey, 'getMethods', (token,))
  767. ProxyType = MakeAutoProxyType(exposed, token.typeid)
  768. proxy = ProxyType(token, manager=manager, authkey=authkey, incref=incref)
  769. return proxy
  770. #
  771. # Types (or functions) which we will register with SyncManager
  772. #
  773. from threading import BoundedSemaphore, Condition, Event, \
  774. Lock, RLock, Semaphore
  775. from Queue import Queue
  776. class Namespace(object):
  777. '''
  778. Instances of this class can be used as namespaces.
  779. A namespace object has no public methods but does have writable
  780. attributes. Its represention shows the values of its attributes.
  781. '''
  782. __exposed__ = ('__getattribute__', '__setattr__', '__delattr__')
  783. def __repr__(self):
  784. items = self.__dict__.items()
  785. temp = []
  786. for name, value in items:
  787. if not name.startswith('_'):
  788. temp.append('%s=%r' % (name, value))
  789. temp.sort()
  790. return 'Namespace(%s)' % str.join(', ', temp)
  791. class Value(object):
  792. '''
  793. Instances have a settable 'value' property
  794. '''
  795. def __init__(self, typecode, value, lock=True):
  796. self._typecode = typecode
  797. self._value = value
  798. def get(self):
  799. return self._value
  800. def set(self, value):
  801. self._value = value
  802. def __repr__(self):
  803. return '%s(%r, %r)'%(type(self).__name__, self._typecode, self._value)
  804. value = property(get, set)
  805. #
  806. # Proxy type used by BaseManager
  807. #
  808. class IteratorProxy(BaseProxy):
  809. '''
  810. Proxy type for iterators
  811. '''
  812. def __iter__(self):
  813. return self
  814. def next(self):
  815. return self._callMethod('next')
  816. BaseManager._Iter = CreatorMethod(iter, IteratorProxy, ('next', '__iter__'))
  817. #
  818. # Proxy types used by SyncManager
  819. #
  820. class AcquirerProxy(BaseProxy):
  821. '''
  822. Base class for proxies which have acquire and release methods
  823. '''
  824. def acquire(self, blocking=1):
  825. return self._callMethod('acquire', (blocking,))
  826. def release(self):
  827. return self._callMethod('release')
  828. def __enter__(self):
  829. self._callMethod('acquire')
  830. return self
  831. def __exit__(self, exc_type, exc_val, exc_tb):
  832. return self._callMethod('release')
  833. class ConditionProxy(AcquirerProxy):
  834. def wait(self, timeout=None):
  835. return self._callMethod('wait', (timeout,))
  836. def notify(self):
  837. return self._callMethod('notify')
  838. def notifyAll(self):
  839. return self._callMethod('notifyAll')
  840. class NamespaceProxy(BaseProxy):
  841. '''
  842. Proxy type for Namespace objects.
  843. Note that attributes beginning with '_' will "belong" to the proxy,
  844. while other attributes "belong" to the referent.
  845. '''
  846. def __getattr__(self, key):
  847. if key[0] == '_':
  848. return object.__getattribute__(self, key)
  849. callmethod = object.__getattribute__(self, '_callMethod')
  850. return callmethod('__getattribute__', (key,))
  851. def __setattr__(self, key, value):
  852. if key[0] == '_':
  853. return object.__setattr__(self, key, value)
  854. callmethod = object.__getattribute__(self, '_callMethod')
  855. return callmethod('__setattr__', (key, value))
  856. def __delattr__(self, key):
  857. if key[0] == '_':
  858. return object.__delattr__(self, key)
  859. callmethod = object.__getattribute__(self, '_callMethod')
  860. return callmethod('__delattr__', (key,))
  861. _list_exposed = (
  862. '__add__', '__contains__', '__delitem__', '__delslice__',
  863. '__cmp__', '__getitem__', '__getslice__', '__iter__', '__imul__',
  864. '__len__', '__mul__', '__reversed__', '__rmul__', '__setitem__',
  865. '__setslice__', 'append', 'count', 'extend', 'index', 'insert',
  866. 'pop', 'remove', 'reverse', 'sort'
  867. )
  868. BaseListProxy = MakeAutoProxyType(_list_exposed, 'BaseListProxy')
  869. class ListProxy(BaseListProxy):
  870. # augmented assignment functions must return self
  871. def __iadd__(self, value):
  872. self._callMethod('extend', (value,))
  873. return self
  874. def __imul__(self, value):
  875. # Inefficient since a copy of the target is transferred and discarded
  876. self._callMethod('__imul__', (value,))
  877. return self
  878. _dict_exposed=(
  879. '__cmp__', '__contains__', '__delitem__', '__getitem__',
  880. '__iter__', '__len__', '__setitem__', 'clear', 'copy', 'get',
  881. 'has_key', 'items', 'iteritems', 'iterkeys', 'itervalues',
  882. 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values'
  883. )
  884. class ValueProxy(BaseProxy):
  885. def get(self):
  886. return self._callMethod('get')
  887. def set(self, value):
  888. return self._callMethod('set', (value,))
  889. value = property(get, set)
  890. def Array(typecode, sequence, lock=True):
  891. return array.array(typecode, sequence)
  892. _arr_exposed = (
  893. '__len__', '__iter__', '__getitem__', '__setitem__',
  894. '__getslice__', '__setslice__'
  895. )
  896. #
  897. # Definition of SyncManager
  898. #
  899. class SyncManager(BaseManager):
  900. '''
  901. Subclass of `BaseManager` which supports a number of shared object types.
  902. The types registered are those intended for the synchronization
  903. of threads, plus `dict`, `list` and `Namespace`.
  904. The `processing.Manager` function creates instances of this class.
  905. '''
  906. Event = CreatorMethod(Event)
  907. Queue = CreatorMethod(Queue)
  908. Lock = CreatorMethod(Lock, AcquirerProxy)
  909. RLock = CreatorMethod(RLock, AcquirerProxy)
  910. Semaphore = CreatorMethod(Semaphore, AcquirerProxy)
  911. BoundedSemaphore = CreatorMethod(BoundedSemaphore, AcquirerProxy)
  912. Condition = CreatorMethod(Condition, ConditionProxy)
  913. Namespace = CreatorMethod(Namespace, NamespaceProxy)
  914. list = CreatorMethod(list, ListProxy, exposed=_list_exposed)
  915. dict = CreatorMethod(dict, exposed=_dict_exposed)
  916. Value = CreatorMethod(Value, ValueProxy)
  917. Array = CreatorMethod(Array, exposed=_arr_exposed)