Class IocpProactor

Lib/asyncio/windows_events.py:417–873

Proactor implementation using IOCP.

class IocpProactor:
    """Proactor implementation using IOCP."""

    def __init__(self, concurrency=INFINITE):
        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 _check_closed(self):
        if self._iocp is None:
            raise RuntimeError('IocpProactor is closed')

    def __repr__(self):
        info = ['overlapped#=%s' % len(self._cache),
                'result#=%s' % len(self._results)]
        if self._iocp is None:
            info.append('closed')
        return '<%s %s>' % (self.__class__.__name__, " ".join(info))

    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 = []
        try:
            return tmp
        finally:
            # Needed to break cycles when an exception occurs.
            tmp = None

    def _result(self, value):
        fut = self._loop.create_future()
        fut.set_result(value)
        return fut

    @staticmethod
    def finish_socket_func(trans, key, ov):
        try:
            return ov.getresult()
        except OSError as exc:
            if exc.winerror in (_overlapped.ERROR_NETNAME_DELETED,
                                _overlapped.ERROR_OPERATION_ABORTED):
                raise ConnectionResetError(*exc.args)
            else:
                raise

    @classmethod
    def _finish_recvfrom(cls, trans, key, ov, *, empty_result):
        try:
            return cls.finish_socket_func(trans, key, ov)