Python code coverage for Lib/multiprocessing/managers.py

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