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'): 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 max_size = 256 * 1024 457 def __init__(self, loop, sock, protocol, address=None, 458 waiter=None, extra=None): 459 self._address = address 460 self._empty_waiter = None 461 # We don't need to call _protocol.connection_made() since our base 462 # constructor does it for us. 463 super().__init__(loop, sock, protocol, waiter=waiter, extra=extra) 464 465 # The base constructor sets _buffer = None, so we set it here 466 self._buffer = collections.deque() 467 self._loop.call_soon(self._loop_reading) 468 469 def _set_extra(self, sock): 470 _set_socket_extra(self, sock) 471 472 def get_write_buffer_size(self): 473 return sum(len(data) for data, _ in self._buffer) 474 475 def abort(self): 476 self._force_close(None) 477 478 def sendto(self, data, addr=None): 479 if not isinstance(data, (bytes, bytearray, memoryview)): 480 raise TypeError('data argument must be bytes-like object (%r)', 481 type(data)) 482 483 if not data: 484 return 485 486 if self._address is not None and addr not in (None, self._address): 487 raise ValueError( 488 f'Invalid address: must be None or {self._address}') 489 490 if self._conn_lost and self._address: 491 if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES: 492 logger.warning('socket.sendto() raised exception.') 493 self._conn_lost += 1 494 return 495 496 # Ensure that what we buffer is immutable. 497 self._buffer.append((bytes(data), addr)) 498 499 if self._write_fut is None: 500 # No current write operations are active, kick one off 501 self._loop_writing() 502 # else: A write operation is already kicked off 503 504 self._maybe_pause_protocol() 505 506 def _loop_writing(self, fut=None): 507 try: 508 if self._conn_lost: 509 return 510 511 assert fut is self._write_fut 512 self._write_fut = None 513 if fut: 514 # We are in a _loop_writing() done callback, get the result 515 fut.result() 516 517 if not self._buffer or (self._conn_lost and self._address): 518 # The connection has been closed 519 if self._closing: 520 self._loop.call_soon(self._call_connection_lost, None) 521 return 522 523 data, addr = self._buffer.popleft() 524 if self._address is not None: 525 self._write_fut = self._loop._proactor.send(self._sock, 526 data) 527 else: 528 self._write_fut = self._loop._proactor.sendto(self._sock, 529 data, 530 addr=addr) 531 except OSError as exc: 532 self._protocol.error_received(exc) 533 except Exception as exc: 534 self._fatal_error(exc, 'Fatal write error on datagram transport') 535 else: 536 self._write_fut.add_done_callback(self._loop_writing) 537 self._maybe_resume_protocol() 538 539 def _loop_reading(self, fut=None): 540 data = None 541 try: 542 if self._conn_lost: 543 return 544 545 assert self._read_fut is fut or (self._read_fut is None and 546 self._closing) 547 548 self._read_fut = None 549 if fut is not None: 550 res = fut.result() 551 552 if self._closing: 553 # since close() has been called we ignore any read data 554 data = None 555 return 556 557 if self._address is not None: 558 data, addr = res, self._address 559 else: 560 data, addr = res 561 562 if self._conn_lost: 563 return 564 if self._address is not None: 565 self._read_fut = self._loop._proactor.recv(self._sock, 566 self.max_size) 567 else: 568 self._read_fut = self._loop._proactor.recvfrom(self._sock, 569 self.max_size) 570 except OSError as exc: 571 self._protocol.error_received(exc) 572 except exceptions.CancelledError: 573 if not self._closing: 574 raise 575 else: 576 if self._read_fut is not None: 577 self._read_fut.add_done_callback(self._loop_reading) 578 finally: 579 if data: 580 self._protocol.datagram_received(data, addr) 581 582 583class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport, 584 _ProactorBaseWritePipeTransport, 585 transports.Transport): 586 """Transport for duplex pipes.""" 587 588 def can_write_eof(self): 589 return False 590 591 def write_eof(self): 592 raise NotImplementedError 593 594 595class _ProactorSocketTransport(_ProactorReadPipeTransport, 596 _ProactorBaseWritePipeTransport, 597 transports.Transport): 598 """Transport for connected sockets.""" 599 600 _sendfile_compatible = constants._SendfileMode.TRY_NATIVE 601 602 def __init__(self, loop, sock, protocol, waiter=None, 603 extra=None, server=None): 604 super().__init__(loop, sock, protocol, waiter, extra, server) 605 base_events._set_nodelay(sock) 606 607 def _set_extra(self, sock): 608 _set_socket_extra(self, sock) 609 610 def can_write_eof(self): 611 return True 612 613 def write_eof(self): 614 if self._closing or self._eof_written: 615 return 616 self._eof_written = True 617 if self._write_fut is None: 618 self._sock.shutdown(socket.SHUT_WR) 619 620 621class BaseProactorEventLoop(base_events.BaseEventLoop): 622 623 def __init__(self, proactor): 624 super().__init__() 625 logger.debug('Using proactor: %s', proactor.__class__.__name__) 626 self._proactor = proactor 627 self._selector = proactor # convenient alias 628 self._self_reading_future = None 629 self._accept_futures = {} # socket file descriptor => Future 630 proactor.set_loop(self) 631 self._make_self_pipe() 632 if threading.current_thread() is threading.main_thread(): 633 # wakeup fd can only be installed to a file descriptor from the main thread 634 signal.set_wakeup_fd(self._csock.fileno()) 635 636 def _make_socket_transport(self, sock, protocol, waiter=None, 637 extra=None, server=None): 638 return _ProactorSocketTransport(self, sock, protocol, waiter, 639 extra, server) 640 641 def _make_ssl_transport( 642 self, rawsock, protocol, sslcontext, waiter=None, 643 *, server_side=False, server_hostname=None, 644 extra=None, server=None, 645 ssl_handshake_timeout=None): 646 ssl_protocol = sslproto.SSLProtocol( 647 self, protocol, sslcontext, waiter, 648 server_side, server_hostname, 649 ssl_handshake_timeout=ssl_handshake_timeout) 650 _ProactorSocketTransport(self, rawsock, ssl_protocol, 651 extra=extra, server=server) 652 return ssl_protocol._app_transport 653 654 def _make_datagram_transport(self, sock, protocol, 655 address=None, waiter=None, extra=None): 656 return _ProactorDatagramTransport(self, sock, protocol, address, 657 waiter, extra) 658 659 def _make_duplex_pipe_transport(self, sock, protocol, waiter=None, 660 extra=None): 661 return _ProactorDuplexPipeTransport(self, 662 sock, protocol, waiter, extra) 663 664 def _make_read_pipe_transport(self, sock, protocol, waiter=None, 665 extra=None): 666 return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra) 667 668 def _make_write_pipe_transport(self, sock, protocol, waiter=None, 669 extra=None): 670 # We want connection_lost() to be called when other end closes 671 return _ProactorWritePipeTransport(self, 672 sock, protocol, waiter, extra) 673 674 def close(self): 675 if self.is_running(): 676 raise RuntimeError("Cannot close a running event loop") 677 if self.is_closed(): 678 return 679 680 if threading.current_thread() is threading.main_thread(): 681 signal.set_wakeup_fd(-1) 682 # Call these methods before closing the event loop (before calling 683 # BaseEventLoop.close), because they can schedule callbacks with 684 # call_soon(), which is forbidden when the event loop is closed. 685 self._stop_accept_futures() 686 self._close_self_pipe() 687 self._proactor.close() 688 self._proactor = None 689 self._selector = None 690 691 # Close the event loop 692 super().close() 693 694 async def sock_recv(self, sock, n): 695 return await self._proactor.recv(sock, n) 696 697 async def sock_recv_into(self, sock, buf): 698 return await self._proactor.recv_into(sock, buf) 699 700 async def sock_sendall(self, sock, data): 701 return await self._proactor.send(sock, data) 702 703 async def sock_connect(self, sock, address): 704 return await self._proactor.connect(sock, address) 705 706 async def sock_accept(self, sock): 707 return await self._proactor.accept(sock) 708 709 async def _sock_sendfile_native(self, sock, file, offset, count): 710 try: 711 fileno = file.fileno() 712 except (AttributeError, io.UnsupportedOperation) as err: 713 raise exceptions.SendfileNotAvailableError("not a regular file") 714 try: 715 fsize = os.fstat(fileno).st_size 716 except OSError: 717 raise exceptions.SendfileNotAvailableError("not a regular file") 718 blocksize = count if count else fsize 719 if not blocksize: 720 return 0 # empty file 721 722 blocksize = min(blocksize, 0xffff_ffff) 723 end_pos = min(offset + count, fsize) if count else fsize 724 offset = min(offset, fsize) 725 total_sent = 0 726 try: 727 while True: 728 blocksize = min(end_pos - offset, blocksize) 729 if blocksize <= 0: 730 return total_sent 731 await self._proactor.sendfile(sock, file, offset, blocksize) 732 offset += blocksize 733 total_sent += blocksize 734 finally: 735 if total_sent > 0: 736 file.seek(offset) 737 738 async def _sendfile_native(self, transp, file, offset, count): 739 resume_reading = transp.is_reading() 740 transp.pause_reading() 741 await transp._make_empty_waiter() 742 try: 743 return await self.sock_sendfile(transp._sock, file, offset, count, 744 fallback=False) 745 finally: 746 transp._reset_empty_waiter() 747 if resume_reading: 748 transp.resume_reading() 749 750 def _close_self_pipe(self): 751 if self._self_reading_future is not None: 752 self._self_reading_future.cancel() 753 self._self_reading_future = None 754 self._ssock.close() 755 self._ssock = None 756 self._csock.close() 757 self._csock = None 758 self._internal_fds -= 1 759 760 def _make_self_pipe(self): 761 # A self-socket, really. :-) 762 self._ssock, self._csock = socket.socketpair() 763 self._ssock.setblocking(False) 764 self._csock.setblocking(False) 765 self._internal_fds += 1 766 767 def _loop_self_reading(self, f=None): 768 try: 769 if f is not None: 770 f.result() # may raise 771 if self._self_reading_future is not f: 772 # When we scheduled this Future, we assigned it to 773 # _self_reading_future. If it's not there now, something has 774 # tried to cancel the loop while this callback was still in the 775 # queue (see windows_events.ProactorEventLoop.run_forever). In 776 # that case stop here instead of continuing to schedule a new 777 # iteration. 778 return 779 f = self._proactor.recv(self._ssock, 4096) 780 except exceptions.CancelledError: 781 # _close_self_pipe() has been called, stop waiting for data 782 return 783 except (SystemExit, KeyboardInterrupt): 784 raise 785 except BaseException as exc: 786 self.call_exception_handler({ 787 'message': 'Error on reading from the event loop self pipe', 788 'exception': exc, 789 'loop': self, 790 }) 791 else: 792 self._self_reading_future = f 793 f.add_done_callback(self._loop_self_reading) 794 795 def _write_to_self(self): 796 # This may be called from a different thread, possibly after 797 # _close_self_pipe() has been called or even while it is 798 # running. Guard for self._csock being None or closed. When 799 # a socket is closed, send() raises OSError (with errno set to 800 # EBADF, but let's not rely on the exact error code). 801 csock = self._csock 802 if csock is None: 803 return 804 805 try: 806 csock.send(b'\0') 807 except OSError: 808 if self._debug: 809 logger.debug("Fail to write a null byte into the " 810 "self-pipe socket", 811 exc_info=True) 812 813 def _start_serving(self, protocol_factory, sock, 814 sslcontext=None, server=None, backlog=100, 815 ssl_handshake_timeout=None): 816 817 def loop(f=None): 818 try: 819 if f is not None: 820 conn, addr = f.result() 821 if self._debug: 822 logger.debug("%r got a new connection from %r: %r", 823 server, addr, conn) 824 protocol = protocol_factory() 825 if sslcontext is not None: 826 self._make_ssl_transport( 827 conn, protocol, sslcontext, server_side=True, 828 extra={'peername': addr}, server=server, 829 ssl_handshake_timeout=ssl_handshake_timeout) 830 else: 831 self._make_socket_transport( 832 conn, protocol, 833 extra={'peername': addr}, server=server) 834 if self.is_closed(): 835 return 836 f = self._proactor.accept(sock) 837 except OSError as exc: 838 if sock.fileno() != -1: 839 self.call_exception_handler({ 840 'message': 'Accept failed on a socket', 841 'exception': exc, 842 'socket': trsock.TransportSocket(sock), 843 }) 844 sock.close() 845 elif self._debug: 846 logger.debug("Accept failed on socket %r", 847 sock, exc_info=True) 848 except exceptions.CancelledError: 849 sock.close() 850 else: 851 self._accept_futures[sock.fileno()] = f 852 f.add_done_callback(loop) 853 854 self.call_soon(loop) 855 856 def _process_events(self, event_list): 857 # Events are processed in the IocpProactor._poll() method 858 pass 859 860 def _stop_accept_futures(self): 861 for future in self._accept_futures.values(): 862 future.cancel() 863 self._accept_futures.clear() 864 865 def _stop_serving(self, sock): 866 future = self._accept_futures.pop(sock.fileno(), None) 867 if future: 868 future.cancel() 869 self._proactor._stop_serving(sock) 870 sock.close() 871