--- /dev/null
+"""Selector and proactor event loops for Windows."""
+
+import errno
+import math
+import socket
+import struct
+import weakref
+
+from . import events
+from . import base_subprocess
+from . import futures
+from . import proactor_events
+from . import py33_winapi as _winapi
+from . import selector_events
+from . import tasks
+from . import windows_utils
+from . import _overlapped
+from .coroutines import coroutine, From, Return
+from .log import logger
+from .py33_exceptions import wrap_error, BrokenPipeError, ConnectionResetError
+
+
+__all__ = ['SelectorEventLoop', 'ProactorEventLoop', 'IocpProactor',
+ 'DefaultEventLoopPolicy',
+ ]
+
+
+NULL = 0
+INFINITE = 0xffffffff
+ERROR_CONNECTION_REFUSED = 1225
+ERROR_CONNECTION_ABORTED = 1236
+
+# Initial delay in seconds for connect_pipe() before retrying to connect
+CONNECT_PIPE_INIT_DELAY = 0.001
+
+# Maximum delay in seconds for connect_pipe() before retrying to connect
+CONNECT_PIPE_MAX_DELAY = 0.100
+
+
+class _OverlappedFuture(futures.Future):
+ """Subclass of Future which represents an overlapped operation.
+
+ Cancelling it will immediately cancel the overlapped operation.
+ """
+
+ def __init__(self, ov, loop=None):
+ super(_OverlappedFuture, self).__init__(loop=loop)
+ if self._source_traceback:
+ del self._source_traceback[-1]
+ self._ov = ov
+
+ def _repr_info(self):
+ info = super(_OverlappedFuture, self)._repr_info()
+ if self._ov is not None:
+ state = 'pending' if self._ov.pending else 'completed'
+ info.insert(1, 'overlapped=<%s, %#x>' % (state, self._ov.address))
+ return info
+
+ def _cancel_overlapped(self):
+ if self._ov is None:
+ return
+ try:
+ self._ov.cancel()
+ except OSError as exc:
+ context = {
+ 'message': 'Cancelling an overlapped future failed',
+ 'exception': exc,
+ 'future': self,
+ }
+ if self._source_traceback:
+ context['source_traceback'] = self._source_traceback
+ self._loop.call_exception_handler(context)
+ self._ov = None
+
+ def cancel(self):
+ self._cancel_overlapped()
+ return super(_OverlappedFuture, self).cancel()
+
+ def set_exception(self, exception):
+ super(_OverlappedFuture, self).set_exception(exception)
+ self._cancel_overlapped()
+
+ def set_result(self, result):
+ super(_OverlappedFuture, self).set_result(result)
+ self._ov = None
+
+
+class _BaseWaitHandleFuture(futures.Future):
+ """Subclass of Future which represents a wait handle."""
+
+ def __init__(self, ov, handle, wait_handle, loop=None):
+ super(_BaseWaitHandleFuture, self).__init__(loop=loop)
+ if self._source_traceback:
+ del self._source_traceback[-1]
+ # Keep a reference to the Overlapped object to keep it alive until the
+ # wait is unregistered
+ self._ov = ov
+ self._handle = handle
+ self._wait_handle = wait_handle
+
+ # Should we call UnregisterWaitEx() if the wait completes
+ # or is cancelled?
+ self._registered = True
+
+ def _poll(self):
+ # non-blocking wait: use a timeout of 0 millisecond
+ return (_winapi.WaitForSingleObject(self._handle, 0) ==
+ _winapi.WAIT_OBJECT_0)
+
+ def _repr_info(self):
+ info = super(_BaseWaitHandleFuture, self)._repr_info()
+ info.append('handle=%#x' % self._handle)
+ if self._handle is not None:
+ state = 'signaled' if self._poll() else 'waiting'
+ info.append(state)
+ if self._wait_handle is not None:
+ info.append('wait_handle=%#x' % self._wait_handle)
+ return info
+
+ def _unregister_wait_cb(self, fut):
+ # The wait was unregistered: it's not safe to destroy the Overlapped
+ # object
+ self._ov = None
+
+ def _unregister_wait(self):
+ if not self._registered:
+ return
+ self._registered = False
+
+ wait_handle = self._wait_handle
+ self._wait_handle = None
+ try:
+ _overlapped.UnregisterWait(wait_handle)
+ except OSError as exc:
+ if exc.winerror != _overlapped.ERROR_IO_PENDING:
+ context = {
+ 'message': 'Failed to unregister the wait handle',
+ 'exception': exc,
+ 'future': self,
+ }
+ if self._source_traceback:
+ context['source_traceback'] = self._source_traceback
+ self._loop.call_exception_handler(context)
+ return
+ # ERROR_IO_PENDING means that the unregister is pending
+
+ self._unregister_wait_cb(None)
+
+ def cancel(self):
+ self._unregister_wait()
+ return super(_BaseWaitHandleFuture, self).cancel()
+
+ def set_exception(self, exception):
+ self._unregister_wait()
+ super(_BaseWaitHandleFuture, self).set_exception(exception)
+
+ def set_result(self, result):
+ self._unregister_wait()
+ super(_BaseWaitHandleFuture, self).set_result(result)
+
+
+class _WaitCancelFuture(_BaseWaitHandleFuture):
+ """Subclass of Future which represents a wait for the cancellation of a
+ _WaitHandleFuture using an event.
+ """
+
+ def __init__(self, ov, event, wait_handle, loop=None):
+ super(_WaitCancelFuture, self).__init__(ov, event, wait_handle,
+ loop=loop)
+
+ self._done_callback = None
+
+ def cancel(self):
+ raise RuntimeError("_WaitCancelFuture must not be cancelled")
+
+ def _schedule_callbacks(self):
+ super(_WaitCancelFuture, self)._schedule_callbacks()
+ if self._done_callback is not None:
+ self._done_callback(self)
+
+
+class _WaitHandleFuture(_BaseWaitHandleFuture):
+ def __init__(self, ov, handle, wait_handle, proactor, loop=None):
+ super(_WaitHandleFuture, self).__init__(ov, handle, wait_handle,
+ loop=loop)
+ self._proactor = proactor
+ self._unregister_proactor = True
+ self._event = _overlapped.CreateEvent(None, True, False, None)
+ self._event_fut = None
+
+ def _unregister_wait_cb(self, fut):
+ if self._event is not None:
+ _winapi.CloseHandle(self._event)
+ self._event = None
+ self._event_fut = None
+
+ # If the wait was cancelled, the wait may never be signalled, so
+ # it's required to unregister it. Otherwise, IocpProactor.close() will
+ # wait forever for an event which will never come.
+ #
+ # If the IocpProactor already received the event, it's safe to call
+ # _unregister() because we kept a reference to the Overlapped object
+ # which is used as an unique key.
+ self._proactor._unregister(self._ov)
+ self._proactor = None
+
+ super(_WaitHandleFuture, self)._unregister_wait_cb(fut)
+
+ def _unregister_wait(self):
+ if not self._registered:
+ return
+ self._registered = False
+
+ wait_handle = self._wait_handle
+ self._wait_handle = None
+ try:
+ _overlapped.UnregisterWaitEx(wait_handle, self._event)
+ except OSError as exc:
+ if exc.winerror != _overlapped.ERROR_IO_PENDING:
+ context = {
+ 'message': 'Failed to unregister the wait handle',
+ 'exception': exc,
+ 'future': self,
+ }
+ if self._source_traceback:
+ context['source_traceback'] = self._source_traceback
+ self._loop.call_exception_handler(context)
+ return
+ # ERROR_IO_PENDING is not an error, the wait was unregistered
+
+ self._event_fut = self._proactor._wait_cancel(self._event,
+ self._unregister_wait_cb)
+
+
+class PipeServer(object):
+ """Class representing a pipe server.
+
+ This is much like a bound, listening socket.
+ """
+ def __init__(self, address):
+ self._address = address
+ self._free_instances = weakref.WeakSet()
+ # initialize the pipe attribute before calling _server_pipe_handle()
+ # because this function can raise an exception and the destructor calls
+ # the close() method
+ self._pipe = None
+ self._accept_pipe_future = None
+ self._pipe = self._server_pipe_handle(True)
+
+ def _get_unconnected_pipe(self):
+ # Create new instance and return previous one. This ensures
+ # that (until the server is closed) there is always at least
+ # one pipe handle for address. Therefore if a client attempt
+ # to connect it will not fail with FileNotFoundError.
+ tmp, self._pipe = self._pipe, self._server_pipe_handle(False)
+ return tmp
+
+ def _server_pipe_handle(self, first):
+ # Return a wrapper for a new pipe handle.
+ if self.closed():
+ return None
+ flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
+ if first:
+ flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
+ h = wrap_error(_winapi.CreateNamedPipe,
+ self._address, flags,
+ _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
+ _winapi.PIPE_WAIT,
+ _winapi.PIPE_UNLIMITED_INSTANCES,
+ windows_utils.BUFSIZE, windows_utils.BUFSIZE,
+ _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
+ pipe = windows_utils.PipeHandle(h)
+ self._free_instances.add(pipe)
+ return pipe
+
+ def closed(self):
+ return (self._address is None)
+
+ def close(self):
+ if self._accept_pipe_future is not None:
+ self._accept_pipe_future.cancel()
+ self._accept_pipe_future = None
+ # Close all instances which have not been connected to by a client.
+ if self._address is not None:
+ for pipe in self._free_instances:
+ pipe.close()
+ self._pipe = None
+ self._address = None
+ self._free_instances.clear()
+
+ __del__ = close
+
+
+class _WindowsSelectorEventLoop(selector_events.BaseSelectorEventLoop):
+ """Windows version of selector event loop."""
+
+ def _socketpair(self):
+ return windows_utils.socketpair()
+
+
+class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
+ """Windows version of proactor event loop using IOCP."""
+
+ def __init__(self, proactor=None):
+ if proactor is None:
+ proactor = IocpProactor()
+ super(ProactorEventLoop, self).__init__(proactor)
+
+ def _socketpair(self):
+ return windows_utils.socketpair()
+
+ @coroutine
+ def create_pipe_connection(self, protocol_factory, address):
+ f = self._proactor.connect_pipe(address)
+ pipe = yield From(f)
+ protocol = protocol_factory()
+ trans = self._make_duplex_pipe_transport(pipe, protocol,
+ extra={'addr': address})
+ raise Return(trans, protocol)
+
+ @coroutine
+ def start_serving_pipe(self, protocol_factory, address):
+ server = PipeServer(address)
+
+ def loop_accept_pipe(f=None):
+ pipe = None
+ try:
+ if f:
+ pipe = f.result()
+ server._free_instances.discard(pipe)
+
+ if server.closed():
+ # A client connected before the server was closed:
+ # drop the client (close the pipe) and exit
+ pipe.close()
+ return
+
+ protocol = protocol_factory()
+ self._make_duplex_pipe_transport(
+ pipe, protocol, extra={'addr': address})
+
+ pipe = server._get_unconnected_pipe()
+ if pipe is None:
+ return
+
+ f = self._proactor.accept_pipe(pipe)
+ except OSError as exc:
+ if pipe and pipe.fileno() != -1:
+ self.call_exception_handler({
+ 'message': 'Pipe accept failed',
+ 'exception': exc,
+ 'pipe': pipe,
+ })
+ pipe.close()
+ elif self._debug:
+ logger.warning("Accept pipe failed on pipe %r",
+ pipe, exc_info=True)
+ except futures.CancelledError:
+ if pipe:
+ pipe.close()
+ else:
+ server._accept_pipe_future = f
+ f.add_done_callback(loop_accept_pipe)
+
+ self.call_soon(loop_accept_pipe)
+ return [server]
+
+ @coroutine
+ def _make_subprocess_transport(self, protocol, args, shell,
+ stdin, stdout, stderr, bufsize,
+ extra=None, **kwargs):
+ waiter = futures.Future(loop=self)
+ transp = _WindowsSubprocessTransport(self, protocol, args, shell,
+ stdin, stdout, stderr, bufsize,
+ waiter=waiter, extra=extra,
+ **kwargs)
+ try:
+ yield From(waiter)
+ except Exception as exc:
+ # Workaround CPython bug #23353: using yield/yield-from in an
+ # except block of a generator doesn't clear properly sys.exc_info()
+ err = exc
+ else:
+ err = None
+
+ if err is not None:
+ transp.close()
+ yield From(transp._wait())
+ raise err
+
+ raise Return(transp)
+
+
+class IocpProactor(object):
+ """Proactor implementation using IOCP."""
+
+ def __init__(self, concurrency=0xffffffff):
+ self._loop = None
+ self._results = []
+ self._iocp = _overlapped.CreateIoCompletionPort(
+ _overlapped.INVALID_HANDLE_VALUE, NULL, 0, concurrency)
+ self._cache = {}
+ self._registered = weakref.WeakSet()
+ self._unregistered = []
+ self._stopped_serving = weakref.WeakSet()
+
+ def __repr__(self):
+ return ('<%s overlapped#=%s result#=%s>'
+ % (self.__class__.__name__, len(self._cache),
+ len(self._results)))
+
+ def set_loop(self, loop):
+ self._loop = loop
+
+ def select(self, timeout=None):
+ if not self._results:
+ self._poll(timeout)
+ tmp = self._results
+ self._results = []
+ return tmp
+
+ def _result(self, value):
+ fut = futures.Future(loop=self._loop)
+ fut.set_result(value)
+ return fut
+
+ def recv(self, conn, nbytes, flags=0):
+ self._register_with_iocp(conn)
+ ov = _overlapped.Overlapped(NULL)
+ try:
+ if isinstance(conn, socket.socket):
+ wrap_error(ov.WSARecv, conn.fileno(), nbytes, flags)
+ else:
+ wrap_error(ov.ReadFile, conn.fileno(), nbytes)
+ except BrokenPipeError:
+ return self._result(b'')
+
+ def finish_recv(trans, key, ov):
+ try:
+ return wrap_error(ov.getresult)
+ except WindowsError as exc:
+ if exc.winerror == _overlapped.ERROR_NETNAME_DELETED:
+ raise ConnectionResetError(*exc.args)
+ else:
+ raise
+
+ return self._register(ov, conn, finish_recv)
+
+ def send(self, conn, buf, flags=0):
+ self._register_with_iocp(conn)
+ ov = _overlapped.Overlapped(NULL)
+ if isinstance(conn, socket.socket):
+ ov.WSASend(conn.fileno(), buf, flags)
+ else:
+ ov.WriteFile(conn.fileno(), buf)
+
+ def finish_send(trans, key, ov):
+ try:
+ return wrap_error(ov.getresult)
+ except WindowsError as exc:
+ if exc.winerror == _overlapped.ERROR_NETNAME_DELETED:
+ raise ConnectionResetError(*exc.args)
+ else:
+ raise
+
+ return self._register(ov, conn, finish_send)
+
+ def accept(self, listener):
+ self._register_with_iocp(listener)
+ conn = self._get_accept_socket(listener.family)
+ ov = _overlapped.Overlapped(NULL)
+ ov.AcceptEx(listener.fileno(), conn.fileno())
+
+ def finish_accept(trans, key, ov):
+ wrap_error(ov.getresult)
+ # Use SO_UPDATE_ACCEPT_CONTEXT so getsockname() etc work.
+ buf = struct.pack('@P', listener.fileno())
+ conn.setsockopt(socket.SOL_SOCKET,
+ _overlapped.SO_UPDATE_ACCEPT_CONTEXT, buf)
+ conn.settimeout(listener.gettimeout())
+ return conn, conn.getpeername()
+
+ @coroutine
+ def accept_coro(future, conn):
+ # Coroutine closing the accept socket if the future is cancelled
+ try:
+ yield From(future)
+ except futures.CancelledError:
+ conn.close()
+ raise
+
+ future = self._register(ov, listener, finish_accept)
+ coro = accept_coro(future, conn)
+ tasks.ensure_future(coro, loop=self._loop)
+ return future
+
+ def connect(self, conn, address):
+ self._register_with_iocp(conn)
+ # The socket needs to be locally bound before we call ConnectEx().
+ try:
+ _overlapped.BindLocal(conn.fileno(), conn.family)
+ except WindowsError as e:
+ if e.winerror != errno.WSAEINVAL:
+ raise
+ # Probably already locally bound; check using getsockname().
+ if conn.getsockname()[1] == 0:
+ raise
+ ov = _overlapped.Overlapped(NULL)
+ ov.ConnectEx(conn.fileno(), address)
+
+ def finish_connect(trans, key, ov):
+ wrap_error(ov.getresult)
+ # Use SO_UPDATE_CONNECT_CONTEXT so getsockname() etc work.
+ conn.setsockopt(socket.SOL_SOCKET,
+ _overlapped.SO_UPDATE_CONNECT_CONTEXT, 0)
+ return conn
+
+ return self._register(ov, conn, finish_connect)
+
+ def accept_pipe(self, pipe):
+ self._register_with_iocp(pipe)
+ ov = _overlapped.Overlapped(NULL)
+ connected = ov.ConnectNamedPipe(pipe.fileno())
+
+ if connected:
+ # ConnectNamePipe() failed with ERROR_PIPE_CONNECTED which means
+ # that the pipe is connected. There is no need to wait for the
+ # completion of the connection.
+ return self._result(pipe)
+
+ def finish_accept_pipe(trans, key, ov):
+ wrap_error(ov.getresult)
+ return pipe
+
+ return self._register(ov, pipe, finish_accept_pipe)
+
+ @coroutine
+ def connect_pipe(self, address):
+ delay = CONNECT_PIPE_INIT_DELAY
+ while True:
+ # Unfortunately there is no way to do an overlapped connect to a pipe.
+ # Call CreateFile() in a loop until it doesn't fail with
+ # ERROR_PIPE_BUSY
+ try:
+ handle = wrap_error(_overlapped.ConnectPipe, address)
+ break
+ except WindowsError as exc:
+ if exc.winerror != _overlapped.ERROR_PIPE_BUSY:
+ raise
+
+ # ConnectPipe() failed with ERROR_PIPE_BUSY: retry later
+ delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
+ yield From(tasks.sleep(delay, loop=self._loop))
+
+ raise Return(windows_utils.PipeHandle(handle))
+
+ def wait_for_handle(self, handle, timeout=None):
+ """Wait for a handle.
+
+ Return a Future object. The result of the future is True if the wait
+ completed, or False if the wait did not complete (on timeout).
+ """
+ return self._wait_for_handle(handle, timeout, False)
+
+ def _wait_cancel(self, event, done_callback):
+ fut = self._wait_for_handle(event, None, True)
+ # add_done_callback() cannot be used because the wait may only complete
+ # in IocpProactor.close(), while the event loop is not running.
+ fut._done_callback = done_callback
+ return fut
+
+ def _wait_for_handle(self, handle, timeout, _is_cancel):
+ if timeout is None:
+ ms = _winapi.INFINITE
+ else:
+ # RegisterWaitForSingleObject() has a resolution of 1 millisecond,
+ # round away from zero to wait *at least* timeout seconds.
+ ms = int(math.ceil(timeout * 1e3))
+
+ # We only create ov so we can use ov.address as a key for the cache.
+ ov = _overlapped.Overlapped(NULL)
+ wait_handle = _overlapped.RegisterWaitWithQueue(
+ handle, self._iocp, ov.address, ms)
+ if _is_cancel:
+ f = _WaitCancelFuture(ov, handle, wait_handle, loop=self._loop)
+ else:
+ f = _WaitHandleFuture(ov, handle, wait_handle, self,
+ loop=self._loop)
+ if f._source_traceback:
+ del f._source_traceback[-1]
+
+ def finish_wait_for_handle(trans, key, ov):
+ # Note that this second wait means that we should only use
+ # this with handles types where a successful wait has no
+ # effect. So events or processes are all right, but locks
+ # or semaphores are not. Also note if the handle is
+ # signalled and then quickly reset, then we may return
+ # False even though we have not timed out.
+ return f._poll()
+
+ self._cache[ov.address] = (f, ov, 0, finish_wait_for_handle)
+ return f
+
+ def _register_with_iocp(self, obj):
+ # To get notifications of finished ops on this objects sent to the
+ # completion port, were must register the handle.
+ if obj not in self._registered:
+ self._registered.add(obj)
+ _overlapped.CreateIoCompletionPort(obj.fileno(), self._iocp, 0, 0)
+ # XXX We could also use SetFileCompletionNotificationModes()
+ # to avoid sending notifications to completion port of ops
+ # that succeed immediately.
+
+ def _register(self, ov, obj, callback):
+ # Return a future which will be set with the result of the
+ # operation when it completes. The future's value is actually
+ # the value returned by callback().
+ f = _OverlappedFuture(ov, loop=self._loop)
+ if f._source_traceback:
+ del f._source_traceback[-1]
+ if not ov.pending:
+ # The operation has completed, so no need to postpone the
+ # work. We cannot take this short cut if we need the
+ # NumberOfBytes, CompletionKey values returned by
+ # PostQueuedCompletionStatus().
+ try:
+ value = callback(None, None, ov)
+ except OSError as e:
+ f.set_exception(e)
+ else:
+ f.set_result(value)
+ # Even if GetOverlappedResult() was called, we have to wait for the
+ # notification of the completion in GetQueuedCompletionStatus().
+ # Register the overlapped operation to keep a reference to the
+ # OVERLAPPED object, otherwise the memory is freed and Windows may
+ # read uninitialized memory.
+
+ # Register the overlapped operation for later. Note that
+ # we only store obj to prevent it from being garbage
+ # collected too early.
+ self._cache[ov.address] = (f, ov, obj, callback)
+ return f
+
+ def _unregister(self, ov):
+ """Unregister an overlapped object.
+
+ Call this method when its future has been cancelled. The event can
+ already be signalled (pending in the proactor event queue). It is also
+ safe if the event is never signalled (because it was cancelled).
+ """
+ self._unregistered.append(ov)
+
+ def _get_accept_socket(self, family):
+ s = socket.socket(family)
+ s.settimeout(0)
+ return s
+
+ def _poll(self, timeout=None):
+ if timeout is None:
+ ms = INFINITE
+ elif timeout < 0:
+ raise ValueError("negative timeout")
+ else:
+ # GetQueuedCompletionStatus() has a resolution of 1 millisecond,
+ # round away from zero to wait *at least* timeout seconds.
+ ms = int(math.ceil(timeout * 1e3))
+ if ms >= INFINITE:
+ raise ValueError("timeout too big")
+
+ while True:
+ status = _overlapped.GetQueuedCompletionStatus(self._iocp, ms)
+ if status is None:
+ break
+ ms = 0
+
+ err, transferred, key, address = status
+ try:
+ f, ov, obj, callback = self._cache.pop(address)
+ except KeyError:
+ if self._loop.get_debug():
+ self._loop.call_exception_handler({
+ 'message': ('GetQueuedCompletionStatus() returned an '
+ 'unexpected event'),
+ 'status': ('err=%s transferred=%s key=%#x address=%#x'
+ % (err, transferred, key, address)),
+ })
+
+ # key is either zero, or it is used to return a pipe
+ # handle which should be closed to avoid a leak.
+ if key not in (0, _overlapped.INVALID_HANDLE_VALUE):
+ _winapi.CloseHandle(key)
+ continue
+
+ if obj in self._stopped_serving:
+ f.cancel()
+ # Don't call the callback if _register() already read the result or
+ # if the overlapped has been cancelled
+ elif not f.done():
+ try:
+ value = callback(transferred, key, ov)
+ except OSError as e:
+ f.set_exception(e)
+ self._results.append(f)
+ else:
+ f.set_result(value)
+ self._results.append(f)
+
+ # Remove unregisted futures
+ for ov in self._unregistered:
+ self._cache.pop(ov.address, None)
+ del self._unregistered[:]
+
+ def _stop_serving(self, obj):
+ # obj is a socket or pipe handle. It will be closed in
+ # BaseProactorEventLoop._stop_serving() which will make any
+ # pending operations fail quickly.
+ self._stopped_serving.add(obj)
+
+ def close(self):
+ # Cancel remaining registered operations.
+ for address, (fut, ov, obj, callback) in list(self._cache.items()):
+ if fut.cancelled():
+ # Nothing to do with cancelled futures
+ pass
+ elif isinstance(fut, _WaitCancelFuture):
+ # _WaitCancelFuture must not be cancelled
+ pass
+ else:
+ try:
+ fut.cancel()
+ except OSError as exc:
+ if self._loop is not None:
+ context = {
+ 'message': 'Cancelling a future failed',
+ 'exception': exc,
+ 'future': fut,
+ }
+ if fut._source_traceback:
+ context['source_traceback'] = fut._source_traceback
+ self._loop.call_exception_handler(context)
+
+ while self._cache:
+ if not self._poll(1):
+ logger.debug('taking long time to close proactor')
+
+ self._results = []
+ if self._iocp is not None:
+ _winapi.CloseHandle(self._iocp)
+ self._iocp = None
+
+ def __del__(self):
+ self.close()
+
+
+class _WindowsSubprocessTransport(base_subprocess.BaseSubprocessTransport):
+
+ def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
+ self._proc = windows_utils.Popen(
+ args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
+ bufsize=bufsize, **kwargs)
+
+ def callback(f):
+ returncode = self._proc.poll()
+ self._process_exited(returncode)
+
+ f = self._loop._proactor.wait_for_handle(int(self._proc._handle))
+ f.add_done_callback(callback)
+
+
+SelectorEventLoop = _WindowsSelectorEventLoop
+
+
+class _WindowsDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
+ _loop_factory = SelectorEventLoop
+
+
+DefaultEventLoopPolicy = _WindowsDefaultEventLoopPolicy
projects / dotfiles / .git / blobdiff