logo

oasis-root

Compiled tree of Oasis Linux based on own branch at <https://hacktivis.me/git/oasis/> git clone https://anongit.hacktivis.me/git/oasis-root.git

proactor_events.py (32256B)


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