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