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