Python code coverage for Lib/asyncio/windows_events.py

#	count	content
1	n/a	"""Selector and proactor event loops for Windows."""
2	n/a
3	n/a	import _winapi
4	n/a	import errno
5	n/a	import math
6	n/a	import socket
7	n/a	import struct
8	n/a	import weakref
9	n/a
10	n/a	from . import events
11	n/a	from . import base_subprocess
12	n/a	from . import futures
13	n/a	from . import proactor_events
14	n/a	from . import selector_events
15	n/a	from . import tasks
16	n/a	from . import windows_utils
17	n/a	from . import _overlapped
18	n/a	from .coroutines import coroutine
19	n/a	from .log import logger
20	n/a
21	n/a
22	n/a	__all__ = ['SelectorEventLoop', 'ProactorEventLoop', 'IocpProactor',
23	n/a	'DefaultEventLoopPolicy',
24	n/a	]
25	n/a
26	n/a
27	n/a	NULL = 0
28	n/a	INFINITE = 0xffffffff
29	n/a	ERROR_CONNECTION_REFUSED = 1225
30	n/a	ERROR_CONNECTION_ABORTED = 1236
31	n/a
32	n/a	# Initial delay in seconds for connect_pipe() before retrying to connect
33	n/a	CONNECT_PIPE_INIT_DELAY = 0.001
34	n/a
35	n/a	# Maximum delay in seconds for connect_pipe() before retrying to connect
36	n/a	CONNECT_PIPE_MAX_DELAY = 0.100
37	n/a
38	n/a
39	n/a	class _OverlappedFuture(futures.Future):
40	n/a	"""Subclass of Future which represents an overlapped operation.
41	n/a
42	n/a	Cancelling it will immediately cancel the overlapped operation.
43	n/a	"""
44	n/a
45	n/a	def __init__(self, ov, *, loop=None):
46	n/a	super().__init__(loop=loop)
47	n/a	if self._source_traceback:
48	n/a	del self._source_traceback[-1]
49	n/a	self._ov = ov
50	n/a
51	n/a	def _repr_info(self):
52	n/a	info = super()._repr_info()
53	n/a	if self._ov is not None:
54	n/a	state = 'pending' if self._ov.pending else 'completed'
55	n/a	info.insert(1, 'overlapped=<%s, %#x>' % (state, self._ov.address))
56	n/a	return info
57	n/a
58	n/a	def _cancel_overlapped(self):
59	n/a	if self._ov is None:
60	n/a	return
61	n/a	try:
62	n/a	self._ov.cancel()
63	n/a	except OSError as exc:
64	n/a	context = {
65	n/a	'message': 'Cancelling an overlapped future failed',
66	n/a	'exception': exc,
67	n/a	'future': self,
68	n/a	}
69	n/a	if self._source_traceback:
70	n/a	context['source_traceback'] = self._source_traceback
71	n/a	self._loop.call_exception_handler(context)
72	n/a	self._ov = None
73	n/a
74	n/a	def cancel(self):
75	n/a	self._cancel_overlapped()
76	n/a	return super().cancel()
77	n/a
78	n/a	def set_exception(self, exception):
79	n/a	super().set_exception(exception)
80	n/a	self._cancel_overlapped()
81	n/a
82	n/a	def set_result(self, result):
83	n/a	super().set_result(result)
84	n/a	self._ov = None
85	n/a
86	n/a
87	n/a	class _BaseWaitHandleFuture(futures.Future):
88	n/a	"""Subclass of Future which represents a wait handle."""
89	n/a
90	n/a	def __init__(self, ov, handle, wait_handle, *, loop=None):
91	n/a	super().__init__(loop=loop)
92	n/a	if self._source_traceback:
93	n/a	del self._source_traceback[-1]
94	n/a	# Keep a reference to the Overlapped object to keep it alive until the
95	n/a	# wait is unregistered
96	n/a	self._ov = ov
97	n/a	self._handle = handle
98	n/a	self._wait_handle = wait_handle
99	n/a
100	n/a	# Should we call UnregisterWaitEx() if the wait completes
101	n/a	# or is cancelled?
102	n/a	self._registered = True
103	n/a
104	n/a	def _poll(self):
105	n/a	# non-blocking wait: use a timeout of 0 millisecond
106	n/a	return (_winapi.WaitForSingleObject(self._handle, 0) ==
107	n/a	_winapi.WAIT_OBJECT_0)
108	n/a
109	n/a	def _repr_info(self):
110	n/a	info = super()._repr_info()
111	n/a	info.append('handle=%#x' % self._handle)
112	n/a	if self._handle is not None:
113	n/a	state = 'signaled' if self._poll() else 'waiting'
114	n/a	info.append(state)
115	n/a	if self._wait_handle is not None:
116	n/a	info.append('wait_handle=%#x' % self._wait_handle)
117	n/a	return info
118	n/a
119	n/a	def _unregister_wait_cb(self, fut):
120	n/a	# The wait was unregistered: it's not safe to destroy the Overlapped
121	n/a	# object
122	n/a	self._ov = None
123	n/a
124	n/a	def _unregister_wait(self):
125	n/a	if not self._registered:
126	n/a	return
127	n/a	self._registered = False
128	n/a
129	n/a	wait_handle = self._wait_handle
130	n/a	self._wait_handle = None
131	n/a	try:
132	n/a	_overlapped.UnregisterWait(wait_handle)
133	n/a	except OSError as exc:
134	n/a	if exc.winerror != _overlapped.ERROR_IO_PENDING:
135	n/a	context = {
136	n/a	'message': 'Failed to unregister the wait handle',
137	n/a	'exception': exc,
138	n/a	'future': self,
139	n/a	}
140	n/a	if self._source_traceback:
141	n/a	context['source_traceback'] = self._source_traceback
142	n/a	self._loop.call_exception_handler(context)
143	n/a	return
144	n/a	# ERROR_IO_PENDING means that the unregister is pending
145	n/a
146	n/a	self._unregister_wait_cb(None)
147	n/a
148	n/a	def cancel(self):
149	n/a	self._unregister_wait()
150	n/a	return super().cancel()
151	n/a
152	n/a	def set_exception(self, exception):
153	n/a	self._unregister_wait()
154	n/a	super().set_exception(exception)
155	n/a
156	n/a	def set_result(self, result):
157	n/a	self._unregister_wait()
158	n/a	super().set_result(result)
159	n/a
160	n/a
161	n/a	class _WaitCancelFuture(_BaseWaitHandleFuture):
162	n/a	"""Subclass of Future which represents a wait for the cancellation of a
163	n/a	_WaitHandleFuture using an event.
164	n/a	"""
165	n/a
166	n/a	def __init__(self, ov, event, wait_handle, *, loop=None):
167	n/a	super().__init__(ov, event, wait_handle, loop=loop)
168	n/a
169	n/a	self._done_callback = None
170	n/a
171	n/a	def cancel(self):
172	n/a	raise RuntimeError("_WaitCancelFuture must not be cancelled")
173	n/a
174	n/a	def set_result(self, result):
175	n/a	super().set_result(result)
176	n/a	if self._done_callback is not None:
177	n/a	self._done_callback(self)
178	n/a
179	n/a	def set_exception(self, exception):
180	n/a	super().set_exception(exception)
181	n/a	if self._done_callback is not None:
182	n/a	self._done_callback(self)
183	n/a
184	n/a
185	n/a	class _WaitHandleFuture(_BaseWaitHandleFuture):
186	n/a	def __init__(self, ov, handle, wait_handle, proactor, *, loop=None):
187	n/a	super().__init__(ov, handle, wait_handle, loop=loop)
188	n/a	self._proactor = proactor
189	n/a	self._unregister_proactor = True
190	n/a	self._event = _overlapped.CreateEvent(None, True, False, None)
191	n/a	self._event_fut = None
192	n/a
193	n/a	def _unregister_wait_cb(self, fut):
194	n/a	if self._event is not None:
195	n/a	_winapi.CloseHandle(self._event)
196	n/a	self._event = None
197	n/a	self._event_fut = None
198	n/a
199	n/a	# If the wait was cancelled, the wait may never be signalled, so
200	n/a	# it's required to unregister it. Otherwise, IocpProactor.close() will
201	n/a	# wait forever for an event which will never come.
202	n/a	#
203	n/a	# If the IocpProactor already received the event, it's safe to call
204	n/a	# _unregister() because we kept a reference to the Overlapped object
205	n/a	# which is used as a unique key.
206	n/a	self._proactor._unregister(self._ov)
207	n/a	self._proactor = None
208	n/a
209	n/a	super()._unregister_wait_cb(fut)
210	n/a
211	n/a	def _unregister_wait(self):
212	n/a	if not self._registered:
213	n/a	return
214	n/a	self._registered = False
215	n/a
216	n/a	wait_handle = self._wait_handle
217	n/a	self._wait_handle = None
218	n/a	try:
219	n/a	_overlapped.UnregisterWaitEx(wait_handle, self._event)
220	n/a	except OSError as exc:
221	n/a	if exc.winerror != _overlapped.ERROR_IO_PENDING:
222	n/a	context = {
223	n/a	'message': 'Failed to unregister the wait handle',
224	n/a	'exception': exc,
225	n/a	'future': self,
226	n/a	}
227	n/a	if self._source_traceback:
228	n/a	context['source_traceback'] = self._source_traceback
229	n/a	self._loop.call_exception_handler(context)
230	n/a	return
231	n/a	# ERROR_IO_PENDING is not an error, the wait was unregistered
232	n/a
233	n/a	self._event_fut = self._proactor._wait_cancel(self._event,
234	n/a	self._unregister_wait_cb)
235	n/a
236	n/a
237	n/a	class PipeServer(object):
238	n/a	"""Class representing a pipe server.
239	n/a
240	n/a	This is much like a bound, listening socket.
241	n/a	"""
242	n/a	def __init__(self, address):
243	n/a	self._address = address
244	n/a	self._free_instances = weakref.WeakSet()
245	n/a	# initialize the pipe attribute before calling _server_pipe_handle()
246	n/a	# because this function can raise an exception and the destructor calls
247	n/a	# the close() method
248	n/a	self._pipe = None
249	n/a	self._accept_pipe_future = None
250	n/a	self._pipe = self._server_pipe_handle(True)
251	n/a
252	n/a	def _get_unconnected_pipe(self):
253	n/a	# Create new instance and return previous one. This ensures
254	n/a	# that (until the server is closed) there is always at least
255	n/a	# one pipe handle for address. Therefore if a client attempt
256	n/a	# to connect it will not fail with FileNotFoundError.
257	n/a	tmp, self._pipe = self._pipe, self._server_pipe_handle(False)
258	n/a	return tmp
259	n/a
260	n/a	def _server_pipe_handle(self, first):
261	n/a	# Return a wrapper for a new pipe handle.
262	n/a	if self.closed():
263	n/a	return None
264	n/a	flags = _winapi.PIPE_ACCESS_DUPLEX \| _winapi.FILE_FLAG_OVERLAPPED
265	n/a	if first:
266	n/a	flags \|= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
267	n/a	h = _winapi.CreateNamedPipe(
268	n/a	self._address, flags,
269	n/a	_winapi.PIPE_TYPE_MESSAGE \| _winapi.PIPE_READMODE_MESSAGE \|
270	n/a	_winapi.PIPE_WAIT,
271	n/a	_winapi.PIPE_UNLIMITED_INSTANCES,
272	n/a	windows_utils.BUFSIZE, windows_utils.BUFSIZE,
273	n/a	_winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
274	n/a	pipe = windows_utils.PipeHandle(h)
275	n/a	self._free_instances.add(pipe)
276	n/a	return pipe
277	n/a
278	n/a	def closed(self):
279	n/a	return (self._address is None)
280	n/a
281	n/a	def close(self):
282	n/a	if self._accept_pipe_future is not None:
283	n/a	self._accept_pipe_future.cancel()
284	n/a	self._accept_pipe_future = None
285	n/a	# Close all instances which have not been connected to by a client.
286	n/a	if self._address is not None:
287	n/a	for pipe in self._free_instances:
288	n/a	pipe.close()
289	n/a	self._pipe = None
290	n/a	self._address = None
291	n/a	self._free_instances.clear()
292	n/a
293	n/a	__del__ = close
294	n/a
295	n/a
296	n/a	class _WindowsSelectorEventLoop(selector_events.BaseSelectorEventLoop):
297	n/a	"""Windows version of selector event loop."""
298	n/a
299	n/a	def _socketpair(self):
300	n/a	return windows_utils.socketpair()
301	n/a
302	n/a
303	n/a	class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
304	n/a	"""Windows version of proactor event loop using IOCP."""
305	n/a
306	n/a	def __init__(self, proactor=None):
307	n/a	if proactor is None:
308	n/a	proactor = IocpProactor()
309	n/a	super().__init__(proactor)
310	n/a
311	n/a	def _socketpair(self):
312	n/a	return windows_utils.socketpair()
313	n/a
314	n/a	@coroutine
315	n/a	def create_pipe_connection(self, protocol_factory, address):
316	n/a	f = self._proactor.connect_pipe(address)
317	n/a	pipe = yield from f
318	n/a	protocol = protocol_factory()
319	n/a	trans = self._make_duplex_pipe_transport(pipe, protocol,
320	n/a	extra={'addr': address})
321	n/a	return trans, protocol
322	n/a
323	n/a	@coroutine
324	n/a	def start_serving_pipe(self, protocol_factory, address):
325	n/a	server = PipeServer(address)
326	n/a
327	n/a	def loop_accept_pipe(f=None):
328	n/a	pipe = None
329	n/a	try:
330	n/a	if f:
331	n/a	pipe = f.result()
332	n/a	server._free_instances.discard(pipe)
333	n/a
334	n/a	if server.closed():
335	n/a	# A client connected before the server was closed:
336	n/a	# drop the client (close the pipe) and exit
337	n/a	pipe.close()
338	n/a	return
339	n/a
340	n/a	protocol = protocol_factory()
341	n/a	self._make_duplex_pipe_transport(
342	n/a	pipe, protocol, extra={'addr': address})
343	n/a
344	n/a	pipe = server._get_unconnected_pipe()
345	n/a	if pipe is None:
346	n/a	return
347	n/a
348	n/a	f = self._proactor.accept_pipe(pipe)
349	n/a	except OSError as exc:
350	n/a	if pipe and pipe.fileno() != -1:
351	n/a	self.call_exception_handler({
352	n/a	'message': 'Pipe accept failed',
353	n/a	'exception': exc,
354	n/a	'pipe': pipe,
355	n/a	})
356	n/a	pipe.close()
357	n/a	elif self._debug:
358	n/a	logger.warning("Accept pipe failed on pipe %r",
359	n/a	pipe, exc_info=True)
360	n/a	except futures.CancelledError:
361	n/a	if pipe:
362	n/a	pipe.close()
363	n/a	else:
364	n/a	server._accept_pipe_future = f
365	n/a	f.add_done_callback(loop_accept_pipe)
366	n/a
367	n/a	self.call_soon(loop_accept_pipe)
368	n/a	return [server]
369	n/a
370	n/a	@coroutine
371	n/a	def _make_subprocess_transport(self, protocol, args, shell,
372	n/a	stdin, stdout, stderr, bufsize,
373	n/a	extra=None, **kwargs):
374	n/a	waiter = self.create_future()
375	n/a	transp = _WindowsSubprocessTransport(self, protocol, args, shell,
376	n/a	stdin, stdout, stderr, bufsize,
377	n/a	waiter=waiter, extra=extra,
378	n/a	**kwargs)
379	n/a	try:
380	n/a	yield from waiter
381	n/a	except Exception as exc:
382	n/a	# Workaround CPython bug #23353: using yield/yield-from in an
383	n/a	# except block of a generator doesn't clear properly sys.exc_info()
384	n/a	err = exc
385	n/a	else:
386	n/a	err = None
387	n/a
388	n/a	if err is not None:
389	n/a	transp.close()
390	n/a	yield from transp._wait()
391	n/a	raise err
392	n/a
393	n/a	return transp
394	n/a
395	n/a
396	n/a	class IocpProactor:
397	n/a	"""Proactor implementation using IOCP."""
398	n/a
399	n/a	def __init__(self, concurrency=0xffffffff):
400	n/a	self._loop = None
401	n/a	self._results = []
402	n/a	self._iocp = _overlapped.CreateIoCompletionPort(
403	n/a	_overlapped.INVALID_HANDLE_VALUE, NULL, 0, concurrency)
404	n/a	self._cache = {}
405	n/a	self._registered = weakref.WeakSet()
406	n/a	self._unregistered = []
407	n/a	self._stopped_serving = weakref.WeakSet()
408	n/a
409	n/a	def __repr__(self):
410	n/a	return ('<%s overlapped#=%s result#=%s>'
411	n/a	% (self.__class__.__name__, len(self._cache),
412	n/a	len(self._results)))
413	n/a
414	n/a	def set_loop(self, loop):
415	n/a	self._loop = loop
416	n/a
417	n/a	def select(self, timeout=None):
418	n/a	if not self._results:
419	n/a	self._poll(timeout)
420	n/a	tmp = self._results
421	n/a	self._results = []
422	n/a	return tmp
423	n/a
424	n/a	def _result(self, value):
425	n/a	fut = self._loop.create_future()
426	n/a	fut.set_result(value)
427	n/a	return fut
428	n/a
429	n/a	def recv(self, conn, nbytes, flags=0):
430	n/a	self._register_with_iocp(conn)
431	n/a	ov = _overlapped.Overlapped(NULL)
432	n/a	try:
433	n/a	if isinstance(conn, socket.socket):
434	n/a	ov.WSARecv(conn.fileno(), nbytes, flags)
435	n/a	else:
436	n/a	ov.ReadFile(conn.fileno(), nbytes)
437	n/a	except BrokenPipeError:
438	n/a	return self._result(b'')
439	n/a
440	n/a	def finish_recv(trans, key, ov):
441	n/a	try:
442	n/a	return ov.getresult()
443	n/a	except OSError as exc:
444	n/a	if exc.winerror == _overlapped.ERROR_NETNAME_DELETED:
445	n/a	raise ConnectionResetError(*exc.args)
446	n/a	else:
447	n/a	raise
448	n/a
449	n/a	return self._register(ov, conn, finish_recv)
450	n/a
451	n/a	def send(self, conn, buf, flags=0):
452	n/a	self._register_with_iocp(conn)
453	n/a	ov = _overlapped.Overlapped(NULL)
454	n/a	if isinstance(conn, socket.socket):
455	n/a	ov.WSASend(conn.fileno(), buf, flags)
456	n/a	else:
457	n/a	ov.WriteFile(conn.fileno(), buf)
458	n/a
459	n/a	def finish_send(trans, key, ov):
460	n/a	try:
461	n/a	return ov.getresult()
462	n/a	except OSError as exc:
463	n/a	if exc.winerror == _overlapped.ERROR_NETNAME_DELETED:
464	n/a	raise ConnectionResetError(*exc.args)
465	n/a	else:
466	n/a	raise
467	n/a
468	n/a	return self._register(ov, conn, finish_send)
469	n/a
470	n/a	def accept(self, listener):
471	n/a	self._register_with_iocp(listener)
472	n/a	conn = self._get_accept_socket(listener.family)
473	n/a	ov = _overlapped.Overlapped(NULL)
474	n/a	ov.AcceptEx(listener.fileno(), conn.fileno())
475	n/a
476	n/a	def finish_accept(trans, key, ov):
477	n/a	ov.getresult()
478	n/a	# Use SO_UPDATE_ACCEPT_CONTEXT so getsockname() etc work.
479	n/a	buf = struct.pack('@P', listener.fileno())
480	n/a	conn.setsockopt(socket.SOL_SOCKET,
481	n/a	_overlapped.SO_UPDATE_ACCEPT_CONTEXT, buf)
482	n/a	conn.settimeout(listener.gettimeout())
483	n/a	return conn, conn.getpeername()
484	n/a
485	n/a	@coroutine
486	n/a	def accept_coro(future, conn):
487	n/a	# Coroutine closing the accept socket if the future is cancelled
488	n/a	try:
489	n/a	yield from future
490	n/a	except futures.CancelledError:
491	n/a	conn.close()
492	n/a	raise
493	n/a
494	n/a	future = self._register(ov, listener, finish_accept)
495	n/a	coro = accept_coro(future, conn)
496	n/a	tasks.ensure_future(coro, loop=self._loop)
497	n/a	return future
498	n/a
499	n/a	def connect(self, conn, address):
500	n/a	self._register_with_iocp(conn)
501	n/a	# The socket needs to be locally bound before we call ConnectEx().
502	n/a	try:
503	n/a	_overlapped.BindLocal(conn.fileno(), conn.family)
504	n/a	except OSError as e:
505	n/a	if e.winerror != errno.WSAEINVAL:
506	n/a	raise
507	n/a	# Probably already locally bound; check using getsockname().
508	n/a	if conn.getsockname()[1] == 0:
509	n/a	raise
510	n/a	ov = _overlapped.Overlapped(NULL)
511	n/a	ov.ConnectEx(conn.fileno(), address)
512	n/a
513	n/a	def finish_connect(trans, key, ov):
514	n/a	ov.getresult()
515	n/a	# Use SO_UPDATE_CONNECT_CONTEXT so getsockname() etc work.
516	n/a	conn.setsockopt(socket.SOL_SOCKET,
517	n/a	_overlapped.SO_UPDATE_CONNECT_CONTEXT, 0)
518	n/a	return conn
519	n/a
520	n/a	return self._register(ov, conn, finish_connect)
521	n/a
522	n/a	def accept_pipe(self, pipe):
523	n/a	self._register_with_iocp(pipe)
524	n/a	ov = _overlapped.Overlapped(NULL)
525	n/a	connected = ov.ConnectNamedPipe(pipe.fileno())
526	n/a
527	n/a	if connected:
528	n/a	# ConnectNamePipe() failed with ERROR_PIPE_CONNECTED which means
529	n/a	# that the pipe is connected. There is no need to wait for the
530	n/a	# completion of the connection.
531	n/a	return self._result(pipe)
532	n/a
533	n/a	def finish_accept_pipe(trans, key, ov):
534	n/a	ov.getresult()
535	n/a	return pipe
536	n/a
537	n/a	return self._register(ov, pipe, finish_accept_pipe)
538	n/a
539	n/a	@coroutine
540	n/a	def connect_pipe(self, address):
541	n/a	delay = CONNECT_PIPE_INIT_DELAY
542	n/a	while True:
543	n/a	# Unfortunately there is no way to do an overlapped connect to a pipe.
544	n/a	# Call CreateFile() in a loop until it doesn't fail with
545	n/a	# ERROR_PIPE_BUSY
546	n/a	try:
547	n/a	handle = _overlapped.ConnectPipe(address)
548	n/a	break
549	n/a	except OSError as exc:
550	n/a	if exc.winerror != _overlapped.ERROR_PIPE_BUSY:
551	n/a	raise
552	n/a
553	n/a	# ConnectPipe() failed with ERROR_PIPE_BUSY: retry later
554	n/a	delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
555	n/a	yield from tasks.sleep(delay, loop=self._loop)
556	n/a
557	n/a	return windows_utils.PipeHandle(handle)
558	n/a
559	n/a	def wait_for_handle(self, handle, timeout=None):
560	n/a	"""Wait for a handle.
561	n/a
562	n/a	Return a Future object. The result of the future is True if the wait
563	n/a	completed, or False if the wait did not complete (on timeout).
564	n/a	"""
565	n/a	return self._wait_for_handle(handle, timeout, False)
566	n/a
567	n/a	def _wait_cancel(self, event, done_callback):
568	n/a	fut = self._wait_for_handle(event, None, True)
569	n/a	# add_done_callback() cannot be used because the wait may only complete
570	n/a	# in IocpProactor.close(), while the event loop is not running.
571	n/a	fut._done_callback = done_callback
572	n/a	return fut
573	n/a
574	n/a	def _wait_for_handle(self, handle, timeout, _is_cancel):
575	n/a	if timeout is None:
576	n/a	ms = _winapi.INFINITE
577	n/a	else:
578	n/a	# RegisterWaitForSingleObject() has a resolution of 1 millisecond,
579	n/a	# round away from zero to wait at least timeout seconds.
580	n/a	ms = math.ceil(timeout * 1e3)
581	n/a
582	n/a	# We only create ov so we can use ov.address as a key for the cache.
583	n/a	ov = _overlapped.Overlapped(NULL)
584	n/a	wait_handle = _overlapped.RegisterWaitWithQueue(
585	n/a	handle, self._iocp, ov.address, ms)
586	n/a	if _is_cancel:
587	n/a	f = _WaitCancelFuture(ov, handle, wait_handle, loop=self._loop)
588	n/a	else:
589	n/a	f = _WaitHandleFuture(ov, handle, wait_handle, self,
590	n/a	loop=self._loop)
591	n/a	if f._source_traceback:
592	n/a	del f._source_traceback[-1]
593	n/a
594	n/a	def finish_wait_for_handle(trans, key, ov):
595	n/a	# Note that this second wait means that we should only use
596	n/a	# this with handles types where a successful wait has no
597	n/a	# effect. So events or processes are all right, but locks
598	n/a	# or semaphores are not. Also note if the handle is
599	n/a	# signalled and then quickly reset, then we may return
600	n/a	# False even though we have not timed out.
601	n/a	return f._poll()
602	n/a
603	n/a	self._cache[ov.address] = (f, ov, 0, finish_wait_for_handle)
604	n/a	return f
605	n/a
606	n/a	def _register_with_iocp(self, obj):
607	n/a	# To get notifications of finished ops on this objects sent to the
608	n/a	# completion port, were must register the handle.
609	n/a	if obj not in self._registered:
610	n/a	self._registered.add(obj)
611	n/a	_overlapped.CreateIoCompletionPort(obj.fileno(), self._iocp, 0, 0)
612	n/a	# XXX We could also use SetFileCompletionNotificationModes()
613	n/a	# to avoid sending notifications to completion port of ops
614	n/a	# that succeed immediately.
615	n/a
616	n/a	def _register(self, ov, obj, callback):
617	n/a	# Return a future which will be set with the result of the
618	n/a	# operation when it completes. The future's value is actually
619	n/a	# the value returned by callback().
620	n/a	f = _OverlappedFuture(ov, loop=self._loop)
621	n/a	if f._source_traceback:
622	n/a	del f._source_traceback[-1]
623	n/a	if not ov.pending:
624	n/a	# The operation has completed, so no need to postpone the
625	n/a	# work. We cannot take this short cut if we need the
626	n/a	# NumberOfBytes, CompletionKey values returned by
627	n/a	# PostQueuedCompletionStatus().
628	n/a	try:
629	n/a	value = callback(None, None, ov)
630	n/a	except OSError as e:
631	n/a	f.set_exception(e)
632	n/a	else:
633	n/a	f.set_result(value)
634	n/a	# Even if GetOverlappedResult() was called, we have to wait for the
635	n/a	# notification of the completion in GetQueuedCompletionStatus().
636	n/a	# Register the overlapped operation to keep a reference to the
637	n/a	# OVERLAPPED object, otherwise the memory is freed and Windows may
638	n/a	# read uninitialized memory.
639	n/a
640	n/a	# Register the overlapped operation for later. Note that
641	n/a	# we only store obj to prevent it from being garbage
642	n/a	# collected too early.
643	n/a	self._cache[ov.address] = (f, ov, obj, callback)
644	n/a	return f
645	n/a
646	n/a	def _unregister(self, ov):
647	n/a	"""Unregister an overlapped object.
648	n/a
649	n/a	Call this method when its future has been cancelled. The event can
650	n/a	already be signalled (pending in the proactor event queue). It is also
651	n/a	safe if the event is never signalled (because it was cancelled).
652	n/a	"""
653	n/a	self._unregistered.append(ov)
654	n/a
655	n/a	def _get_accept_socket(self, family):
656	n/a	s = socket.socket(family)
657	n/a	s.settimeout(0)
658	n/a	return s
659	n/a
660	n/a	def _poll(self, timeout=None):
661	n/a	if timeout is None:
662	n/a	ms = INFINITE
663	n/a	elif timeout < 0:
664	n/a	raise ValueError("negative timeout")
665	n/a	else:
666	n/a	# GetQueuedCompletionStatus() has a resolution of 1 millisecond,
667	n/a	# round away from zero to wait at least timeout seconds.
668	n/a	ms = math.ceil(timeout * 1e3)
669	n/a	if ms >= INFINITE:
670	n/a	raise ValueError("timeout too big")
671	n/a
672	n/a	while True:
673	n/a	status = _overlapped.GetQueuedCompletionStatus(self._iocp, ms)
674	n/a	if status is None:
675	n/a	break
676	n/a	ms = 0
677	n/a
678	n/a	err, transferred, key, address = status
679	n/a	try:
680	n/a	f, ov, obj, callback = self._cache.pop(address)
681	n/a	except KeyError:
682	n/a	if self._loop.get_debug():
683	n/a	self._loop.call_exception_handler({
684	n/a	'message': ('GetQueuedCompletionStatus() returned an '
685	n/a	'unexpected event'),
686	n/a	'status': ('err=%s transferred=%s key=%#x address=%#x'
687	n/a	% (err, transferred, key, address)),
688	n/a	})
689	n/a
690	n/a	# key is either zero, or it is used to return a pipe
691	n/a	# handle which should be closed to avoid a leak.
692	n/a	if key not in (0, _overlapped.INVALID_HANDLE_VALUE):
693	n/a	_winapi.CloseHandle(key)
694	n/a	continue
695	n/a
696	n/a	if obj in self._stopped_serving:
697	n/a	f.cancel()
698	n/a	# Don't call the callback if _register() already read the result or
699	n/a	# if the overlapped has been cancelled
700	n/a	elif not f.done():
701	n/a	try:
702	n/a	value = callback(transferred, key, ov)
703	n/a	except OSError as e:
704	n/a	f.set_exception(e)
705	n/a	self._results.append(f)
706	n/a	else:
707	n/a	f.set_result(value)
708	n/a	self._results.append(f)
709	n/a
710	n/a	# Remove unregisted futures
711	n/a	for ov in self._unregistered:
712	n/a	self._cache.pop(ov.address, None)
713	n/a	self._unregistered.clear()
714	n/a
715	n/a	def _stop_serving(self, obj):
716	n/a	# obj is a socket or pipe handle. It will be closed in
717	n/a	# BaseProactorEventLoop._stop_serving() which will make any
718	n/a	# pending operations fail quickly.
719	n/a	self._stopped_serving.add(obj)
720	n/a
721	n/a	def close(self):
722	n/a	# Cancel remaining registered operations.
723	n/a	for address, (fut, ov, obj, callback) in list(self._cache.items()):
724	n/a	if fut.cancelled():
725	n/a	# Nothing to do with cancelled futures
726	n/a	pass
727	n/a	elif isinstance(fut, _WaitCancelFuture):
728	n/a	# _WaitCancelFuture must not be cancelled
729	n/a	pass
730	n/a	else:
731	n/a	try:
732	n/a	fut.cancel()
733	n/a	except OSError as exc:
734	n/a	if self._loop is not None:
735	n/a	context = {
736	n/a	'message': 'Cancelling a future failed',
737	n/a	'exception': exc,
738	n/a	'future': fut,
739	n/a	}
740	n/a	if fut._source_traceback:
741	n/a	context['source_traceback'] = fut._source_traceback
742	n/a	self._loop.call_exception_handler(context)
743	n/a
744	n/a	while self._cache:
745	n/a	if not self._poll(1):
746	n/a	logger.debug('taking long time to close proactor')
747	n/a
748	n/a	self._results = []
749	n/a	if self._iocp is not None:
750	n/a	_winapi.CloseHandle(self._iocp)
751	n/a	self._iocp = None
752	n/a
753	n/a	def __del__(self):
754	n/a	self.close()
755	n/a
756	n/a
757	n/a	class _WindowsSubprocessTransport(base_subprocess.BaseSubprocessTransport):
758	n/a
759	n/a	def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
760	n/a	self._proc = windows_utils.Popen(
761	n/a	args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
762	n/a	bufsize=bufsize, **kwargs)
763	n/a
764	n/a	def callback(f):
765	n/a	returncode = self._proc.poll()
766	n/a	self._process_exited(returncode)
767	n/a
768	n/a	f = self._loop._proactor.wait_for_handle(int(self._proc._handle))
769	n/a	f.add_done_callback(callback)
770	n/a
771	n/a
772	n/a	SelectorEventLoop = _WindowsSelectorEventLoop
773	n/a
774	n/a
775	n/a	class _WindowsDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
776	n/a	_loop_factory = SelectorEventLoop
777	n/a
778	n/a
779	n/a	DefaultEventLoopPolicy = _WindowsDefaultEventLoopPolicy