windows_events.py
python
"""Циклы событий Selector и Proactor для Windows."""
import _winapi
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 selector_events
from . import tasks
from . import windows_utils
from . import _overlapped
from .coroutines import coroutine
from .log import logger
__all__ = ['SelectorEventLoop', 'ProactorEventLoop', 'IocpProactor',
'DefaultEventLoopPolicy',
]
NULL = 0
INFINITE = 0xffffffff
ERROR_CONNECTION_REFUSED = 1225
ERROR_CONNECTION_ABORTED = 1236
# Начальная задержка в секундах для connect_pipe() перед повторной попыткой подключения
CONNECT_PIPE_INIT_DELAY = 0.001
# Максимальная задержка в секундах для connect_pipe() перед повторной попыткой подключения
CONNECT_PIPE_MAX_DELAY = 0.100
class _OverlappedFuture(futures.Future):
"""Подкласс Future, представляющий перекрывающуюся операцию.
Его отмена немедленно отменит перекрывающуюся операцию.
"""
def __init__(self, ov, *, loop=None):
super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
self._ov = ov
def _repr_info(self):
info = super()._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().cancel()
def set_exception(self, exception):
super().set_exception(exception)
self._cancel_overlapped()
def set_result(self, result):
super().set_result(result)
self._ov = None
class _BaseWaitHandleFuture(futures.Future):
"""Подкласс Future, представляющий дескриптор ожидания."""
def __init__(self, ov, handle, wait_handle, *, loop=None):
super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
# Храните ссылку на объект Overlapped, чтобы он оставался живым, пока
# ожидание не будет отменено
self._ov = ov
self._handle = handle
self._wait_handle = wait_handle
# Следует ли вызывать UnregisterWaitEx(), если ожидание завершается
# или отменяется?
self._registered = True
def _poll(self):
# неблокирующее ожидание: используйте тайм-аут 0 миллисекунд
return (_winapi.WaitForSingleObject(self._handle, 0) ==
_winapi.WAIT_OBJECT_0)
def _repr_info(self):
info = super()._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):
# Ожидание было отменено: небезопасно уничтожать Overlapped
# объект
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 означает, что дерегистрация ожидает выполнения
self._unregister_wait_cb(None)
def cancel(self):
self._unregister_wait()
return super().cancel()
def set_exception(self, exception):
self._unregister_wait()
super().set_exception(exception)
def set_result(self, result):
self._unregister_wait()
super().set_result(result)
class _WaitCancelFuture(_BaseWaitHandleFuture):
"""Подкласс Future, представляющий ожидание отмены
_WaitHandleFuture с использованием события.
"""
def __init__(self, ov, event, wait_handle, *, loop=None):
super().__init__(ov, event, wait_handle, loop=loop)
self._done_callback = None
def cancel(self):
raise RuntimeError("_WaitCancelFuture must not be cancelled")
def set_result(self, result):
super().set_result(result)
if self._done_callback is not None:
self._done_callback(self)
def set_exception(self, exception):
super().set_exception(exception)
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().__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
# Если ожидание было отменено, оно может никогда не быть сигнализировано, поэтому
# необходимо отменить его регистрацию. В противном случае IocpProactor.close() будет
# ждать вечно события, которое никогда не наступит.
#
# Если IocpProactor уже получил событие, можно безопасно вызвать
# _unregister(), поскольку мы сохранили ссылку на объект Overlapped
# который используется как уникальный ключ.
self._proactor._unregister(self._ov)
self._proactor = None
super()._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 — это не ошибка, ожидание было отменено
self._event_fut = self._proactor._wait_cancel(self._event,
self._unregister_wait_cb)
class PipeServer(object):
"""Класс, представляющий сервер канала.
Он очень похож на привязанный прослушивающий сокет.
"""
def __init__(self, address):
self._address = address
self._free_instances = weakref.WeakSet()
# инициализируйте атрибут pipe перед вызовом _server_pipe_handle()
# поскольку эта функция может вызвать исключение, а деструктор вызывает
# метод close()
self._pipe = None
self._accept_pipe_future = None
self._pipe = self._server_pipe_handle(True)
def _get_unconnected_pipe(self):
# Создать новый экземпляр и вернуть предыдущий. Это гарантирует,
# что (пока сервер не закрыт) существует как минимум
# один дескриптор канала для адреса. Поэтому, если клиент попытается
# подключиться, он не получит ошибку FileNotFoundError.
tmp, self._pipe = self._pipe, self._server_pipe_handle(False)
return tmp
def _server_pipe_handle(self, first):
# Возвращает обёртку для нового дескриптора канала.
if self.closed():
return None
flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
if first:
flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
h = _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
# Закрыть все экземпляры, к которым не подключился клиент.
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."""
def _socketpair(self):
return windows_utils.socketpair()
class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
"""Версия проакторного цикла событий для Windows с использованием IOCP."""
def __init__(self, proactor=None):
if proactor is None:
proactor = IocpProactor()
super().__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})
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():
# Клиент подключился до закрытия сервера:
# отключить клиента (закрыть канал) и завершиться
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 = self.create_future()
transp = _WindowsSubprocessTransport(self, protocol, args, shell,
stdin, stdout, stderr, bufsize,
waiter=waiter, extra=extra,
**kwargs)
try:
yield from waiter
except Exception as exc:
# Обход ошибки CPython #23353: использование yield/yield-from в
# блок except генератора не очищает должным образом sys.exc_info()
err = exc
else:
err = None
if err is not None:
transp.close()
yield from transp._wait()
raise err
return transp
class IocpProactor:
"""Реализация проактора с использованием 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 = self._loop.create_future()
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):
ov.WSARecv(conn.fileno(), nbytes, flags)
else:
ov.ReadFile(conn.fileno(), nbytes)
except BrokenPipeError:
return self._result(b'')
def finish_recv(trans, key, ov):
try:
return ov.getresult()
except OSError 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 ov.getresult()
except OSError 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):
ov.getresult()
# Используйте SO_UPDATE_ACCEPT_CONTEXT, чтобы getsockname() и т.д. работали.
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):
# Корутина, закрывающая принимающий сокет, если future отменён
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)
# Сокет должен быть привязан локально перед вызовом ConnectEx().
try:
_overlapped.BindLocal(conn.fileno(), conn.family)
except OSError as e:
if e.winerror != errno.WSAEINVAL:
raise
# Вероятно, уже локально привязан; проверьте с помощью getsockname().
if conn.getsockname()[1] == 0:
raise
ov = _overlapped.Overlapped(NULL)
ov.ConnectEx(conn.fileno(), address)
def finish_connect(trans, key, ov):
ov.getresult()
# Используйте SO_UPDATE_CONNECT_CONTEXT, чтобы getsockname() и прочее работали.
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() завершился с ошибкой ERROR_PIPE_CONNECTED, что означает
# что канал подключен. Нет необходимости ждать
# завершения подключения.
return self._result(pipe)
def finish_accept_pipe(trans, key, ov):
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:
# К сожалению, нет способа выполнить перекрывающееся подключение к каналу.
# Вызывайте CreateFile() в цикле, пока не прекратится ошибка
# ERROR_PIPE_BUSY
try:
handle = _overlapped.ConnectPipe(address)
break
except OSError as exc:
if exc.winerror != _overlapped.ERROR_PIPE_BUSY:
raise
# ConnectPipe() завершился с ошибкой ERROR_PIPE_BUSY: повторите позже
delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
yield from tasks.sleep(delay, loop=self._loop)
return windows_utils.PipeHandle(handle)
def wait_for_handle(self, handle, timeout=None):
"""Ожидание дескриптора.
Возвращает объект Future. Результат future равен True, если ожидание
завершено, или False, если ожидание не завершилось (по таймауту).
"""
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() нельзя использовать, так как ожидание может завершиться только
# в IocpProactor.close(), когда цикл событий не запущен.
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() имеет разрешение в 1 миллисекунду,
# округляйте от нуля, чтобы ожидать *не менее* timeout секунд.
ms = math.ceil(timeout * 1e3)
# Мы создаём ov только для того, чтобы использовать ov.address как ключ для кэша.
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):
# Обратите внимание, что это второе ожидание означает, что мы должны использовать только
# это с типами дескрипторов, где успешное ожидание не имеет
# эффекта. Так что события или процессы подходят, но блокировки
# или семафоры — нет. Также обратите внимание, если дескриптор
# сигнализирован, а затем быстро сброшен, то мы можем вернуть
# False, даже если мы не вышли по тайм-ауту.
return f._poll()
self._cache[ov.address] = (f, ov, 0, finish_wait_for_handle)
return f
def _register_with_iocp(self, obj):
# Чтобы получать уведомления о завершённых операциях для этих объектов, отправленные на
# порт завершения, мы должны зарегистрировать дескриптор.
if obj not in self._registered:
self._registered.add(obj)
_overlapped.CreateIoCompletionPort(obj.fileno(), self._iocp, 0, 0)
# XXX Мы также могли бы использовать SetFileCompletionNotificationModes()
# чтобы избежать отправки уведомлений в порт завершения для операций
# которые завершаются успешно немедленно.
def _register(self, ov, obj, callback):
# Возвращает future, который будет установлен результатом
# операции, когда она завершится. Значение future на самом деле
# значение, возвращаемое callback().
f = _OverlappedFuture(ov, loop=self._loop)
if f._source_traceback:
del f._source_traceback[-1]
if not ov.pending:
# Операция завершена, поэтому нет необходимости откладывать
# работу. Мы не можем использовать этот сокращённый путь, если нам нужны
# значения NumberOfBytes, CompletionKey, возвращаемые
# PostQueuedCompletionStatus().
try:
value = callback(None, None, ov)
except OSError as e:
f.set_exception(e)
else:
f.set_result(value)
# Даже если GetOverlappedResult() была вызвана, необходимо дождаться
# уведомления о завершении в GetQueuedCompletionStatus().
# Зарегистрируйте перекрытую операцию, чтобы сохранить ссылку на
# объект OVERLAPPED, иначе память освобождается и Windows может
# прочитать неинициализированную память.
# Зарегистрируйте перекрытую операцию для дальнейшего использования. Обратите внимание, что
# мы храним obj только для предотвращения его
# сборки мусора слишком рано.
self._cache[ov.address] = (f, ov, obj, callback)
return f
def _unregister(self, ov):
"""Отмена регистрации перекрытого объекта.
Вызывайте этот метод, когда его future была отменена. Событие уже может быть сигнализировано (ожидает в очереди событий проактора). Это также безопасно, если событие никогда не будет сигнализировано (поскольку оно было отменено).
"""
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() имеет разрешение 1 миллисекунду,
# округляйте от нуля, чтобы ожидать *не менее* timeout секунд.
ms = 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 либо равен нулю, либо используется для возврата канала
# дескриптор которого следует закрыть, чтобы избежать утечки.
if key not in (0, _overlapped.INVALID_HANDLE_VALUE):
_winapi.CloseHandle(key)
continue
if obj in self._stopped_serving:
f.cancel()
# Не вызывайте колбэк, если _register() уже прочитала результат или
# если перекрытая операция была отменена
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)
# Удалить незарегистрированные фьючерсы
for ov in self._unregistered:
self._cache.pop(ov.address, None)
self._unregistered.clear()
def _stop_serving(self, obj):
# obj — это дескриптор сокета или канала. Он будет закрыт в
# BaseProactorEventLoop._stop_serving(), что приведет к
# быстрому завершению ожидающих операций.
self._stopped_serving.add(obj)
def close(self):
# Отменить оставшиеся зарегистрированные операции.
for address, (fut, ov, obj, callback) in list(self._cache.items()):
if fut.cancelled():
# Не имеет отношения к отменённым future
pass
elif isinstance(fut, _WaitCancelFuture):
# _WaitCancelFuture не должен быть отменён
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