Python code coverage for Lib/multiprocessing/connection.py

#	count	content
1	n/a	#
2	n/a	# A higher level module for using sockets (or Windows named pipes)
3	n/a	#
4	n/a	# multiprocessing/connection.py
5	n/a	#
6	n/a	# Copyright (c) 2006-2008, R Oudkerk
7	n/a	# Licensed to PSF under a Contributor Agreement.
8	n/a	#
9	n/a
10	n/a	__all__ = [ 'Client', 'Listener', 'Pipe', 'wait' ]
11	n/a
12	n/a	import io
13	n/a	import os
14	n/a	import sys
15	n/a	import socket
16	n/a	import struct
17	n/a	import time
18	n/a	import tempfile
19	n/a	import itertools
20	n/a
21	n/a	import _multiprocessing
22	n/a
23	n/a	from . import util
24	n/a
25	n/a	from . import AuthenticationError, BufferTooShort
26	n/a	from .context import reduction
27	n/a	_ForkingPickler = reduction.ForkingPickler
28	n/a
29	n/a	try:
30	n/a	import _winapi
31	n/a	from _winapi import WAIT_OBJECT_0, WAIT_ABANDONED_0, WAIT_TIMEOUT, INFINITE
32	n/a	except ImportError:
33	n/a	if sys.platform == 'win32':
34	n/a	raise
35	n/a	_winapi = None
36	n/a
37	n/a	#
38	n/a	#
39	n/a	#
40	n/a
41	n/a	BUFSIZE = 8192
42	n/a	# A very generous timeout when it comes to local connections...
43	n/a	CONNECTION_TIMEOUT = 20.
44	n/a
45	n/a	_mmap_counter = itertools.count()
46	n/a
47	n/a	default_family = 'AF_INET'
48	n/a	families = ['AF_INET']
49	n/a
50	n/a	if hasattr(socket, 'AF_UNIX'):
51	n/a	default_family = 'AF_UNIX'
52	n/a	families += ['AF_UNIX']
53	n/a
54	n/a	if sys.platform == 'win32':
55	n/a	default_family = 'AF_PIPE'
56	n/a	families += ['AF_PIPE']
57	n/a
58	n/a
59	n/a	def _init_timeout(timeout=CONNECTION_TIMEOUT):
60	n/a	return time.time() + timeout
61	n/a
62	n/a	def _check_timeout(t):
63	n/a	return time.time() > t
64	n/a
65	n/a	#
66	n/a	#
67	n/a	#
68	n/a
69	n/a	def arbitrary_address(family):
70	n/a	'''
71	n/a	Return an arbitrary free address for the given family
72	n/a	'''
73	n/a	if family == 'AF_INET':
74	n/a	return ('localhost', 0)
75	n/a	elif family == 'AF_UNIX':
76	n/a	return tempfile.mktemp(prefix='listener-', dir=util.get_temp_dir())
77	n/a	elif family == 'AF_PIPE':
78	n/a	return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' %
79	n/a	(os.getpid(), next(_mmap_counter)), dir="")
80	n/a	else:
81	n/a	raise ValueError('unrecognized family')
82	n/a
83	n/a	def _validate_family(family):
84	n/a	'''
85	n/a	Checks if the family is valid for the current environment.
86	n/a	'''
87	n/a	if sys.platform != 'win32' and family == 'AF_PIPE':
88	n/a	raise ValueError('Family %s is not recognized.' % family)
89	n/a
90	n/a	if sys.platform == 'win32' and family == 'AF_UNIX':
91	n/a	# double check
92	n/a	if not hasattr(socket, family):
93	n/a	raise ValueError('Family %s is not recognized.' % family)
94	n/a
95	n/a	def address_type(address):
96	n/a	'''
97	n/a	Return the types of the address
98	n/a
99	n/a	This can be 'AF_INET', 'AF_UNIX', or 'AF_PIPE'
100	n/a	'''
101	n/a	if type(address) == tuple:
102	n/a	return 'AF_INET'
103	n/a	elif type(address) is str and address.startswith('\\\\'):
104	n/a	return 'AF_PIPE'
105	n/a	elif type(address) is str:
106	n/a	return 'AF_UNIX'
107	n/a	else:
108	n/a	raise ValueError('address type of %r unrecognized' % address)
109	n/a
110	n/a	#
111	n/a	# Connection classes
112	n/a	#
113	n/a
114	n/a	class _ConnectionBase:
115	n/a	_handle = None
116	n/a
117	n/a	def __init__(self, handle, readable=True, writable=True):
118	n/a	handle = handle.__index__()
119	n/a	if handle < 0:
120	n/a	raise ValueError("invalid handle")
121	n/a	if not readable and not writable:
122	n/a	raise ValueError(
123	n/a	"at least one of `readable` and `writable` must be True")
124	n/a	self._handle = handle
125	n/a	self._readable = readable
126	n/a	self._writable = writable
127	n/a
128	n/a	# XXX should we use util.Finalize instead of a __del__?
129	n/a
130	n/a	def __del__(self):
131	n/a	if self._handle is not None:
132	n/a	self._close()
133	n/a
134	n/a	def _check_closed(self):
135	n/a	if self._handle is None:
136	n/a	raise OSError("handle is closed")
137	n/a
138	n/a	def _check_readable(self):
139	n/a	if not self._readable:
140	n/a	raise OSError("connection is write-only")
141	n/a
142	n/a	def _check_writable(self):
143	n/a	if not self._writable:
144	n/a	raise OSError("connection is read-only")
145	n/a
146	n/a	def _bad_message_length(self):
147	n/a	if self._writable:
148	n/a	self._readable = False
149	n/a	else:
150	n/a	self.close()
151	n/a	raise OSError("bad message length")
152	n/a
153	n/a	@property
154	n/a	def closed(self):
155	n/a	"""True if the connection is closed"""
156	n/a	return self._handle is None
157	n/a
158	n/a	@property
159	n/a	def readable(self):
160	n/a	"""True if the connection is readable"""
161	n/a	return self._readable
162	n/a
163	n/a	@property
164	n/a	def writable(self):
165	n/a	"""True if the connection is writable"""
166	n/a	return self._writable
167	n/a
168	n/a	def fileno(self):
169	n/a	"""File descriptor or handle of the connection"""
170	n/a	self._check_closed()
171	n/a	return self._handle
172	n/a
173	n/a	def close(self):
174	n/a	"""Close the connection"""
175	n/a	if self._handle is not None:
176	n/a	try:
177	n/a	self._close()
178	n/a	finally:
179	n/a	self._handle = None
180	n/a
181	n/a	def send_bytes(self, buf, offset=0, size=None):
182	n/a	"""Send the bytes data from a bytes-like object"""
183	n/a	self._check_closed()
184	n/a	self._check_writable()
185	n/a	m = memoryview(buf)
186	n/a	# HACK for byte-indexing of non-bytewise buffers (e.g. array.array)
187	n/a	if m.itemsize > 1:
188	n/a	m = memoryview(bytes(m))
189	n/a	n = len(m)
190	n/a	if offset < 0:
191	n/a	raise ValueError("offset is negative")
192	n/a	if n < offset:
193	n/a	raise ValueError("buffer length < offset")
194	n/a	if size is None:
195	n/a	size = n - offset
196	n/a	elif size < 0:
197	n/a	raise ValueError("size is negative")
198	n/a	elif offset + size > n:
199	n/a	raise ValueError("buffer length < offset + size")
200	n/a	self._send_bytes(m[offset:offset + size])
201	n/a
202	n/a	def send(self, obj):
203	n/a	"""Send a (picklable) object"""
204	n/a	self._check_closed()
205	n/a	self._check_writable()
206	n/a	self._send_bytes(_ForkingPickler.dumps(obj))
207	n/a
208	n/a	def recv_bytes(self, maxlength=None):
209	n/a	"""
210	n/a	Receive bytes data as a bytes object.
211	n/a	"""
212	n/a	self._check_closed()
213	n/a	self._check_readable()
214	n/a	if maxlength is not None and maxlength < 0:
215	n/a	raise ValueError("negative maxlength")
216	n/a	buf = self._recv_bytes(maxlength)
217	n/a	if buf is None:
218	n/a	self._bad_message_length()
219	n/a	return buf.getvalue()
220	n/a
221	n/a	def recv_bytes_into(self, buf, offset=0):
222	n/a	"""
223	n/a	Receive bytes data into a writeable bytes-like object.
224	n/a	Return the number of bytes read.
225	n/a	"""
226	n/a	self._check_closed()
227	n/a	self._check_readable()
228	n/a	with memoryview(buf) as m:
229	n/a	# Get bytesize of arbitrary buffer
230	n/a	itemsize = m.itemsize
231	n/a	bytesize = itemsize * len(m)
232	n/a	if offset < 0:
233	n/a	raise ValueError("negative offset")
234	n/a	elif offset > bytesize:
235	n/a	raise ValueError("offset too large")
236	n/a	result = self._recv_bytes()
237	n/a	size = result.tell()
238	n/a	if bytesize < offset + size:
239	n/a	raise BufferTooShort(result.getvalue())
240	n/a	# Message can fit in dest
241	n/a	result.seek(0)
242	n/a	result.readinto(m[offset // itemsize :
243	n/a	(offset + size) // itemsize])
244	n/a	return size
245	n/a
246	n/a	def recv(self):
247	n/a	"""Receive a (picklable) object"""
248	n/a	self._check_closed()
249	n/a	self._check_readable()
250	n/a	buf = self._recv_bytes()
251	n/a	return _ForkingPickler.loads(buf.getbuffer())
252	n/a
253	n/a	def poll(self, timeout=0.0):
254	n/a	"""Whether there is any input available to be read"""
255	n/a	self._check_closed()
256	n/a	self._check_readable()
257	n/a	return self._poll(timeout)
258	n/a
259	n/a	def __enter__(self):
260	n/a	return self
261	n/a
262	n/a	def __exit__(self, exc_type, exc_value, exc_tb):
263	n/a	self.close()
264	n/a
265	n/a
266	n/a	if _winapi:
267	n/a
268	n/a	class PipeConnection(_ConnectionBase):
269	n/a	"""
270	n/a	Connection class based on a Windows named pipe.
271	n/a	Overlapped I/O is used, so the handles must have been created
272	n/a	with FILE_FLAG_OVERLAPPED.
273	n/a	"""
274	n/a	_got_empty_message = False
275	n/a
276	n/a	def _close(self, _CloseHandle=_winapi.CloseHandle):
277	n/a	_CloseHandle(self._handle)
278	n/a
279	n/a	def _send_bytes(self, buf):
280	n/a	ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
281	n/a	try:
282	n/a	if err == _winapi.ERROR_IO_PENDING:
283	n/a	waitres = _winapi.WaitForMultipleObjects(
284	n/a	[ov.event], False, INFINITE)
285	n/a	assert waitres == WAIT_OBJECT_0
286	n/a	except:
287	n/a	ov.cancel()
288	n/a	raise
289	n/a	finally:
290	n/a	nwritten, err = ov.GetOverlappedResult(True)
291	n/a	assert err == 0
292	n/a	assert nwritten == len(buf)
293	n/a
294	n/a	def _recv_bytes(self, maxsize=None):
295	n/a	if self._got_empty_message:
296	n/a	self._got_empty_message = False
297	n/a	return io.BytesIO()
298	n/a	else:
299	n/a	bsize = 128 if maxsize is None else min(maxsize, 128)
300	n/a	try:
301	n/a	ov, err = _winapi.ReadFile(self._handle, bsize,
302	n/a	overlapped=True)
303	n/a	try:
304	n/a	if err == _winapi.ERROR_IO_PENDING:
305	n/a	waitres = _winapi.WaitForMultipleObjects(
306	n/a	[ov.event], False, INFINITE)
307	n/a	assert waitres == WAIT_OBJECT_0
308	n/a	except:
309	n/a	ov.cancel()
310	n/a	raise
311	n/a	finally:
312	n/a	nread, err = ov.GetOverlappedResult(True)
313	n/a	if err == 0:
314	n/a	f = io.BytesIO()
315	n/a	f.write(ov.getbuffer())
316	n/a	return f
317	n/a	elif err == _winapi.ERROR_MORE_DATA:
318	n/a	return self._get_more_data(ov, maxsize)
319	n/a	except OSError as e:
320	n/a	if e.winerror == _winapi.ERROR_BROKEN_PIPE:
321	n/a	raise EOFError
322	n/a	else:
323	n/a	raise
324	n/a	raise RuntimeError("shouldn't get here; expected KeyboardInterrupt")
325	n/a
326	n/a	def _poll(self, timeout):
327	n/a	if (self._got_empty_message or
328	n/a	_winapi.PeekNamedPipe(self._handle)[0] != 0):
329	n/a	return True
330	n/a	return bool(wait([self], timeout))
331	n/a
332	n/a	def _get_more_data(self, ov, maxsize):
333	n/a	buf = ov.getbuffer()
334	n/a	f = io.BytesIO()
335	n/a	f.write(buf)
336	n/a	left = _winapi.PeekNamedPipe(self._handle)[1]
337	n/a	assert left > 0
338	n/a	if maxsize is not None and len(buf) + left > maxsize:
339	n/a	self._bad_message_length()
340	n/a	ov, err = _winapi.ReadFile(self._handle, left, overlapped=True)
341	n/a	rbytes, err = ov.GetOverlappedResult(True)
342	n/a	assert err == 0
343	n/a	assert rbytes == left
344	n/a	f.write(ov.getbuffer())
345	n/a	return f
346	n/a
347	n/a
348	n/a	class Connection(_ConnectionBase):
349	n/a	"""
350	n/a	Connection class based on an arbitrary file descriptor (Unix only), or
351	n/a	a socket handle (Windows).
352	n/a	"""
353	n/a
354	n/a	if _winapi:
355	n/a	def _close(self, _close=_multiprocessing.closesocket):
356	n/a	_close(self._handle)
357	n/a	_write = _multiprocessing.send
358	n/a	_read = _multiprocessing.recv
359	n/a	else:
360	n/a	def _close(self, _close=os.close):
361	n/a	_close(self._handle)
362	n/a	_write = os.write
363	n/a	_read = os.read
364	n/a
365	n/a	def _send(self, buf, write=_write):
366	n/a	remaining = len(buf)
367	n/a	while True:
368	n/a	n = write(self._handle, buf)
369	n/a	remaining -= n
370	n/a	if remaining == 0:
371	n/a	break
372	n/a	buf = buf[n:]
373	n/a
374	n/a	def _recv(self, size, read=_read):
375	n/a	buf = io.BytesIO()
376	n/a	handle = self._handle
377	n/a	remaining = size
378	n/a	while remaining > 0:
379	n/a	chunk = read(handle, remaining)
380	n/a	n = len(chunk)
381	n/a	if n == 0:
382	n/a	if remaining == size:
383	n/a	raise EOFError
384	n/a	else:
385	n/a	raise OSError("got end of file during message")
386	n/a	buf.write(chunk)
387	n/a	remaining -= n
388	n/a	return buf
389	n/a
390	n/a	def _send_bytes(self, buf):
391	n/a	n = len(buf)
392	n/a	# For wire compatibility with 3.2 and lower
393	n/a	header = struct.pack("!i", n)
394	n/a	if n > 16384:
395	n/a	# The payload is large so Nagle's algorithm won't be triggered
396	n/a	# and we'd better avoid the cost of concatenation.
397	n/a	self._send(header)
398	n/a	self._send(buf)
399	n/a	else:
400	n/a	# Issue #20540: concatenate before sending, to avoid delays due
401	n/a	# to Nagle's algorithm on a TCP socket.
402	n/a	# Also note we want to avoid sending a 0-length buffer separately,
403	n/a	# to avoid "broken pipe" errors if the other end closed the pipe.
404	n/a	self._send(header + buf)
405	n/a
406	n/a	def _recv_bytes(self, maxsize=None):
407	n/a	buf = self._recv(4)
408	n/a	size, = struct.unpack("!i", buf.getvalue())
409	n/a	if maxsize is not None and size > maxsize:
410	n/a	return None
411	n/a	return self._recv(size)
412	n/a
413	n/a	def _poll(self, timeout):
414	n/a	r = wait([self], timeout)
415	n/a	return bool(r)
416	n/a
417	n/a
418	n/a	#
419	n/a	# Public functions
420	n/a	#
421	n/a
422	n/a	class Listener(object):
423	n/a	'''
424	n/a	Returns a listener object.
425	n/a
426	n/a	This is a wrapper for a bound socket which is 'listening' for
427	n/a	connections, or for a Windows named pipe.
428	n/a	'''
429	n/a	def __init__(self, address=None, family=None, backlog=1, authkey=None):
430	n/a	family = family or (address and address_type(address)) \
431	n/a	or default_family
432	n/a	address = address or arbitrary_address(family)
433	n/a
434	n/a	_validate_family(family)
435	n/a	if family == 'AF_PIPE':
436	n/a	self._listener = PipeListener(address, backlog)
437	n/a	else:
438	n/a	self._listener = SocketListener(address, family, backlog)
439	n/a
440	n/a	if authkey is not None and not isinstance(authkey, bytes):
441	n/a	raise TypeError('authkey should be a byte string')
442	n/a
443	n/a	self._authkey = authkey
444	n/a
445	n/a	def accept(self):
446	n/a	'''
447	n/a	Accept a connection on the bound socket or named pipe of `self`.
448	n/a
449	n/a	Returns a `Connection` object.
450	n/a	'''
451	n/a	if self._listener is None:
452	n/a	raise OSError('listener is closed')
453	n/a	c = self._listener.accept()
454	n/a	if self._authkey:
455	n/a	deliver_challenge(c, self._authkey)
456	n/a	answer_challenge(c, self._authkey)
457	n/a	return c
458	n/a
459	n/a	def close(self):
460	n/a	'''
461	n/a	Close the bound socket or named pipe of `self`.
462	n/a	'''
463	n/a	listener = self._listener
464	n/a	if listener is not None:
465	n/a	self._listener = None
466	n/a	listener.close()
467	n/a
468	n/a	address = property(lambda self: self._listener._address)
469	n/a	last_accepted = property(lambda self: self._listener._last_accepted)
470	n/a
471	n/a	def __enter__(self):
472	n/a	return self
473	n/a
474	n/a	def __exit__(self, exc_type, exc_value, exc_tb):
475	n/a	self.close()
476	n/a
477	n/a
478	n/a	def Client(address, family=None, authkey=None):
479	n/a	'''
480	n/a	Returns a connection to the address of a `Listener`
481	n/a	'''
482	n/a	family = family or address_type(address)
483	n/a	_validate_family(family)
484	n/a	if family == 'AF_PIPE':
485	n/a	c = PipeClient(address)
486	n/a	else:
487	n/a	c = SocketClient(address)
488	n/a
489	n/a	if authkey is not None and not isinstance(authkey, bytes):
490	n/a	raise TypeError('authkey should be a byte string')
491	n/a
492	n/a	if authkey is not None:
493	n/a	answer_challenge(c, authkey)
494	n/a	deliver_challenge(c, authkey)
495	n/a
496	n/a	return c
497	n/a
498	n/a
499	n/a	if sys.platform != 'win32':
500	n/a
501	n/a	def Pipe(duplex=True):
502	n/a	'''
503	n/a	Returns pair of connection objects at either end of a pipe
504	n/a	'''
505	n/a	if duplex:
506	n/a	s1, s2 = socket.socketpair()
507	n/a	s1.setblocking(True)
508	n/a	s2.setblocking(True)
509	n/a	c1 = Connection(s1.detach())
510	n/a	c2 = Connection(s2.detach())
511	n/a	else:
512	n/a	fd1, fd2 = os.pipe()
513	n/a	c1 = Connection(fd1, writable=False)
514	n/a	c2 = Connection(fd2, readable=False)
515	n/a
516	n/a	return c1, c2
517	n/a
518	n/a	else:
519	n/a
520	n/a	def Pipe(duplex=True):
521	n/a	'''
522	n/a	Returns pair of connection objects at either end of a pipe
523	n/a	'''
524	n/a	address = arbitrary_address('AF_PIPE')
525	n/a	if duplex:
526	n/a	openmode = _winapi.PIPE_ACCESS_DUPLEX
527	n/a	access = _winapi.GENERIC_READ \| _winapi.GENERIC_WRITE
528	n/a	obsize, ibsize = BUFSIZE, BUFSIZE
529	n/a	else:
530	n/a	openmode = _winapi.PIPE_ACCESS_INBOUND
531	n/a	access = _winapi.GENERIC_WRITE
532	n/a	obsize, ibsize = 0, BUFSIZE
533	n/a
534	n/a	h1 = _winapi.CreateNamedPipe(
535	n/a	address, openmode \| _winapi.FILE_FLAG_OVERLAPPED \|
536	n/a	_winapi.FILE_FLAG_FIRST_PIPE_INSTANCE,
537	n/a	_winapi.PIPE_TYPE_MESSAGE \| _winapi.PIPE_READMODE_MESSAGE \|
538	n/a	_winapi.PIPE_WAIT,
539	n/a	1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER,
540	n/a	# default security descriptor: the handle cannot be inherited
541	n/a	_winapi.NULL
542	n/a	)
543	n/a	h2 = _winapi.CreateFile(
544	n/a	address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
545	n/a	_winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
546	n/a	)
547	n/a	_winapi.SetNamedPipeHandleState(
548	n/a	h2, _winapi.PIPE_READMODE_MESSAGE, None, None
549	n/a	)
550	n/a
551	n/a	overlapped = _winapi.ConnectNamedPipe(h1, overlapped=True)
552	n/a	_, err = overlapped.GetOverlappedResult(True)
553	n/a	assert err == 0
554	n/a
555	n/a	c1 = PipeConnection(h1, writable=duplex)
556	n/a	c2 = PipeConnection(h2, readable=duplex)
557	n/a
558	n/a	return c1, c2
559	n/a
560	n/a	#
561	n/a	# Definitions for connections based on sockets
562	n/a	#
563	n/a
564	n/a	class SocketListener(object):
565	n/a	'''
566	n/a	Representation of a socket which is bound to an address and listening
567	n/a	'''
568	n/a	def __init__(self, address, family, backlog=1):
569	n/a	self._socket = socket.socket(getattr(socket, family))
570	n/a	try:
571	n/a	# SO_REUSEADDR has different semantics on Windows (issue #2550).
572	n/a	if os.name == 'posix':
573	n/a	self._socket.setsockopt(socket.SOL_SOCKET,
574	n/a	socket.SO_REUSEADDR, 1)
575	n/a	self._socket.setblocking(True)
576	n/a	self._socket.bind(address)
577	n/a	self._socket.listen(backlog)
578	n/a	self._address = self._socket.getsockname()
579	n/a	except OSError:
580	n/a	self._socket.close()
581	n/a	raise
582	n/a	self._family = family
583	n/a	self._last_accepted = None
584	n/a
585	n/a	if family == 'AF_UNIX':
586	n/a	self._unlink = util.Finalize(
587	n/a	self, os.unlink, args=(address,), exitpriority=0
588	n/a	)
589	n/a	else:
590	n/a	self._unlink = None
591	n/a
592	n/a	def accept(self):
593	n/a	s, self._last_accepted = self._socket.accept()
594	n/a	s.setblocking(True)
595	n/a	return Connection(s.detach())
596	n/a
597	n/a	def close(self):
598	n/a	try:
599	n/a	self._socket.close()
600	n/a	finally:
601	n/a	unlink = self._unlink
602	n/a	if unlink is not None:
603	n/a	self._unlink = None
604	n/a	unlink()
605	n/a
606	n/a
607	n/a	def SocketClient(address):
608	n/a	'''
609	n/a	Return a connection object connected to the socket given by `address`
610	n/a	'''
611	n/a	family = address_type(address)
612	n/a	with socket.socket( getattr(socket, family) ) as s:
613	n/a	s.setblocking(True)
614	n/a	s.connect(address)
615	n/a	return Connection(s.detach())
616	n/a
617	n/a	#
618	n/a	# Definitions for connections based on named pipes
619	n/a	#
620	n/a
621	n/a	if sys.platform == 'win32':
622	n/a
623	n/a	class PipeListener(object):
624	n/a	'''
625	n/a	Representation of a named pipe
626	n/a	'''
627	n/a	def __init__(self, address, backlog=None):
628	n/a	self._address = address
629	n/a	self._handle_queue = [self._new_handle(first=True)]
630	n/a
631	n/a	self._last_accepted = None
632	n/a	util.sub_debug('listener created with address=%r', self._address)
633	n/a	self.close = util.Finalize(
634	n/a	self, PipeListener._finalize_pipe_listener,
635	n/a	args=(self._handle_queue, self._address), exitpriority=0
636	n/a	)
637	n/a
638	n/a	def _new_handle(self, first=False):
639	n/a	flags = _winapi.PIPE_ACCESS_DUPLEX \| _winapi.FILE_FLAG_OVERLAPPED
640	n/a	if first:
641	n/a	flags \|= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
642	n/a	return _winapi.CreateNamedPipe(
643	n/a	self._address, flags,
644	n/a	_winapi.PIPE_TYPE_MESSAGE \| _winapi.PIPE_READMODE_MESSAGE \|
645	n/a	_winapi.PIPE_WAIT,
646	n/a	_winapi.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
647	n/a	_winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
648	n/a	)
649	n/a
650	n/a	def accept(self):
651	n/a	self._handle_queue.append(self._new_handle())
652	n/a	handle = self._handle_queue.pop(0)
653	n/a	try:
654	n/a	ov = _winapi.ConnectNamedPipe(handle, overlapped=True)
655	n/a	except OSError as e:
656	n/a	if e.winerror != _winapi.ERROR_NO_DATA:
657	n/a	raise
658	n/a	# ERROR_NO_DATA can occur if a client has already connected,
659	n/a	# written data and then disconnected -- see Issue 14725.
660	n/a	else:
661	n/a	try:
662	n/a	res = _winapi.WaitForMultipleObjects(
663	n/a	[ov.event], False, INFINITE)
664	n/a	except:
665	n/a	ov.cancel()
666	n/a	_winapi.CloseHandle(handle)
667	n/a	raise
668	n/a	finally:
669	n/a	_, err = ov.GetOverlappedResult(True)
670	n/a	assert err == 0
671	n/a	return PipeConnection(handle)
672	n/a
673	n/a	@staticmethod
674	n/a	def _finalize_pipe_listener(queue, address):
675	n/a	util.sub_debug('closing listener with address=%r', address)
676	n/a	for handle in queue:
677	n/a	_winapi.CloseHandle(handle)
678	n/a
679	n/a	def PipeClient(address):
680	n/a	'''
681	n/a	Return a connection object connected to the pipe given by `address`
682	n/a	'''
683	n/a	t = _init_timeout()
684	n/a	while 1:
685	n/a	try:
686	n/a	_winapi.WaitNamedPipe(address, 1000)
687	n/a	h = _winapi.CreateFile(
688	n/a	address, _winapi.GENERIC_READ \| _winapi.GENERIC_WRITE,
689	n/a	0, _winapi.NULL, _winapi.OPEN_EXISTING,
690	n/a	_winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
691	n/a	)
692	n/a	except OSError as e:
693	n/a	if e.winerror not in (_winapi.ERROR_SEM_TIMEOUT,
694	n/a	_winapi.ERROR_PIPE_BUSY) or _check_timeout(t):
695	n/a	raise
696	n/a	else:
697	n/a	break
698	n/a	else:
699	n/a	raise
700	n/a
701	n/a	_winapi.SetNamedPipeHandleState(
702	n/a	h, _winapi.PIPE_READMODE_MESSAGE, None, None
703	n/a	)
704	n/a	return PipeConnection(h)
705	n/a
706	n/a	#
707	n/a	# Authentication stuff
708	n/a	#
709	n/a
710	n/a	MESSAGE_LENGTH = 20
711	n/a
712	n/a	CHALLENGE = b'#CHALLENGE#'
713	n/a	WELCOME = b'#WELCOME#'
714	n/a	FAILURE = b'#FAILURE#'
715	n/a
716	n/a	def deliver_challenge(connection, authkey):
717	n/a	import hmac
718	n/a	assert isinstance(authkey, bytes)
719	n/a	message = os.urandom(MESSAGE_LENGTH)
720	n/a	connection.send_bytes(CHALLENGE + message)
721	n/a	digest = hmac.new(authkey, message, 'md5').digest()
722	n/a	response = connection.recv_bytes(256) # reject large message
723	n/a	if response == digest:
724	n/a	connection.send_bytes(WELCOME)
725	n/a	else:
726	n/a	connection.send_bytes(FAILURE)
727	n/a	raise AuthenticationError('digest received was wrong')
728	n/a
729	n/a	def answer_challenge(connection, authkey):
730	n/a	import hmac
731	n/a	assert isinstance(authkey, bytes)
732	n/a	message = connection.recv_bytes(256) # reject large message
733	n/a	assert message[:len(CHALLENGE)] == CHALLENGE, 'message = %r' % message
734	n/a	message = message[len(CHALLENGE):]
735	n/a	digest = hmac.new(authkey, message, 'md5').digest()
736	n/a	connection.send_bytes(digest)
737	n/a	response = connection.recv_bytes(256) # reject large message
738	n/a	if response != WELCOME:
739	n/a	raise AuthenticationError('digest sent was rejected')
740	n/a
741	n/a	#
742	n/a	# Support for using xmlrpclib for serialization
743	n/a	#
744	n/a
745	n/a	class ConnectionWrapper(object):
746	n/a	def __init__(self, conn, dumps, loads):
747	n/a	self._conn = conn
748	n/a	self._dumps = dumps
749	n/a	self._loads = loads
750	n/a	for attr in ('fileno', 'close', 'poll', 'recv_bytes', 'send_bytes'):
751	n/a	obj = getattr(conn, attr)
752	n/a	setattr(self, attr, obj)
753	n/a	def send(self, obj):
754	n/a	s = self._dumps(obj)
755	n/a	self._conn.send_bytes(s)
756	n/a	def recv(self):
757	n/a	s = self._conn.recv_bytes()
758	n/a	return self._loads(s)
759	n/a
760	n/a	def _xml_dumps(obj):
761	n/a	return xmlrpclib.dumps((obj,), None, None, None, 1).encode('utf-8')
762	n/a
763	n/a	def _xml_loads(s):
764	n/a	(obj,), method = xmlrpclib.loads(s.decode('utf-8'))
765	n/a	return obj
766	n/a
767	n/a	class XmlListener(Listener):
768	n/a	def accept(self):
769	n/a	global xmlrpclib
770	n/a	import xmlrpc.client as xmlrpclib
771	n/a	obj = Listener.accept(self)
772	n/a	return ConnectionWrapper(obj, _xml_dumps, _xml_loads)
773	n/a
774	n/a	def XmlClient(args, *kwds):
775	n/a	global xmlrpclib
776	n/a	import xmlrpc.client as xmlrpclib
777	n/a	return ConnectionWrapper(Client(args, *kwds), _xml_dumps, _xml_loads)
778	n/a
779	n/a	#
780	n/a	# Wait
781	n/a	#
782	n/a
783	n/a	if sys.platform == 'win32':
784	n/a
785	n/a	def _exhaustive_wait(handles, timeout):
786	n/a	# Return ALL handles which are currently signalled. (Only
787	n/a	# returning the first signalled might create starvation issues.)
788	n/a	L = list(handles)
789	n/a	ready = []
790	n/a	while L:
791	n/a	res = _winapi.WaitForMultipleObjects(L, False, timeout)
792	n/a	if res == WAIT_TIMEOUT:
793	n/a	break
794	n/a	elif WAIT_OBJECT_0 <= res < WAIT_OBJECT_0 + len(L):
795	n/a	res -= WAIT_OBJECT_0
796	n/a	elif WAIT_ABANDONED_0 <= res < WAIT_ABANDONED_0 + len(L):
797	n/a	res -= WAIT_ABANDONED_0
798	n/a	else:
799	n/a	raise RuntimeError('Should not get here')
800	n/a	ready.append(L[res])
801	n/a	L = L[res+1:]
802	n/a	timeout = 0
803	n/a	return ready
804	n/a
805	n/a	_ready_errors = {_winapi.ERROR_BROKEN_PIPE, _winapi.ERROR_NETNAME_DELETED}
806	n/a
807	n/a	def wait(object_list, timeout=None):
808	n/a	'''
809	n/a	Wait till an object in object_list is ready/readable.
810	n/a
811	n/a	Returns list of those objects in object_list which are ready/readable.
812	n/a	'''
813	n/a	if timeout is None:
814	n/a	timeout = INFINITE
815	n/a	elif timeout < 0:
816	n/a	timeout = 0
817	n/a	else:
818	n/a	timeout = int(timeout * 1000 + 0.5)
819	n/a
820	n/a	object_list = list(object_list)
821	n/a	waithandle_to_obj = {}
822	n/a	ov_list = []
823	n/a	ready_objects = set()
824	n/a	ready_handles = set()
825	n/a
826	n/a	try:
827	n/a	for o in object_list:
828	n/a	try:
829	n/a	fileno = getattr(o, 'fileno')
830	n/a	except AttributeError:
831	n/a	waithandle_to_obj[o.__index__()] = o
832	n/a	else:
833	n/a	# start an overlapped read of length zero
834	n/a	try:
835	n/a	ov, err = _winapi.ReadFile(fileno(), 0, True)
836	n/a	except OSError as e:
837	n/a	ov, err = None, e.winerror
838	n/a	if err not in _ready_errors:
839	n/a	raise
840	n/a	if err == _winapi.ERROR_IO_PENDING:
841	n/a	ov_list.append(ov)
842	n/a	waithandle_to_obj[ov.event] = o
843	n/a	else:
844	n/a	# If o.fileno() is an overlapped pipe handle and
845	n/a	# err == 0 then there is a zero length message
846	n/a	# in the pipe, but it HAS NOT been consumed...
847	n/a	if ov and sys.getwindowsversion()[:2] >= (6, 2):
848	n/a	# ... except on Windows 8 and later, where
849	n/a	# the message HAS been consumed.
850	n/a	try:
851	n/a	_, err = ov.GetOverlappedResult(False)
852	n/a	except OSError as e:
853	n/a	err = e.winerror
854	n/a	if not err and hasattr(o, '_got_empty_message'):
855	n/a	o._got_empty_message = True
856	n/a	ready_objects.add(o)
857	n/a	timeout = 0
858	n/a
859	n/a	ready_handles = _exhaustive_wait(waithandle_to_obj.keys(), timeout)
860	n/a	finally:
861	n/a	# request that overlapped reads stop
862	n/a	for ov in ov_list:
863	n/a	ov.cancel()
864	n/a
865	n/a	# wait for all overlapped reads to stop
866	n/a	for ov in ov_list:
867	n/a	try:
868	n/a	_, err = ov.GetOverlappedResult(True)
869	n/a	except OSError as e:
870	n/a	err = e.winerror
871	n/a	if err not in _ready_errors:
872	n/a	raise
873	n/a	if err != _winapi.ERROR_OPERATION_ABORTED:
874	n/a	o = waithandle_to_obj[ov.event]
875	n/a	ready_objects.add(o)
876	n/a	if err == 0:
877	n/a	# If o.fileno() is an overlapped pipe handle then
878	n/a	# a zero length message HAS been consumed.
879	n/a	if hasattr(o, '_got_empty_message'):
880	n/a	o._got_empty_message = True
881	n/a
882	n/a	ready_objects.update(waithandle_to_obj[h] for h in ready_handles)
883	n/a	return [o for o in object_list if o in ready_objects]
884	n/a
885	n/a	else:
886	n/a
887	n/a	import selectors
888	n/a
889	n/a	# poll/select have the advantage of not requiring any extra file
890	n/a	# descriptor, contrarily to epoll/kqueue (also, they require a single
891	n/a	# syscall).
892	n/a	if hasattr(selectors, 'PollSelector'):
893	n/a	_WaitSelector = selectors.PollSelector
894	n/a	else:
895	n/a	_WaitSelector = selectors.SelectSelector
896	n/a
897	n/a	def wait(object_list, timeout=None):
898	n/a	'''
899	n/a	Wait till an object in object_list is ready/readable.
900	n/a
901	n/a	Returns list of those objects in object_list which are ready/readable.
902	n/a	'''
903	n/a	with _WaitSelector() as selector:
904	n/a	for obj in object_list:
905	n/a	selector.register(obj, selectors.EVENT_READ)
906	n/a
907	n/a	if timeout is not None:
908	n/a	deadline = time.time() + timeout
909	n/a
910	n/a	while True:
911	n/a	ready = selector.select(timeout)
912	n/a	if ready:
913	n/a	return [key.fileobj for (key, events) in ready]
914	n/a	else:
915	n/a	if timeout is not None:
916	n/a	timeout = deadline - time.time()
917	n/a	if timeout < 0:
918	n/a	return ready
919	n/a
920	n/a	#
921	n/a	# Make connection and socket objects sharable if possible
922	n/a	#
923	n/a
924	n/a	if sys.platform == 'win32':
925	n/a	def reduce_connection(conn):
926	n/a	handle = conn.fileno()
927	n/a	with socket.fromfd(handle, socket.AF_INET, socket.SOCK_STREAM) as s:
928	n/a	from . import resource_sharer
929	n/a	ds = resource_sharer.DupSocket(s)
930	n/a	return rebuild_connection, (ds, conn.readable, conn.writable)
931	n/a	def rebuild_connection(ds, readable, writable):
932	n/a	sock = ds.detach()
933	n/a	return Connection(sock.detach(), readable, writable)
934	n/a	reduction.register(Connection, reduce_connection)
935	n/a
936	n/a	def reduce_pipe_connection(conn):
937	n/a	access = ((_winapi.FILE_GENERIC_READ if conn.readable else 0) \|
938	n/a	(_winapi.FILE_GENERIC_WRITE if conn.writable else 0))
939	n/a	dh = reduction.DupHandle(conn.fileno(), access)
940	n/a	return rebuild_pipe_connection, (dh, conn.readable, conn.writable)
941	n/a	def rebuild_pipe_connection(dh, readable, writable):
942	n/a	handle = dh.detach()
943	n/a	return PipeConnection(handle, readable, writable)
944	n/a	reduction.register(PipeConnection, reduce_pipe_connection)
945	n/a
946	n/a	else:
947	n/a	def reduce_connection(conn):
948	n/a	df = reduction.DupFd(conn.fileno())
949	n/a	return rebuild_connection, (df, conn.readable, conn.writable)
950	n/a	def rebuild_connection(df, readable, writable):
951	n/a	fd = df.detach()
952	n/a	return Connection(fd, readable, writable)
953	n/a	reduction.register(Connection, reduce_connection)