1"""Event loop using a proactor and related classes. 2 3A proactor is a "notify-on-completion" multiplexer. Currently a 4proactor is only implemented on Windows with IOCP. 5""" 6 7__all__ = 'BaseProactorEventLoop', 8 9import io 10import os 11import socket 12import warnings 13import signal 14import threading 15import collections 16 17from . import base_events 18from . import constants 19from . import futures 20from . import exceptions 21from . import protocols 22from . import sslproto 23from . import transports 24from . import trsock 25from .log import logger 26 27 28def _set_socket_extra(transport, sock): 29 transport._extra['socket'] = trsock.TransportSocket(sock) 30 31 try: 32 transport._extra['sockname'] = sock.getsockname() 33 except socket.error: 34 if transport._loop.get_debug(): 35 logger.warning( 36 "getsockname() failed on %r", sock, exc_info=True) 37 38 if 'peername' not in transport._extra: 39 try: 40 transport._extra['peername'] = sock.getpeername() 41 except socket.error: 42 # UDP sockets may not have a peer name 43 transport._extra['peername'] = None 44 45 46class _ProactorBasePipeTransport(transports._FlowControlMixin, 47 transports.BaseTransport): 48 """Base class for pipe and socket transports.""" 49 50 def __init__(self, loop, sock, protocol, waiter=None, 51 extra=None, server=None): 52 super().__init__(extra, loop) 53 self._set_extra(sock) 54 self._sock = sock 55 self.set_protocol(protocol) 56 self._server = server 57 self._buffer = None # None or bytearray. 58 self._read_fut = None 59 self._write_fut = None 60 self._pending_write = 0 61 self._conn_lost = 0 62 self._closing = False # Set when close() called. 63 self._eof_written = False 64 if self._server is not None: 65 self._server._attach() 66 self._loop.call_soon(self._protocol.connection_made, self) 67 if waiter is not None: 68 # only wake up the waiter when connection_made() has been called 69 self._loop.call_soon(futures._set_result_unless_cancelled, 70 waiter, None) 71 72 def __repr__(self): 73 info = [self.__class__.__name__] 74 if self._sock is None: 75 info.append('closed') 76 elif self._closing: 77 info.append('closing') 78 if self._sock is not None: 79 info.append(f'fd={self._sock.fileno()}') 80 if self._read_fut is not None: 81 info.append(f'read={self._read_fut!r}') 82 if self._write_fut is not None: 83 info.append(f'write={self._write_fut!r}') 84 if self._buffer: 85 info.append(f'write_bufsize={len(self._buffer)}') 86 if self._eof_written: 87 info.append('EOF written') 88 return '<{}>'.format(' '.join(info)) 89 90 def _set_extra(self, sock): 91 self._extra['pipe'] = sock 92 93 def set_protocol(self, protocol): 94 self._protocol = protocol 95 96 def get_protocol(self): 97 return self._protocol 98 99 def is_closing(self): 100 return self._closing 101 102 def close(self): 103 if self._closing: 104 return 105 self._closing = True 106 self._conn_lost += 1 107 if not self._buffer and self._write_fut is None: 108 self._loop.call_soon(self._call_connection_lost, None) 109 if self._read_fut is not None: 110 self._read_fut.cancel() 111 self._read_fut = None 112 113 def __del__(self, _warn=warnings.warn): 114 if self._sock is not None: 115 _warn(f"unclosed transport {self!r}", ResourceWarning, source=self) 116 self.close() 117 118 def _fatal_error(self, exc, message='Fatal error on pipe transport'): 119 try: 120 if isinstance(exc, OSError): 121 if self._loop.get_debug(): 122 logger.debug("%r: %s", self, message, exc_info=True) 123 else: 124 self._loop.call_exception_handler({ 125 'message': message, 126 'exception': exc, 127 'transport': self, 128 'protocol': self._protocol, 129 }) 130 finally: 131 self._force_close(exc) 132 133 def _force_close(self, exc): 134 if self._empty_waiter is not None and not self._empty_waiter.done(): 135 if exc is None: 136 self._empty_waiter.set_result(None) 137 else: 138 self._empty_waiter.set_exception(exc) 139 if self._closing: 140 return 141 self._closing = True 142 self._conn_lost += 1 143 if self._write_fut: 144 self._write_fut.cancel() 145 self._write_fut = None 146 if self._read_fut: 147 self._read_fut.cancel() 148 self._read_fut = None 149 self._pending_write = 0 150 self._buffer = None 151 self._loop.call_soon(self._call_connection_lost, exc) 152 153 def _call_connection_lost(self, exc): 154 try: 155 self._protocol.connection_lost(exc) 156 finally: 157 # XXX If there is a pending overlapped read on the other 158 # end then it may fail with ERROR_NETNAME_DELETED if we 159 # just close our end. First calling shutdown() seems to 160 # cure it, but maybe using DisconnectEx() would be better. 161 if hasattr(self._sock, 'shutdown') and self._sock.fileno() != -1: 162 self._sock.shutdown(socket.SHUT_RDWR) 163 self._sock.close() 164 self._sock = None 165 server = self._server 166 if server is not None: 167 server._detach() 168 self._server = None 169 170 def get_write_buffer_size(self): 171 size = self._pending_write 172 if self._buffer is not None: 173 size += len(self._buffer) 174 return size 175 176 177class _ProactorReadPipeTransport(_ProactorBasePipeTransport, 178 transports.ReadTransport): 179 """Transport for read pipes.""" 180 181 def __init__(self, loop, sock, protocol, waiter=None, 182 extra=None, server=None, buffer_size=65536): 183 self._pending_data_length = -1 184 self._paused = True 185 super().__init__(loop, sock, protocol, waiter, extra, server) 186 187 self._data = bytearray(buffer_size) 188 self._loop.call_soon(self._loop_reading) 189 self._paused = False 190 191 def is_reading(self): 192 return not self._paused and not self._closing 193 194 def pause_reading(self): 195 if self._closing or self._paused: 196 return 197 self._paused = True 198 199 # bpo-33694: Don't cancel self._read_fut because cancelling an 200 # overlapped WSASend() loss silently data with the current proactor 201 # implementation. 202 # 203 # If CancelIoEx() fails with ERROR_NOT_FOUND, it means that WSASend() 204 # completed (even if HasOverlappedIoCompleted() returns 0), but 205 # Overlapped.cancel() currently silently ignores the ERROR_NOT_FOUND 206 # error. Once the overlapped is ignored, the IOCP loop will ignores the 207 # completion I/O event and so not read the result of the overlapped 208 # WSARecv(). 209 210 if self._loop.get_debug(): 211 logger.debug("%r pauses reading", self) 212 213 def resume_reading(self): 214 if self._closing or not self._paused: 215 return 216 217 self._paused = False 218 if self._read_fut is None: 219 self._loop.call_soon(self._loop_reading, None) 220 221 length = self._pending_data_length 222 self._pending_data_length = -1 223 if length > -1: 224 # Call the protocol methode after calling _loop_reading(), 225 # since the protocol can decide to pause reading again. 226 self._loop.call_soon(self._data_received, self._data[:length], length) 227 228 if self._loop.get_debug(): 229 logger.debug("%r resumes reading", self) 230 231 def _eof_received(self): 232 if self._loop.get_debug(): 233 logger.debug("%r received EOF", self) 234 235 try: 236 keep_open = self._protocol.eof_received() 237 except (SystemExit, KeyboardInterrupt): 238 raise 239 except BaseException as exc: 240 self._fatal_error( 241 exc, 'Fatal error: protocol.eof_received() call failed.') 242 return 243 244 if not keep_open: 245 self.close() 246 247 def _data_received(self, data, length): 248 if self._paused: 249 # Don't call any protocol method while reading is paused. 250 # The protocol will be called on resume_reading(). 251 assert self._pending_data_length == -1 252 self._pending_data_length = length 253 return 254 255 if length == 0: 256 self._eof_received() 257 return 258 259 if isinstance(self._protocol, protocols.BufferedProtocol): 260 try: 261 protocols._feed_data_to_buffered_proto(self._protocol, data) 262 except (SystemExit, KeyboardInterrupt): 263 raise 264 except BaseException as exc: 265 self._fatal_error(exc, 266 'Fatal error: protocol.buffer_updated() ' 267 'call failed.') 268 return 269 else: 270 self._protocol.data_received(data) 271 272 def _loop_reading(self, fut=None): 273 length = -1 274 data = None 275 try: 276 if fut is not None: 277 assert self._read_fut is fut or (self._read_fut is None and 278 self._closing) 279 self._read_fut = None 280 if fut.done(): 281 # deliver data later in "finally" clause 282 length = fut.result() 283 if length == 0: 284 # we got end-of-file so no need to reschedule a new read 285 return 286 287 data = self._data[:length] 288 else: 289 # the future will be replaced by next proactor.recv call 290 fut.cancel() 291 292 if self._closing: 293 # since close() has been called we ignore any read data 294 return 295 296 # bpo-33694: buffer_updated() has currently no fast path because of 297 # a data loss issue caused by overlapped WSASend() cancellation. 298 299 if not self._paused: 300 # reschedule a new read 301 self._read_fut = self._loop._proactor.recv_into(self._sock, self._data) 302 except ConnectionAbortedError as exc: 303 if not self._closing: 304 self._fatal_error(exc, 'Fatal read error on pipe transport') 305 elif self._loop.get_debug(): 306 logger.debug("Read error on pipe transport while closing", 307 exc_info=True) 308 except ConnectionResetError as exc: 309 self._force_close(exc) 310 except OSError as exc: 311 self._fatal_error(exc, 'Fatal read error on pipe transport') 312 except exceptions.CancelledError: 313 if not self._closing: 314 raise 315 else: 316 if not self._paused: 317 self._read_fut.add_done_callback(self._loop_reading) 318 finally: 319 if length > -1: 320 self._data_received(data, length) 321 322 323class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport, 324 transports.WriteTransport): 325 """Transport for write pipes.""" 326 327 _start_tls_compatible = True 328 329 def __init__(self, *args, **kw): 330 super().__init__(*args, **kw) 331 self._empty_waiter = None 332 333 def write(self, data): 334 if not isinstance(data, (bytes, bytearray, memoryview)): 335 raise TypeError( 336 f"data argument must be a bytes-like object, " 337 f"not {type(data).__name__}") 338 if self._eof_written: 339 raise RuntimeError('write_eof() already called') 340 if self._empty_waiter is not None: 341 raise RuntimeError('unable to write; sendfile is in progress') 342 343 if not data: 344 return 345 346 if self._conn_lost: 347 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 348 logger.warning('socket.send() raised exception.') 349 self._conn_lost += 1 350 return 351 352 # Observable states: 353 # 1. IDLE: _write_fut and _buffer both None 354 # 2. WRITING: _write_fut set; _buffer None 355 # 3. BACKED UP: _write_fut set; _buffer a bytearray 356 # We always copy the data, so the caller can't modify it 357 # while we're still waiting for the I/O to happen. 358 if self._write_fut is None: # IDLE -> WRITING 359 assert self._buffer is None 360 # Pass a copy, except if it's already immutable. 361 self._loop_writing(data=bytes(data)) 362 elif not self._buffer: # WRITING -> BACKED UP 363 # Make a mutable copy which we can extend. 364 self._buffer = bytearray(data) 365 self._maybe_pause_protocol() 366 else: # BACKED UP 367 # Append to buffer (also copies). 368 self._buffer.extend(data) 369 self._maybe_pause_protocol() 370 371 def _loop_writing(self, f=None, data=None): 372 try: 373 if f is not None and self._write_fut is None and self._closing: 374 # XXX most likely self._force_close() has been called, and 375 # it has set self._write_fut to None. 376 return 377 assert f is self._write_fut 378 self._write_fut = None 379 self._pending_write = 0 380 if f: 381 f.result() 382 if data is None: 383 data = self._buffer 384 self._buffer = None 385 if not data: 386 if self._closing: 387 self._loop.call_soon(self._call_connection_lost, None) 388 if self._eof_written: 389 self._sock.shutdown(socket.SHUT_WR) 390 # Now that we've reduced the buffer size, tell the 391 # protocol to resume writing if it was paused. Note that 392 # we do this last since the callback is called immediately 393 # and it may add more data to the buffer (even causing the 394 # protocol to be paused again). 395 self._maybe_resume_protocol() 396 else: 397 self._write_fut = self._loop._proactor.send(self._sock, data) 398 if not self._write_fut.done(): 399 assert self._pending_write == 0 400 self._pending_write = len(data) 401 self._write_fut.add_done_callback(self._loop_writing) 402 self._maybe_pause_protocol() 403 else: 404 self._write_fut.add_done_callback(self._loop_writing) 405 if self._empty_waiter is not None and self._write_fut is None: 406 self._empty_waiter.set_result(None) 407 except ConnectionResetError as exc: 408 self._force_close(exc) 409 except OSError as exc: 410 self._fatal_error(exc, 'Fatal write error on pipe transport') 411 412 def can_write_eof(self): 413 return True 414 415 def write_eof(self): 416 self.close() 417 418 def abort(self): 419 self._force_close(None) 420 421 def _make_empty_waiter(self): 422 if self._empty_waiter is not None: 423 raise RuntimeError("Empty waiter is already set") 424 self._empty_waiter = self._loop.create_future() 425 if self._write_fut is None: 426 self._empty_waiter.set_result(None) 427 return self._empty_waiter 428 429 def _reset_empty_waiter(self): 430 self._empty_waiter = None 431 432 433class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport): 434 def __init__(self, *args, **kw): 435 super().__init__(*args, **kw) 436 self._read_fut = self._loop._proactor.recv(self._sock, 16) 437 self._read_fut.add_done_callback(self._pipe_closed) 438 439 def _pipe_closed(self, fut): 440 if fut.cancelled(): 441 # the transport has been closed 442 return 443 assert fut.result() == b'' 444 if self._closing: 445 assert self._read_fut is None 446 return 447 assert fut is self._read_fut, (fut, self._read_fut) 448 self._read_fut = None 449 if self._write_fut is not None: 450 self._force_close(BrokenPipeError()) 451 else: 452 self.close() 453 454 455class _ProactorDatagramTransport(_ProactorBasePipeTransport, 456 transports.DatagramTransport): 457 max_size = 256 * 1024 458 def __init__(self, loop, sock, protocol, address=None, 459 waiter=None, extra=None): 460 self._address = address 461 self._empty_waiter = None 462 # We don't need to call _protocol.connection_made() since our base 463 # constructor does it for us. 464 super().__init__(loop, sock, protocol, waiter=waiter, extra=extra) 465 466 # The base constructor sets _buffer = None, so we set it here 467 self._buffer = collections.deque() 468 self._loop.call_soon(self._loop_reading) 469 470 def _set_extra(self, sock): 471 _set_socket_extra(self, sock) 472 473 def get_write_buffer_size(self): 474 return sum(len(data) for data, _ in self._buffer) 475 476 def abort(self): 477 self._force_close(None) 478 479 def sendto(self, data, addr=None): 480 if not isinstance(data, (bytes, bytearray, memoryview)): 481 raise TypeError('data argument must be bytes-like object (%r)', 482 type(data)) 483 484 if not data: 485 return 486 487 if self._address is not None and addr not in (None, self._address): 488 raise ValueError( 489 f'Invalid address: must be None or {self._address}') 490 491 if self._conn_lost and self._address: 492 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 493 logger.warning('socket.sendto() raised exception.') 494 self._conn_lost += 1 495 return 496 497 # Ensure that what we buffer is immutable. 498 self._buffer.append((bytes(data), addr)) 499 500 if self._write_fut is None: 501 # No current write operations are active, kick one off 502 self._loop_writing() 503 # else: A write operation is already kicked off 504 505 self._maybe_pause_protocol() 506 507 def _loop_writing(self, fut=None): 508 try: 509 if self._conn_lost: 510 return 511 512 assert fut is self._write_fut 513 self._write_fut = None 514 if fut: 515 # We are in a _loop_writing() done callback, get the result 516 fut.result() 517 518 if not self._buffer or (self._conn_lost and self._address): 519 # The connection has been closed 520 if self._closing: 521 self._loop.call_soon(self._call_connection_lost, None) 522 return 523 524 data, addr = self._buffer.popleft() 525 if self._address is not None: 526 self._write_fut = self._loop._proactor.send(self._sock, 527 data) 528 else: 529 self._write_fut = self._loop._proactor.sendto(self._sock, 530 data, 531 addr=addr) 532 except OSError as exc: 533 self._protocol.error_received(exc) 534 except Exception as exc: 535 self._fatal_error(exc, 'Fatal write error on datagram transport') 536 else: 537 self._write_fut.add_done_callback(self._loop_writing) 538 self._maybe_resume_protocol() 539 540 def _loop_reading(self, fut=None): 541 data = None 542 try: 543 if self._conn_lost: 544 return 545 546 assert self._read_fut is fut or (self._read_fut is None and 547 self._closing) 548 549 self._read_fut = None 550 if fut is not None: 551 res = fut.result() 552 553 if self._closing: 554 # since close() has been called we ignore any read data 555 data = None 556 return 557 558 if self._address is not None: 559 data, addr = res, self._address 560 else: 561 data, addr = res 562 563 if self._conn_lost: 564 return 565 if self._address is not None: 566 self._read_fut = self._loop._proactor.recv(self._sock, 567 self.max_size) 568 else: 569 self._read_fut = self._loop._proactor.recvfrom(self._sock, 570 self.max_size) 571 except OSError as exc: 572 self._protocol.error_received(exc) 573 except exceptions.CancelledError: 574 if not self._closing: 575 raise 576 else: 577 if self._read_fut is not None: 578 self._read_fut.add_done_callback(self._loop_reading) 579 finally: 580 if data: 581 self._protocol.datagram_received(data, addr) 582 583 584class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 585 _ProactorBaseWritePipeTransport, 586 transports.Transport): 587 """Transport for duplex pipes.""" 588 589 def can_write_eof(self): 590 return False 591 592 def write_eof(self): 593 raise NotImplementedError 594 595 596class _ProactorSocketTransport(_ProactorReadPipeTransport, 597 _ProactorBaseWritePipeTransport, 598 transports.Transport): 599 """Transport for connected sockets.""" 600 601 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 602 603 def __init__(self, loop, sock, protocol, waiter=None, 604 extra=None, server=None): 605 super().__init__(loop, sock, protocol, waiter, extra, server) 606 base_events._set_nodelay(sock) 607 608 def _set_extra(self, sock): 609 _set_socket_extra(self, sock) 610 611 def can_write_eof(self): 612 return True 613 614 def write_eof(self): 615 if self._closing or self._eof_written: 616 return 617 self._eof_written = True 618 if self._write_fut is None: 619 self._sock.shutdown(socket.SHUT_WR) 620 621 622class BaseProactorEventLoop(base_events.BaseEventLoop): 623 624 def __init__(self, proactor): 625 super().__init__() 626 logger.debug('Using proactor: %s', proactor.__class__.__name__) 627 self._proactor = proactor 628 self._selector = proactor # convenient alias 629 self._self_reading_future = None 630 self._accept_futures = {} # socket file descriptor => Future 631 proactor.set_loop(self) 632 self._make_self_pipe() 633 if threading.current_thread() is threading.main_thread(): 634 # wakeup fd can only be installed to a file descriptor from the main thread 635 signal.set_wakeup_fd(self._csock.fileno()) 636 637 def _make_socket_transport(self, sock, protocol, waiter=None, 638 extra=None, server=None): 639 return _ProactorSocketTransport(self, sock, protocol, waiter, 640 extra, server) 641 642 def _make_ssl_transport( 643 self, rawsock, protocol, sslcontext, waiter=None, 644 *, server_side=False, server_hostname=None, 645 extra=None, server=None, 646 ssl_handshake_timeout=None): 647 ssl_protocol = sslproto.SSLProtocol( 648 self, protocol, sslcontext, waiter, 649 server_side, server_hostname, 650 ssl_handshake_timeout=ssl_handshake_timeout) 651 _ProactorSocketTransport(self, rawsock, ssl_protocol, 652 extra=extra, server=server) 653 return ssl_protocol._app_transport 654 655 def _make_datagram_transport(self, sock, protocol, 656 address=None, waiter=None, extra=None): 657 return _ProactorDatagramTransport(self, sock, protocol, address, 658 waiter, extra) 659 660 def _make_duplex_pipe_transport(self, sock, protocol, waiter=None, 661 extra=None): 662 return _ProactorDuplexPipeTransport(self, 663 sock, protocol, waiter, extra) 664 665 def _make_read_pipe_transport(self, sock, protocol, waiter=None, 666 extra=None): 667 return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra) 668 669 def _make_write_pipe_transport(self, sock, protocol, waiter=None, 670 extra=None): 671 # We want connection_lost() to be called when other end closes 672 return _ProactorWritePipeTransport(self, 673 sock, protocol, waiter, extra) 674 675 def close(self): 676 if self.is_running(): 677 raise RuntimeError("Cannot close a running event loop") 678 if self.is_closed(): 679 return 680 681 if threading.current_thread() is threading.main_thread(): 682 signal.set_wakeup_fd(-1) 683 # Call these methods before closing the event loop (before calling 684 # BaseEventLoop.close), because they can schedule callbacks with 685 # call_soon(), which is forbidden when the event loop is closed. 686 self._stop_accept_futures() 687 self._close_self_pipe() 688 self._proactor.close() 689 self._proactor = None 690 self._selector = None 691 692 # Close the event loop 693 super().close() 694 695 async def sock_recv(self, sock, n): 696 return await self._proactor.recv(sock, n) 697 698 async def sock_recv_into(self, sock, buf): 699 return await self._proactor.recv_into(sock, buf) 700 701 async def sock_sendall(self, sock, data): 702 return await self._proactor.send(sock, data) 703 704 async def sock_connect(self, sock, address): 705 return await self._proactor.connect(sock, address) 706 707 async def sock_accept(self, sock): 708 return await self._proactor.accept(sock) 709 710 async def _sock_sendfile_native(self, sock, file, offset, count): 711 try: 712 fileno = file.fileno() 713 except (AttributeError, io.UnsupportedOperation) as err: 714 raise exceptions.SendfileNotAvailableError("not a regular file") 715 try: 716 fsize = os.fstat(fileno).st_size 717 except OSError: 718 raise exceptions.SendfileNotAvailableError("not a regular file") 719 blocksize = count if count else fsize 720 if not blocksize: 721 return 0 # empty file 722 723 blocksize = min(blocksize, 0xffff_ffff) 724 end_pos = min(offset + count, fsize) if count else fsize 725 offset = min(offset, fsize) 726 total_sent = 0 727 try: 728 while True: 729 blocksize = min(end_pos - offset, blocksize) 730 if blocksize <= 0: 731 return total_sent 732 await self._proactor.sendfile(sock, file, offset, blocksize) 733 offset += blocksize 734 total_sent += blocksize 735 finally: 736 if total_sent > 0: 737 file.seek(offset) 738 739 async def _sendfile_native(self, transp, file, offset, count): 740 resume_reading = transp.is_reading() 741 transp.pause_reading() 742 await transp._make_empty_waiter() 743 try: 744 return await self.sock_sendfile(transp._sock, file, offset, count, 745 fallback=False) 746 finally: 747 transp._reset_empty_waiter() 748 if resume_reading: 749 transp.resume_reading() 750 751 def _close_self_pipe(self): 752 if self._self_reading_future is not None: 753 self._self_reading_future.cancel() 754 self._self_reading_future = None 755 self._ssock.close() 756 self._ssock = None 757 self._csock.close() 758 self._csock = None 759 self._internal_fds -= 1 760 761 def _make_self_pipe(self): 762 # A self-socket, really. :-) 763 self._ssock, self._csock = socket.socketpair() 764 self._ssock.setblocking(False) 765 self._csock.setblocking(False) 766 self._internal_fds += 1 767 768 def _loop_self_reading(self, f=None): 769 try: 770 if f is not None: 771 f.result() # may raise 772 if self._self_reading_future is not f: 773 # When we scheduled this Future, we assigned it to 774 # _self_reading_future. If it's not there now, something has 775 # tried to cancel the loop while this callback was still in the 776 # queue (see windows_events.ProactorEventLoop.run_forever). In 777 # that case stop here instead of continuing to schedule a new 778 # iteration. 779 return 780 f = self._proactor.recv(self._ssock, 4096) 781 except exceptions.CancelledError: 782 # _close_self_pipe() has been called, stop waiting for data 783 return 784 except (SystemExit, KeyboardInterrupt): 785 raise 786 except BaseException as exc: 787 self.call_exception_handler({ 788 'message': 'Error on reading from the event loop self pipe', 789 'exception': exc, 790 'loop': self, 791 }) 792 else: 793 self._self_reading_future = f 794 f.add_done_callback(self._loop_self_reading) 795 796 def _write_to_self(self): 797 # This may be called from a different thread, possibly after 798 # _close_self_pipe() has been called or even while it is 799 # running. Guard for self._csock being None or closed. When 800 # a socket is closed, send() raises OSError (with errno set to 801 # EBADF, but let's not rely on the exact error code). 802 csock = self._csock 803 if csock is None: 804 return 805 806 try: 807 csock.send(b'\0') 808 except OSError: 809 if self._debug: 810 logger.debug("Fail to write a null byte into the " 811 "self-pipe socket", 812 exc_info=True) 813 814 def _start_serving(self, protocol_factory, sock, 815 sslcontext=None, server=None, backlog=100, 816 ssl_handshake_timeout=None): 817 818 def loop(f=None): 819 try: 820 if f is not None: 821 conn, addr = f.result() 822 if self._debug: 823 logger.debug("%r got a new connection from %r: %r", 824 server, addr, conn) 825 protocol = protocol_factory() 826 if sslcontext is not None: 827 self._make_ssl_transport( 828 conn, protocol, sslcontext, server_side=True, 829 extra={'peername': addr}, server=server, 830 ssl_handshake_timeout=ssl_handshake_timeout) 831 else: 832 self._make_socket_transport( 833 conn, protocol, 834 extra={'peername': addr}, server=server) 835 if self.is_closed(): 836 return 837 f = self._proactor.accept(sock) 838 except OSError as exc: 839 if sock.fileno() != -1: 840 self.call_exception_handler({ 841 'message': 'Accept failed on a socket', 842 'exception': exc, 843 'socket': trsock.TransportSocket(sock), 844 }) 845 sock.close() 846 elif self._debug: 847 logger.debug("Accept failed on socket %r", 848 sock, exc_info=True) 849 except exceptions.CancelledError: 850 sock.close() 851 else: 852 self._accept_futures[sock.fileno()] = f 853 f.add_done_callback(loop) 854 855 self.call_soon(loop) 856 857 def _process_events(self, event_list): 858 # Events are processed in the IocpProactor._poll() method 859 pass 860 861 def _stop_accept_futures(self): 862 for future in self._accept_futures.values(): 863 future.cancel() 864 self._accept_futures.clear() 865 866 def _stop_serving(self, sock): 867 future = self._accept_futures.pop(sock.fileno(), None) 868 if future: 869 future.cancel() 870 self._proactor._stop_serving(sock) 871 sock.close() 872