proactor_events.py 19.5 KB
Newer Older
1 2 3 4 5 6
"""Event loop using a proactor and related classes.

A proactor is a "notify-on-completion" multiplexer.  Currently a
proactor is only implemented on Windows with IOCP.
"""

7 8
__all__ = ['BaseProactorEventLoop']

9
import socket
10
import warnings
11 12 13 14

from . import base_events
from . import constants
from . import futures
15
from . import sslproto
16
from . import transports
17
from .log import logger
18 19


20 21
class _ProactorBasePipeTransport(transports._FlowControlMixin,
                                 transports.BaseTransport):
22 23 24 25
    """Base class for pipe and socket transports."""

    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None):
26
        super().__init__(extra, loop)
27 28 29 30
        self._set_extra(sock)
        self._sock = sock
        self._protocol = protocol
        self._server = server
31
        self._buffer = None  # None or bytearray.
32 33
        self._read_fut = None
        self._write_fut = None
34
        self._pending_write = 0
35 36 37 38
        self._conn_lost = 0
        self._closing = False  # Set when close() called.
        self._eof_written = False
        if self._server is not None:
39
            self._server._attach()
40 41
        self._loop.call_soon(self._protocol.connection_made, self)
        if waiter is not None:
42
            # only wake up the waiter when connection_made() has been called
43 44
            self._loop.call_soon(futures._set_result_unless_cancelled,
                                 waiter, None)
45

46
    def __repr__(self):
47
        info = [self.__class__.__name__]
48
        if self._sock is None:
49 50 51
            info.append('closed')
        elif self._closing:
            info.append('closing')
52 53
        if self._sock is not None:
            info.append('fd=%s' % self._sock.fileno())
54
        if self._read_fut is not None:
55
            info.append('read=%s' % self._read_fut)
56
        if self._write_fut is not None:
57
            info.append("write=%r" % self._write_fut)
58 59 60 61 62 63 64
        if self._buffer:
            bufsize = len(self._buffer)
            info.append('write_bufsize=%s' % bufsize)
        if self._eof_written:
            info.append('EOF written')
        return '<%s>' % ' '.join(info)

65 66 67
    def _set_extra(self, sock):
        self._extra['pipe'] = sock

68 69 70 71 72 73
    def set_protocol(self, protocol):
        self._protocol = protocol

    def get_protocol(self):
        return self._protocol

74 75 76
    def is_closing(self):
        return self._closing

77 78 79 80 81 82 83 84 85
    def close(self):
        if self._closing:
            return
        self._closing = True
        self._conn_lost += 1
        if not self._buffer and self._write_fut is None:
            self._loop.call_soon(self._call_connection_lost, None)
        if self._read_fut is not None:
            self._read_fut.cancel()
86
            self._read_fut = None
87

88 89 90 91 92
    def __del__(self):
        if self._sock is not None:
            warnings.warn("unclosed transport %r" % self, ResourceWarning,
                          source=self)
            self.close()
93

94
    def _fatal_error(self, exc, message='Fatal error on pipe transport'):
95
        if isinstance(exc, base_events._FATAL_ERROR_IGNORE):
96 97 98
            if self._loop.get_debug():
                logger.debug("%r: %s", self, message, exc_info=True)
        else:
99
            self._loop.call_exception_handler({
100
                'message': message,
101 102 103 104
                'exception': exc,
                'transport': self,
                'protocol': self._protocol,
            })
105 106 107 108 109 110 111 112 113
        self._force_close(exc)

    def _force_close(self, exc):
        if self._closing:
            return
        self._closing = True
        self._conn_lost += 1
        if self._write_fut:
            self._write_fut.cancel()
114
            self._write_fut = None
115 116
        if self._read_fut:
            self._read_fut.cancel()
117
            self._read_fut = None
118
        self._pending_write = 0
119
        self._buffer = None
120 121 122 123 124 125 126 127 128 129 130 131 132
        self._loop.call_soon(self._call_connection_lost, exc)

    def _call_connection_lost(self, exc):
        try:
            self._protocol.connection_lost(exc)
        finally:
            # XXX If there is a pending overlapped read on the other
            # end then it may fail with ERROR_NETNAME_DELETED if we
            # just close our end.  First calling shutdown() seems to
            # cure it, but maybe using DisconnectEx() would be better.
            if hasattr(self._sock, 'shutdown'):
                self._sock.shutdown(socket.SHUT_RDWR)
            self._sock.close()
133
            self._sock = None
134 135
            server = self._server
            if server is not None:
136
                server._detach()
137 138
                self._server = None

139
    def get_write_buffer_size(self):
140 141 142 143
        size = self._pending_write
        if self._buffer is not None:
            size += len(self._buffer)
        return size
144

145 146 147 148 149 150 151 152 153 154 155

class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
                                 transports.ReadTransport):
    """Transport for read pipes."""

    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None):
        super().__init__(loop, sock, protocol, waiter, extra, server)
        self._paused = False
        self._loop.call_soon(self._loop_reading)

156
    def pause_reading(self):
157 158 159 160
        if self._closing:
            raise RuntimeError('Cannot pause_reading() when closing')
        if self._paused:
            raise RuntimeError('Already paused')
161
        self._paused = True
162 163
        if self._loop.get_debug():
            logger.debug("%r pauses reading", self)
164

165
    def resume_reading(self):
166 167
        if not self._paused:
            raise RuntimeError('Not paused')
168 169 170 171
        self._paused = False
        if self._closing:
            return
        self._loop.call_soon(self._loop_reading, self._read_fut)
172 173
        if self._loop.get_debug():
            logger.debug("%r resumes reading", self)
174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199

    def _loop_reading(self, fut=None):
        if self._paused:
            return
        data = None

        try:
            if fut is not None:
                assert self._read_fut is fut or (self._read_fut is None and
                                                 self._closing)
                self._read_fut = None
                data = fut.result()  # deliver data later in "finally" clause

            if self._closing:
                # since close() has been called we ignore any read data
                data = None
                return

            if data == b'':
                # we got end-of-file so no need to reschedule a new read
                return

            # reschedule a new read
            self._read_fut = self._loop._proactor.recv(self._sock, 4096)
        except ConnectionAbortedError as exc:
            if not self._closing:
200
                self._fatal_error(exc, 'Fatal read error on pipe transport')
201 202 203
            elif self._loop.get_debug():
                logger.debug("Read error on pipe transport while closing",
                             exc_info=True)
204 205 206
        except ConnectionResetError as exc:
            self._force_close(exc)
        except OSError as exc:
207
            self._fatal_error(exc, 'Fatal read error on pipe transport')
208 209 210 211 212 213 214 215 216
        except futures.CancelledError:
            if not self._closing:
                raise
        else:
            self._read_fut.add_done_callback(self._loop_reading)
        finally:
            if data:
                self._protocol.data_received(data)
            elif data is not None:
217 218
                if self._loop.get_debug():
                    logger.debug("%r received EOF", self)
219 220 221 222 223
                keep_open = self._protocol.eof_received()
                if not keep_open:
                    self.close()


224
class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport,
225
                                      transports.WriteTransport):
226 227 228
    """Transport for write pipes."""

    def write(self, data):
229
        if not isinstance(data, (bytes, bytearray, memoryview)):
230 231 232
            msg = ("data argument must be a bytes-like object, not '%s'" %
                   type(data).__name__)
            raise TypeError(msg)
233
        if self._eof_written:
234
            raise RuntimeError('write_eof() already called')
235 236 237 238 239 240

        if not data:
            return

        if self._conn_lost:
            if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
241
                logger.warning('socket.send() raised exception.')
242 243 244
            self._conn_lost += 1
            return

245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264
        # Observable states:
        # 1. IDLE: _write_fut and _buffer both None
        # 2. WRITING: _write_fut set; _buffer None
        # 3. BACKED UP: _write_fut set; _buffer a bytearray
        # We always copy the data, so the caller can't modify it
        # while we're still waiting for the I/O to happen.
        if self._write_fut is None:  # IDLE -> WRITING
            assert self._buffer is None
            # Pass a copy, except if it's already immutable.
            self._loop_writing(data=bytes(data))
        elif not self._buffer:  # WRITING -> BACKED UP
            # Make a mutable copy which we can extend.
            self._buffer = bytearray(data)
            self._maybe_pause_protocol()
        else:  # BACKED UP
            # Append to buffer (also copies).
            self._buffer.extend(data)
            self._maybe_pause_protocol()

    def _loop_writing(self, f=None, data=None):
265 266 267
        try:
            assert f is self._write_fut
            self._write_fut = None
268
            self._pending_write = 0
269 270
            if f:
                f.result()
271 272 273
            if data is None:
                data = self._buffer
                self._buffer = None
274 275 276 277 278
            if not data:
                if self._closing:
                    self._loop.call_soon(self._call_connection_lost, None)
                if self._eof_written:
                    self._sock.shutdown(socket.SHUT_WR)
279 280 281 282 283 284
                # Now that we've reduced the buffer size, tell the
                # protocol to resume writing if it was paused.  Note that
                # we do this last since the callback is called immediately
                # and it may add more data to the buffer (even causing the
                # protocol to be paused again).
                self._maybe_resume_protocol()
285 286
            else:
                self._write_fut = self._loop._proactor.send(self._sock, data)
287 288 289 290 291 292 293
                if not self._write_fut.done():
                    assert self._pending_write == 0
                    self._pending_write = len(data)
                    self._write_fut.add_done_callback(self._loop_writing)
                    self._maybe_pause_protocol()
                else:
                    self._write_fut.add_done_callback(self._loop_writing)
294 295 296
        except ConnectionResetError as exc:
            self._force_close(exc)
        except OSError as exc:
297
            self._fatal_error(exc, 'Fatal write error on pipe transport')
298 299 300 301 302 303 304 305 306 307 308

    def can_write_eof(self):
        return True

    def write_eof(self):
        self.close()

    def abort(self):
        self._force_close(None)


309 310 311 312 313 314 315 316 317 318
class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport):
    def __init__(self, *args, **kw):
        super().__init__(*args, **kw)
        self._read_fut = self._loop._proactor.recv(self._sock, 16)
        self._read_fut.add_done_callback(self._pipe_closed)

    def _pipe_closed(self, fut):
        if fut.cancelled():
            # the transport has been closed
            return
319 320 321 322
        assert fut.result() == b''
        if self._closing:
            assert self._read_fut is None
            return
323 324 325
        assert fut is self._read_fut, (fut, self._read_fut)
        self._read_fut = None
        if self._write_fut is not None:
326
            self._force_close(BrokenPipeError())
327 328 329 330
        else:
            self.close()


331
class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport,
332
                                   _ProactorBaseWritePipeTransport,
333 334 335 336 337 338 339 340 341 342 343
                                   transports.Transport):
    """Transport for duplex pipes."""

    def can_write_eof(self):
        return False

    def write_eof(self):
        raise NotImplementedError


class _ProactorSocketTransport(_ProactorReadPipeTransport,
344
                               _ProactorBaseWritePipeTransport,
345 346 347 348 349 350 351 352
                               transports.Transport):
    """Transport for connected sockets."""

    def _set_extra(self, sock):
        self._extra['socket'] = sock
        try:
            self._extra['sockname'] = sock.getsockname()
        except (socket.error, AttributeError):
353 354 355
            if self._loop.get_debug():
                logger.warning("getsockname() failed on %r",
                             sock, exc_info=True)
356 357 358 359
        if 'peername' not in self._extra:
            try:
                self._extra['peername'] = sock.getpeername()
            except (socket.error, AttributeError):
360 361 362
                if self._loop.get_debug():
                    logger.warning("getpeername() failed on %r",
                                   sock, exc_info=True)
363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378

    def can_write_eof(self):
        return True

    def write_eof(self):
        if self._closing or self._eof_written:
            return
        self._eof_written = True
        if self._write_fut is None:
            self._sock.shutdown(socket.SHUT_WR)


class BaseProactorEventLoop(base_events.BaseEventLoop):

    def __init__(self, proactor):
        super().__init__()
379
        logger.debug('Using proactor: %s', proactor.__class__.__name__)
380 381
        self._proactor = proactor
        self._selector = proactor   # convenient alias
382 383
        self._self_reading_future = None
        self._accept_futures = {}   # socket file descriptor => Future
384 385 386 387 388 389 390 391
        proactor.set_loop(self)
        self._make_self_pipe()

    def _make_socket_transport(self, sock, protocol, waiter=None,
                               extra=None, server=None):
        return _ProactorSocketTransport(self, sock, protocol, waiter,
                                        extra, server)

392 393 394 395 396 397 398 399 400 401 402 403 404 405
    def _make_ssl_transport(self, rawsock, protocol, sslcontext, waiter=None,
                            *, server_side=False, server_hostname=None,
                            extra=None, server=None):
        if not sslproto._is_sslproto_available():
            raise NotImplementedError("Proactor event loop requires Python 3.5"
                                      " or newer (ssl.MemoryBIO) to support "
                                      "SSL")

        ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter,
                                            server_side, server_hostname)
        _ProactorSocketTransport(self, rawsock, ssl_protocol,
                                 extra=extra, server=server)
        return ssl_protocol._app_transport

406 407 408 409 410 411 412 413 414 415
    def _make_duplex_pipe_transport(self, sock, protocol, waiter=None,
                                    extra=None):
        return _ProactorDuplexPipeTransport(self,
                                            sock, protocol, waiter, extra)

    def _make_read_pipe_transport(self, sock, protocol, waiter=None,
                                  extra=None):
        return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra)

    def _make_write_pipe_transport(self, sock, protocol, waiter=None,
416 417 418 419
                                   extra=None):
        # We want connection_lost() to be called when other end closes
        return _ProactorWritePipeTransport(self,
                                           sock, protocol, waiter, extra)
420 421

    def close(self):
422
        if self.is_running():
423
            raise RuntimeError("Cannot close a running event loop")
424 425
        if self.is_closed():
            return
426 427 428 429

        # Call these methods before closing the event loop (before calling
        # BaseEventLoop.close), because they can schedule callbacks with
        # call_soon(), which is forbidden when the event loop is closed.
430 431 432 433 434
        self._stop_accept_futures()
        self._close_self_pipe()
        self._proactor.close()
        self._proactor = None
        self._selector = None
435

436 437 438
        # Close the event loop
        super().close()

439 440 441 442 443 444 445
    def sock_recv(self, sock, n):
        return self._proactor.recv(sock, n)

    def sock_sendall(self, sock, data):
        return self._proactor.send(sock, data)

    def sock_connect(self, sock, address):
446
        return self._proactor.connect(sock, address)
447 448 449 450 451 452 453 454

    def sock_accept(self, sock):
        return self._proactor.accept(sock)

    def _socketpair(self):
        raise NotImplementedError

    def _close_self_pipe(self):
455 456 457
        if self._self_reading_future is not None:
            self._self_reading_future.cancel()
            self._self_reading_future = None
458 459 460 461 462 463 464 465 466 467 468 469
        self._ssock.close()
        self._ssock = None
        self._csock.close()
        self._csock = None
        self._internal_fds -= 1

    def _make_self_pipe(self):
        # A self-socket, really. :-)
        self._ssock, self._csock = self._socketpair()
        self._ssock.setblocking(False)
        self._csock.setblocking(False)
        self._internal_fds += 1
470
        self.call_soon(self._loop_self_reading)
471 472 473 474 475 476

    def _loop_self_reading(self, f=None):
        try:
            if f is not None:
                f.result()  # may raise
            f = self._proactor.recv(self._ssock, 4096)
477 478 479 480 481 482 483 484 485
        except futures.CancelledError:
            # _close_self_pipe() has been called, stop waiting for data
            return
        except Exception as exc:
            self.call_exception_handler({
                'message': 'Error on reading from the event loop self pipe',
                'exception': exc,
                'loop': self,
            })
486
        else:
487
            self._self_reading_future = f
488 489 490
            f.add_done_callback(self._loop_self_reading)

    def _write_to_self(self):
491
        self._csock.send(b'\0')
492

493
    def _start_serving(self, protocol_factory, sock,
494
                       sslcontext=None, server=None, backlog=100):
495 496 497 498 499

        def loop(f=None):
            try:
                if f is not None:
                    conn, addr = f.result()
500 501 502
                    if self._debug:
                        logger.debug("%r got a new connection from %r: %r",
                                     server, addr, conn)
503
                    protocol = protocol_factory()
504 505 506 507 508 509 510 511
                    if sslcontext is not None:
                        self._make_ssl_transport(
                            conn, protocol, sslcontext, server_side=True,
                            extra={'peername': addr}, server=server)
                    else:
                        self._make_socket_transport(
                            conn, protocol,
                            extra={'peername': addr}, server=server)
512 513
                if self.is_closed():
                    return
514
                f = self._proactor.accept(sock)
515
            except OSError as exc:
516
                if sock.fileno() != -1:
517
                    self.call_exception_handler({
518
                        'message': 'Accept failed on a socket',
519 520 521
                        'exception': exc,
                        'socket': sock,
                    })
522
                    sock.close()
523 524 525
                elif self._debug:
                    logger.debug("Accept failed on socket %r",
                                 sock, exc_info=True)
526 527 528
            except futures.CancelledError:
                sock.close()
            else:
529
                self._accept_futures[sock.fileno()] = f
530 531 532 533 534
                f.add_done_callback(loop)

        self.call_soon(loop)

    def _process_events(self, event_list):
535 536
        # Events are processed in the IocpProactor._poll() method
        pass
537

538
    def _stop_accept_futures(self):
539 540
        for future in self._accept_futures.values():
            future.cancel()
541 542 543 544
        self._accept_futures.clear()

    def _stop_serving(self, sock):
        self._stop_accept_futures()
545 546
        self._proactor._stop_serving(sock)
        sock.close()