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

unix_events.py (51599B)


  1. """Selector event loop for Unix with signal handling."""
  2. import errno
  3. import io
  4. import itertools
  5. import os
  6. import selectors
  7. import signal
  8. import socket
  9. import stat
  10. import subprocess
  11. import sys
  12. import threading
  13. import warnings
  14. from . import base_events
  15. from . import base_subprocess
  16. from . import constants
  17. from . import coroutines
  18. from . import events
  19. from . import exceptions
  20. from . import futures
  21. from . import selector_events
  22. from . import tasks
  23. from . import transports
  24. from .log import logger
  25. __all__ = (
  26. 'SelectorEventLoop',
  27. 'AbstractChildWatcher', 'SafeChildWatcher',
  28. 'FastChildWatcher', 'PidfdChildWatcher',
  29. 'MultiLoopChildWatcher', 'ThreadedChildWatcher',
  30. 'DefaultEventLoopPolicy',
  31. )
  32. if sys.platform == 'win32': # pragma: no cover
  33. raise ImportError('Signals are not really supported on Windows')
  34. def _sighandler_noop(signum, frame):
  35. """Dummy signal handler."""
  36. pass
  37. def waitstatus_to_exitcode(status):
  38. try:
  39. return os.waitstatus_to_exitcode(status)
  40. except ValueError:
  41. # The child exited, but we don't understand its status.
  42. # This shouldn't happen, but if it does, let's just
  43. # return that status; perhaps that helps debug it.
  44. return status
  45. class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
  46. """Unix event loop.
  47. Adds signal handling and UNIX Domain Socket support to SelectorEventLoop.
  48. """
  49. def __init__(self, selector=None):
  50. super().__init__(selector)
  51. self._signal_handlers = {}
  52. def close(self):
  53. super().close()
  54. if not sys.is_finalizing():
  55. for sig in list(self._signal_handlers):
  56. self.remove_signal_handler(sig)
  57. else:
  58. if self._signal_handlers:
  59. warnings.warn(f"Closing the loop {self!r} "
  60. f"on interpreter shutdown "
  61. f"stage, skipping signal handlers removal",
  62. ResourceWarning,
  63. source=self)
  64. self._signal_handlers.clear()
  65. def _process_self_data(self, data):
  66. for signum in data:
  67. if not signum:
  68. # ignore null bytes written by _write_to_self()
  69. continue
  70. self._handle_signal(signum)
  71. def add_signal_handler(self, sig, callback, *args):
  72. """Add a handler for a signal. UNIX only.
  73. Raise ValueError if the signal number is invalid or uncatchable.
  74. Raise RuntimeError if there is a problem setting up the handler.
  75. """
  76. if (coroutines.iscoroutine(callback) or
  77. coroutines.iscoroutinefunction(callback)):
  78. raise TypeError("coroutines cannot be used "
  79. "with add_signal_handler()")
  80. self._check_signal(sig)
  81. self._check_closed()
  82. try:
  83. # set_wakeup_fd() raises ValueError if this is not the
  84. # main thread. By calling it early we ensure that an
  85. # event loop running in another thread cannot add a signal
  86. # handler.
  87. signal.set_wakeup_fd(self._csock.fileno())
  88. except (ValueError, OSError) as exc:
  89. raise RuntimeError(str(exc))
  90. handle = events.Handle(callback, args, self, None)
  91. self._signal_handlers[sig] = handle
  92. try:
  93. # Register a dummy signal handler to ask Python to write the signal
  94. # number in the wakeup file descriptor. _process_self_data() will
  95. # read signal numbers from this file descriptor to handle signals.
  96. signal.signal(sig, _sighandler_noop)
  97. # Set SA_RESTART to limit EINTR occurrences.
  98. signal.siginterrupt(sig, False)
  99. except OSError as exc:
  100. del self._signal_handlers[sig]
  101. if not self._signal_handlers:
  102. try:
  103. signal.set_wakeup_fd(-1)
  104. except (ValueError, OSError) as nexc:
  105. logger.info('set_wakeup_fd(-1) failed: %s', nexc)
  106. if exc.errno == errno.EINVAL:
  107. raise RuntimeError(f'sig {sig} cannot be caught')
  108. else:
  109. raise
  110. def _handle_signal(self, sig):
  111. """Internal helper that is the actual signal handler."""
  112. handle = self._signal_handlers.get(sig)
  113. if handle is None:
  114. return # Assume it's some race condition.
  115. if handle._cancelled:
  116. self.remove_signal_handler(sig) # Remove it properly.
  117. else:
  118. self._add_callback_signalsafe(handle)
  119. def remove_signal_handler(self, sig):
  120. """Remove a handler for a signal. UNIX only.
  121. Return True if a signal handler was removed, False if not.
  122. """
  123. self._check_signal(sig)
  124. try:
  125. del self._signal_handlers[sig]
  126. except KeyError:
  127. return False
  128. if sig == signal.SIGINT:
  129. handler = signal.default_int_handler
  130. else:
  131. handler = signal.SIG_DFL
  132. try:
  133. signal.signal(sig, handler)
  134. except OSError as exc:
  135. if exc.errno == errno.EINVAL:
  136. raise RuntimeError(f'sig {sig} cannot be caught')
  137. else:
  138. raise
  139. if not self._signal_handlers:
  140. try:
  141. signal.set_wakeup_fd(-1)
  142. except (ValueError, OSError) as exc:
  143. logger.info('set_wakeup_fd(-1) failed: %s', exc)
  144. return True
  145. def _check_signal(self, sig):
  146. """Internal helper to validate a signal.
  147. Raise ValueError if the signal number is invalid or uncatchable.
  148. Raise RuntimeError if there is a problem setting up the handler.
  149. """
  150. if not isinstance(sig, int):
  151. raise TypeError(f'sig must be an int, not {sig!r}')
  152. if sig not in signal.valid_signals():
  153. raise ValueError(f'invalid signal number {sig}')
  154. def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
  155. extra=None):
  156. return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
  157. def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
  158. extra=None):
  159. return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
  160. async def _make_subprocess_transport(self, protocol, args, shell,
  161. stdin, stdout, stderr, bufsize,
  162. extra=None, **kwargs):
  163. with events.get_child_watcher() as watcher:
  164. if not watcher.is_active():
  165. # Check early.
  166. # Raising exception before process creation
  167. # prevents subprocess execution if the watcher
  168. # is not ready to handle it.
  169. raise RuntimeError("asyncio.get_child_watcher() is not activated, "
  170. "subprocess support is not installed.")
  171. waiter = self.create_future()
  172. transp = _UnixSubprocessTransport(self, protocol, args, shell,
  173. stdin, stdout, stderr, bufsize,
  174. waiter=waiter, extra=extra,
  175. **kwargs)
  176. watcher.add_child_handler(transp.get_pid(),
  177. self._child_watcher_callback, transp)
  178. try:
  179. await waiter
  180. except (SystemExit, KeyboardInterrupt):
  181. raise
  182. except BaseException:
  183. transp.close()
  184. await transp._wait()
  185. raise
  186. return transp
  187. def _child_watcher_callback(self, pid, returncode, transp):
  188. self.call_soon_threadsafe(transp._process_exited, returncode)
  189. async def create_unix_connection(
  190. self, protocol_factory, path=None, *,
  191. ssl=None, sock=None,
  192. server_hostname=None,
  193. ssl_handshake_timeout=None):
  194. assert server_hostname is None or isinstance(server_hostname, str)
  195. if ssl:
  196. if server_hostname is None:
  197. raise ValueError(
  198. 'you have to pass server_hostname when using ssl')
  199. else:
  200. if server_hostname is not None:
  201. raise ValueError('server_hostname is only meaningful with ssl')
  202. if ssl_handshake_timeout is not None:
  203. raise ValueError(
  204. 'ssl_handshake_timeout is only meaningful with ssl')
  205. if path is not None:
  206. if sock is not None:
  207. raise ValueError(
  208. 'path and sock can not be specified at the same time')
  209. path = os.fspath(path)
  210. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
  211. try:
  212. sock.setblocking(False)
  213. await self.sock_connect(sock, path)
  214. except:
  215. sock.close()
  216. raise
  217. else:
  218. if sock is None:
  219. raise ValueError('no path and sock were specified')
  220. if (sock.family != socket.AF_UNIX or
  221. sock.type != socket.SOCK_STREAM):
  222. raise ValueError(
  223. f'A UNIX Domain Stream Socket was expected, got {sock!r}')
  224. sock.setblocking(False)
  225. transport, protocol = await self._create_connection_transport(
  226. sock, protocol_factory, ssl, server_hostname,
  227. ssl_handshake_timeout=ssl_handshake_timeout)
  228. return transport, protocol
  229. async def create_unix_server(
  230. self, protocol_factory, path=None, *,
  231. sock=None, backlog=100, ssl=None,
  232. ssl_handshake_timeout=None,
  233. start_serving=True):
  234. if isinstance(ssl, bool):
  235. raise TypeError('ssl argument must be an SSLContext or None')
  236. if ssl_handshake_timeout is not None and not ssl:
  237. raise ValueError(
  238. 'ssl_handshake_timeout is only meaningful with ssl')
  239. if path is not None:
  240. if sock is not None:
  241. raise ValueError(
  242. 'path and sock can not be specified at the same time')
  243. path = os.fspath(path)
  244. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  245. # Check for abstract socket. `str` and `bytes` paths are supported.
  246. if path[0] not in (0, '\x00'):
  247. try:
  248. if stat.S_ISSOCK(os.stat(path).st_mode):
  249. os.remove(path)
  250. except FileNotFoundError:
  251. pass
  252. except OSError as err:
  253. # Directory may have permissions only to create socket.
  254. logger.error('Unable to check or remove stale UNIX socket '
  255. '%r: %r', path, err)
  256. try:
  257. sock.bind(path)
  258. except OSError as exc:
  259. sock.close()
  260. if exc.errno == errno.EADDRINUSE:
  261. # Let's improve the error message by adding
  262. # with what exact address it occurs.
  263. msg = f'Address {path!r} is already in use'
  264. raise OSError(errno.EADDRINUSE, msg) from None
  265. else:
  266. raise
  267. except:
  268. sock.close()
  269. raise
  270. else:
  271. if sock is None:
  272. raise ValueError(
  273. 'path was not specified, and no sock specified')
  274. if (sock.family != socket.AF_UNIX or
  275. sock.type != socket.SOCK_STREAM):
  276. raise ValueError(
  277. f'A UNIX Domain Stream Socket was expected, got {sock!r}')
  278. sock.setblocking(False)
  279. server = base_events.Server(self, [sock], protocol_factory,
  280. ssl, backlog, ssl_handshake_timeout)
  281. if start_serving:
  282. server._start_serving()
  283. # Skip one loop iteration so that all 'loop.add_reader'
  284. # go through.
  285. await tasks.sleep(0)
  286. return server
  287. async def _sock_sendfile_native(self, sock, file, offset, count):
  288. try:
  289. os.sendfile
  290. except AttributeError:
  291. raise exceptions.SendfileNotAvailableError(
  292. "os.sendfile() is not available")
  293. try:
  294. fileno = file.fileno()
  295. except (AttributeError, io.UnsupportedOperation) as err:
  296. raise exceptions.SendfileNotAvailableError("not a regular file")
  297. try:
  298. fsize = os.fstat(fileno).st_size
  299. except OSError:
  300. raise exceptions.SendfileNotAvailableError("not a regular file")
  301. blocksize = count if count else fsize
  302. if not blocksize:
  303. return 0 # empty file
  304. fut = self.create_future()
  305. self._sock_sendfile_native_impl(fut, None, sock, fileno,
  306. offset, count, blocksize, 0)
  307. return await fut
  308. def _sock_sendfile_native_impl(self, fut, registered_fd, sock, fileno,
  309. offset, count, blocksize, total_sent):
  310. fd = sock.fileno()
  311. if registered_fd is not None:
  312. # Remove the callback early. It should be rare that the
  313. # selector says the fd is ready but the call still returns
  314. # EAGAIN, and I am willing to take a hit in that case in
  315. # order to simplify the common case.
  316. self.remove_writer(registered_fd)
  317. if fut.cancelled():
  318. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  319. return
  320. if count:
  321. blocksize = count - total_sent
  322. if blocksize <= 0:
  323. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  324. fut.set_result(total_sent)
  325. return
  326. try:
  327. sent = os.sendfile(fd, fileno, offset, blocksize)
  328. except (BlockingIOError, InterruptedError):
  329. if registered_fd is None:
  330. self._sock_add_cancellation_callback(fut, sock)
  331. self.add_writer(fd, self._sock_sendfile_native_impl, fut,
  332. fd, sock, fileno,
  333. offset, count, blocksize, total_sent)
  334. except OSError as exc:
  335. if (registered_fd is not None and
  336. exc.errno == errno.ENOTCONN and
  337. type(exc) is not ConnectionError):
  338. # If we have an ENOTCONN and this isn't a first call to
  339. # sendfile(), i.e. the connection was closed in the middle
  340. # of the operation, normalize the error to ConnectionError
  341. # to make it consistent across all Posix systems.
  342. new_exc = ConnectionError(
  343. "socket is not connected", errno.ENOTCONN)
  344. new_exc.__cause__ = exc
  345. exc = new_exc
  346. if total_sent == 0:
  347. # We can get here for different reasons, the main
  348. # one being 'file' is not a regular mmap(2)-like
  349. # file, in which case we'll fall back on using
  350. # plain send().
  351. err = exceptions.SendfileNotAvailableError(
  352. "os.sendfile call failed")
  353. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  354. fut.set_exception(err)
  355. else:
  356. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  357. fut.set_exception(exc)
  358. except (SystemExit, KeyboardInterrupt):
  359. raise
  360. except BaseException as exc:
  361. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  362. fut.set_exception(exc)
  363. else:
  364. if sent == 0:
  365. # EOF
  366. self._sock_sendfile_update_filepos(fileno, offset, total_sent)
  367. fut.set_result(total_sent)
  368. else:
  369. offset += sent
  370. total_sent += sent
  371. if registered_fd is None:
  372. self._sock_add_cancellation_callback(fut, sock)
  373. self.add_writer(fd, self._sock_sendfile_native_impl, fut,
  374. fd, sock, fileno,
  375. offset, count, blocksize, total_sent)
  376. def _sock_sendfile_update_filepos(self, fileno, offset, total_sent):
  377. if total_sent > 0:
  378. os.lseek(fileno, offset, os.SEEK_SET)
  379. def _sock_add_cancellation_callback(self, fut, sock):
  380. def cb(fut):
  381. if fut.cancelled():
  382. fd = sock.fileno()
  383. if fd != -1:
  384. self.remove_writer(fd)
  385. fut.add_done_callback(cb)
  386. class _UnixReadPipeTransport(transports.ReadTransport):
  387. max_size = 256 * 1024 # max bytes we read in one event loop iteration
  388. def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
  389. super().__init__(extra)
  390. self._extra['pipe'] = pipe
  391. self._loop = loop
  392. self._pipe = pipe
  393. self._fileno = pipe.fileno()
  394. self._protocol = protocol
  395. self._closing = False
  396. self._paused = False
  397. mode = os.fstat(self._fileno).st_mode
  398. if not (stat.S_ISFIFO(mode) or
  399. stat.S_ISSOCK(mode) or
  400. stat.S_ISCHR(mode)):
  401. self._pipe = None
  402. self._fileno = None
  403. self._protocol = None
  404. raise ValueError("Pipe transport is for pipes/sockets only.")
  405. os.set_blocking(self._fileno, False)
  406. self._loop.call_soon(self._protocol.connection_made, self)
  407. # only start reading when connection_made() has been called
  408. self._loop.call_soon(self._loop._add_reader,
  409. self._fileno, self._read_ready)
  410. if waiter is not None:
  411. # only wake up the waiter when connection_made() has been called
  412. self._loop.call_soon(futures._set_result_unless_cancelled,
  413. waiter, None)
  414. def __repr__(self):
  415. info = [self.__class__.__name__]
  416. if self._pipe is None:
  417. info.append('closed')
  418. elif self._closing:
  419. info.append('closing')
  420. info.append(f'fd={self._fileno}')
  421. selector = getattr(self._loop, '_selector', None)
  422. if self._pipe is not None and selector is not None:
  423. polling = selector_events._test_selector_event(
  424. selector, self._fileno, selectors.EVENT_READ)
  425. if polling:
  426. info.append('polling')
  427. else:
  428. info.append('idle')
  429. elif self._pipe is not None:
  430. info.append('open')
  431. else:
  432. info.append('closed')
  433. return '<{}>'.format(' '.join(info))
  434. def _read_ready(self):
  435. try:
  436. data = os.read(self._fileno, self.max_size)
  437. except (BlockingIOError, InterruptedError):
  438. pass
  439. except OSError as exc:
  440. self._fatal_error(exc, 'Fatal read error on pipe transport')
  441. else:
  442. if data:
  443. self._protocol.data_received(data)
  444. else:
  445. if self._loop.get_debug():
  446. logger.info("%r was closed by peer", self)
  447. self._closing = True
  448. self._loop._remove_reader(self._fileno)
  449. self._loop.call_soon(self._protocol.eof_received)
  450. self._loop.call_soon(self._call_connection_lost, None)
  451. def pause_reading(self):
  452. if self._closing or self._paused:
  453. return
  454. self._paused = True
  455. self._loop._remove_reader(self._fileno)
  456. if self._loop.get_debug():
  457. logger.debug("%r pauses reading", self)
  458. def resume_reading(self):
  459. if self._closing or not self._paused:
  460. return
  461. self._paused = False
  462. self._loop._add_reader(self._fileno, self._read_ready)
  463. if self._loop.get_debug():
  464. logger.debug("%r resumes reading", self)
  465. def set_protocol(self, protocol):
  466. self._protocol = protocol
  467. def get_protocol(self):
  468. return self._protocol
  469. def is_closing(self):
  470. return self._closing
  471. def close(self):
  472. if not self._closing:
  473. self._close(None)
  474. def __del__(self, _warn=warnings.warn):
  475. if self._pipe is not None:
  476. _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
  477. self._pipe.close()
  478. def _fatal_error(self, exc, message='Fatal error on pipe transport'):
  479. # should be called by exception handler only
  480. if (isinstance(exc, OSError) and exc.errno == errno.EIO):
  481. if self._loop.get_debug():
  482. logger.debug("%r: %s", self, message, exc_info=True)
  483. else:
  484. self._loop.call_exception_handler({
  485. 'message': message,
  486. 'exception': exc,
  487. 'transport': self,
  488. 'protocol': self._protocol,
  489. })
  490. self._close(exc)
  491. def _close(self, exc):
  492. self._closing = True
  493. self._loop._remove_reader(self._fileno)
  494. self._loop.call_soon(self._call_connection_lost, exc)
  495. def _call_connection_lost(self, exc):
  496. try:
  497. self._protocol.connection_lost(exc)
  498. finally:
  499. self._pipe.close()
  500. self._pipe = None
  501. self._protocol = None
  502. self._loop = None
  503. class _UnixWritePipeTransport(transports._FlowControlMixin,
  504. transports.WriteTransport):
  505. def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
  506. super().__init__(extra, loop)
  507. self._extra['pipe'] = pipe
  508. self._pipe = pipe
  509. self._fileno = pipe.fileno()
  510. self._protocol = protocol
  511. self._buffer = bytearray()
  512. self._conn_lost = 0
  513. self._closing = False # Set when close() or write_eof() called.
  514. mode = os.fstat(self._fileno).st_mode
  515. is_char = stat.S_ISCHR(mode)
  516. is_fifo = stat.S_ISFIFO(mode)
  517. is_socket = stat.S_ISSOCK(mode)
  518. if not (is_char or is_fifo or is_socket):
  519. self._pipe = None
  520. self._fileno = None
  521. self._protocol = None
  522. raise ValueError("Pipe transport is only for "
  523. "pipes, sockets and character devices")
  524. os.set_blocking(self._fileno, False)
  525. self._loop.call_soon(self._protocol.connection_made, self)
  526. # On AIX, the reader trick (to be notified when the read end of the
  527. # socket is closed) only works for sockets. On other platforms it
  528. # works for pipes and sockets. (Exception: OS X 10.4? Issue #19294.)
  529. if is_socket or (is_fifo and not sys.platform.startswith("aix")):
  530. # only start reading when connection_made() has been called
  531. self._loop.call_soon(self._loop._add_reader,
  532. self._fileno, self._read_ready)
  533. if waiter is not None:
  534. # only wake up the waiter when connection_made() has been called
  535. self._loop.call_soon(futures._set_result_unless_cancelled,
  536. waiter, None)
  537. def __repr__(self):
  538. info = [self.__class__.__name__]
  539. if self._pipe is None:
  540. info.append('closed')
  541. elif self._closing:
  542. info.append('closing')
  543. info.append(f'fd={self._fileno}')
  544. selector = getattr(self._loop, '_selector', None)
  545. if self._pipe is not None and selector is not None:
  546. polling = selector_events._test_selector_event(
  547. selector, self._fileno, selectors.EVENT_WRITE)
  548. if polling:
  549. info.append('polling')
  550. else:
  551. info.append('idle')
  552. bufsize = self.get_write_buffer_size()
  553. info.append(f'bufsize={bufsize}')
  554. elif self._pipe is not None:
  555. info.append('open')
  556. else:
  557. info.append('closed')
  558. return '<{}>'.format(' '.join(info))
  559. def get_write_buffer_size(self):
  560. return len(self._buffer)
  561. def _read_ready(self):
  562. # Pipe was closed by peer.
  563. if self._loop.get_debug():
  564. logger.info("%r was closed by peer", self)
  565. if self._buffer:
  566. self._close(BrokenPipeError())
  567. else:
  568. self._close()
  569. def write(self, data):
  570. assert isinstance(data, (bytes, bytearray, memoryview)), repr(data)
  571. if isinstance(data, bytearray):
  572. data = memoryview(data)
  573. if not data:
  574. return
  575. if self._conn_lost or self._closing:
  576. if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
  577. logger.warning('pipe closed by peer or '
  578. 'os.write(pipe, data) raised exception.')
  579. self._conn_lost += 1
  580. return
  581. if not self._buffer:
  582. # Attempt to send it right away first.
  583. try:
  584. n = os.write(self._fileno, data)
  585. except (BlockingIOError, InterruptedError):
  586. n = 0
  587. except (SystemExit, KeyboardInterrupt):
  588. raise
  589. except BaseException as exc:
  590. self._conn_lost += 1
  591. self._fatal_error(exc, 'Fatal write error on pipe transport')
  592. return
  593. if n == len(data):
  594. return
  595. elif n > 0:
  596. data = memoryview(data)[n:]
  597. self._loop._add_writer(self._fileno, self._write_ready)
  598. self._buffer += data
  599. self._maybe_pause_protocol()
  600. def _write_ready(self):
  601. assert self._buffer, 'Data should not be empty'
  602. try:
  603. n = os.write(self._fileno, self._buffer)
  604. except (BlockingIOError, InterruptedError):
  605. pass
  606. except (SystemExit, KeyboardInterrupt):
  607. raise
  608. except BaseException as exc:
  609. self._buffer.clear()
  610. self._conn_lost += 1
  611. # Remove writer here, _fatal_error() doesn't it
  612. # because _buffer is empty.
  613. self._loop._remove_writer(self._fileno)
  614. self._fatal_error(exc, 'Fatal write error on pipe transport')
  615. else:
  616. if n == len(self._buffer):
  617. self._buffer.clear()
  618. self._loop._remove_writer(self._fileno)
  619. self._maybe_resume_protocol() # May append to buffer.
  620. if self._closing:
  621. self._loop._remove_reader(self._fileno)
  622. self._call_connection_lost(None)
  623. return
  624. elif n > 0:
  625. del self._buffer[:n]
  626. def can_write_eof(self):
  627. return True
  628. def write_eof(self):
  629. if self._closing:
  630. return
  631. assert self._pipe
  632. self._closing = True
  633. if not self._buffer:
  634. self._loop._remove_reader(self._fileno)
  635. self._loop.call_soon(self._call_connection_lost, None)
  636. def set_protocol(self, protocol):
  637. self._protocol = protocol
  638. def get_protocol(self):
  639. return self._protocol
  640. def is_closing(self):
  641. return self._closing
  642. def close(self):
  643. if self._pipe is not None and not self._closing:
  644. # write_eof is all what we needed to close the write pipe
  645. self.write_eof()
  646. def __del__(self, _warn=warnings.warn):
  647. if self._pipe is not None:
  648. _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
  649. self._pipe.close()
  650. def abort(self):
  651. self._close(None)
  652. def _fatal_error(self, exc, message='Fatal error on pipe transport'):
  653. # should be called by exception handler only
  654. if isinstance(exc, OSError):
  655. if self._loop.get_debug():
  656. logger.debug("%r: %s", self, message, exc_info=True)
  657. else:
  658. self._loop.call_exception_handler({
  659. 'message': message,
  660. 'exception': exc,
  661. 'transport': self,
  662. 'protocol': self._protocol,
  663. })
  664. self._close(exc)
  665. def _close(self, exc=None):
  666. self._closing = True
  667. if self._buffer:
  668. self._loop._remove_writer(self._fileno)
  669. self._buffer.clear()
  670. self._loop._remove_reader(self._fileno)
  671. self._loop.call_soon(self._call_connection_lost, exc)
  672. def _call_connection_lost(self, exc):
  673. try:
  674. self._protocol.connection_lost(exc)
  675. finally:
  676. self._pipe.close()
  677. self._pipe = None
  678. self._protocol = None
  679. self._loop = None
  680. class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
  681. def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
  682. stdin_w = None
  683. if stdin == subprocess.PIPE:
  684. # Use a socket pair for stdin, since not all platforms
  685. # support selecting read events on the write end of a
  686. # socket (which we use in order to detect closing of the
  687. # other end). Notably this is needed on AIX, and works
  688. # just fine on other platforms.
  689. stdin, stdin_w = socket.socketpair()
  690. try:
  691. self._proc = subprocess.Popen(
  692. args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
  693. universal_newlines=False, bufsize=bufsize, **kwargs)
  694. if stdin_w is not None:
  695. stdin.close()
  696. self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
  697. stdin_w = None
  698. finally:
  699. if stdin_w is not None:
  700. stdin.close()
  701. stdin_w.close()
  702. class AbstractChildWatcher:
  703. """Abstract base class for monitoring child processes.
  704. Objects derived from this class monitor a collection of subprocesses and
  705. report their termination or interruption by a signal.
  706. New callbacks are registered with .add_child_handler(). Starting a new
  707. process must be done within a 'with' block to allow the watcher to suspend
  708. its activity until the new process if fully registered (this is needed to
  709. prevent a race condition in some implementations).
  710. Example:
  711. with watcher:
  712. proc = subprocess.Popen("sleep 1")
  713. watcher.add_child_handler(proc.pid, callback)
  714. Notes:
  715. Implementations of this class must be thread-safe.
  716. Since child watcher objects may catch the SIGCHLD signal and call
  717. waitpid(-1), there should be only one active object per process.
  718. """
  719. def add_child_handler(self, pid, callback, *args):
  720. """Register a new child handler.
  721. Arrange for callback(pid, returncode, *args) to be called when
  722. process 'pid' terminates. Specifying another callback for the same
  723. process replaces the previous handler.
  724. Note: callback() must be thread-safe.
  725. """
  726. raise NotImplementedError()
  727. def remove_child_handler(self, pid):
  728. """Removes the handler for process 'pid'.
  729. The function returns True if the handler was successfully removed,
  730. False if there was nothing to remove."""
  731. raise NotImplementedError()
  732. def attach_loop(self, loop):
  733. """Attach the watcher to an event loop.
  734. If the watcher was previously attached to an event loop, then it is
  735. first detached before attaching to the new loop.
  736. Note: loop may be None.
  737. """
  738. raise NotImplementedError()
  739. def close(self):
  740. """Close the watcher.
  741. This must be called to make sure that any underlying resource is freed.
  742. """
  743. raise NotImplementedError()
  744. def is_active(self):
  745. """Return ``True`` if the watcher is active and is used by the event loop.
  746. Return True if the watcher is installed and ready to handle process exit
  747. notifications.
  748. """
  749. raise NotImplementedError()
  750. def __enter__(self):
  751. """Enter the watcher's context and allow starting new processes
  752. This function must return self"""
  753. raise NotImplementedError()
  754. def __exit__(self, a, b, c):
  755. """Exit the watcher's context"""
  756. raise NotImplementedError()
  757. class PidfdChildWatcher(AbstractChildWatcher):
  758. """Child watcher implementation using Linux's pid file descriptors.
  759. This child watcher polls process file descriptors (pidfds) to await child
  760. process termination. In some respects, PidfdChildWatcher is a "Goldilocks"
  761. child watcher implementation. It doesn't require signals or threads, doesn't
  762. interfere with any processes launched outside the event loop, and scales
  763. linearly with the number of subprocesses launched by the event loop. The
  764. main disadvantage is that pidfds are specific to Linux, and only work on
  765. recent (5.3+) kernels.
  766. """
  767. def __init__(self):
  768. self._loop = None
  769. self._callbacks = {}
  770. def __enter__(self):
  771. return self
  772. def __exit__(self, exc_type, exc_value, exc_traceback):
  773. pass
  774. def is_active(self):
  775. return self._loop is not None and self._loop.is_running()
  776. def close(self):
  777. self.attach_loop(None)
  778. def attach_loop(self, loop):
  779. if self._loop is not None and loop is None and self._callbacks:
  780. warnings.warn(
  781. 'A loop is being detached '
  782. 'from a child watcher with pending handlers',
  783. RuntimeWarning)
  784. for pidfd, _, _ in self._callbacks.values():
  785. self._loop._remove_reader(pidfd)
  786. os.close(pidfd)
  787. self._callbacks.clear()
  788. self._loop = loop
  789. def add_child_handler(self, pid, callback, *args):
  790. existing = self._callbacks.get(pid)
  791. if existing is not None:
  792. self._callbacks[pid] = existing[0], callback, args
  793. else:
  794. pidfd = os.pidfd_open(pid)
  795. self._loop._add_reader(pidfd, self._do_wait, pid)
  796. self._callbacks[pid] = pidfd, callback, args
  797. def _do_wait(self, pid):
  798. pidfd, callback, args = self._callbacks.pop(pid)
  799. self._loop._remove_reader(pidfd)
  800. try:
  801. _, status = os.waitpid(pid, 0)
  802. except ChildProcessError:
  803. # The child process is already reaped
  804. # (may happen if waitpid() is called elsewhere).
  805. returncode = 255
  806. logger.warning(
  807. "child process pid %d exit status already read: "
  808. " will report returncode 255",
  809. pid)
  810. else:
  811. returncode = waitstatus_to_exitcode(status)
  812. os.close(pidfd)
  813. callback(pid, returncode, *args)
  814. def remove_child_handler(self, pid):
  815. try:
  816. pidfd, _, _ = self._callbacks.pop(pid)
  817. except KeyError:
  818. return False
  819. self._loop._remove_reader(pidfd)
  820. os.close(pidfd)
  821. return True
  822. class BaseChildWatcher(AbstractChildWatcher):
  823. def __init__(self):
  824. self._loop = None
  825. self._callbacks = {}
  826. def close(self):
  827. self.attach_loop(None)
  828. def is_active(self):
  829. return self._loop is not None and self._loop.is_running()
  830. def _do_waitpid(self, expected_pid):
  831. raise NotImplementedError()
  832. def _do_waitpid_all(self):
  833. raise NotImplementedError()
  834. def attach_loop(self, loop):
  835. assert loop is None or isinstance(loop, events.AbstractEventLoop)
  836. if self._loop is not None and loop is None and self._callbacks:
  837. warnings.warn(
  838. 'A loop is being detached '
  839. 'from a child watcher with pending handlers',
  840. RuntimeWarning)
  841. if self._loop is not None:
  842. self._loop.remove_signal_handler(signal.SIGCHLD)
  843. self._loop = loop
  844. if loop is not None:
  845. loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)
  846. # Prevent a race condition in case a child terminated
  847. # during the switch.
  848. self._do_waitpid_all()
  849. def _sig_chld(self):
  850. try:
  851. self._do_waitpid_all()
  852. except (SystemExit, KeyboardInterrupt):
  853. raise
  854. except BaseException as exc:
  855. # self._loop should always be available here
  856. # as '_sig_chld' is added as a signal handler
  857. # in 'attach_loop'
  858. self._loop.call_exception_handler({
  859. 'message': 'Unknown exception in SIGCHLD handler',
  860. 'exception': exc,
  861. })
  862. class SafeChildWatcher(BaseChildWatcher):
  863. """'Safe' child watcher implementation.
  864. This implementation avoids disrupting other code spawning processes by
  865. polling explicitly each process in the SIGCHLD handler instead of calling
  866. os.waitpid(-1).
  867. This is a safe solution but it has a significant overhead when handling a
  868. big number of children (O(n) each time SIGCHLD is raised)
  869. """
  870. def close(self):
  871. self._callbacks.clear()
  872. super().close()
  873. def __enter__(self):
  874. return self
  875. def __exit__(self, a, b, c):
  876. pass
  877. def add_child_handler(self, pid, callback, *args):
  878. self._callbacks[pid] = (callback, args)
  879. # Prevent a race condition in case the child is already terminated.
  880. self._do_waitpid(pid)
  881. def remove_child_handler(self, pid):
  882. try:
  883. del self._callbacks[pid]
  884. return True
  885. except KeyError:
  886. return False
  887. def _do_waitpid_all(self):
  888. for pid in list(self._callbacks):
  889. self._do_waitpid(pid)
  890. def _do_waitpid(self, expected_pid):
  891. assert expected_pid > 0
  892. try:
  893. pid, status = os.waitpid(expected_pid, os.WNOHANG)
  894. except ChildProcessError:
  895. # The child process is already reaped
  896. # (may happen if waitpid() is called elsewhere).
  897. pid = expected_pid
  898. returncode = 255
  899. logger.warning(
  900. "Unknown child process pid %d, will report returncode 255",
  901. pid)
  902. else:
  903. if pid == 0:
  904. # The child process is still alive.
  905. return
  906. returncode = waitstatus_to_exitcode(status)
  907. if self._loop.get_debug():
  908. logger.debug('process %s exited with returncode %s',
  909. expected_pid, returncode)
  910. try:
  911. callback, args = self._callbacks.pop(pid)
  912. except KeyError: # pragma: no cover
  913. # May happen if .remove_child_handler() is called
  914. # after os.waitpid() returns.
  915. if self._loop.get_debug():
  916. logger.warning("Child watcher got an unexpected pid: %r",
  917. pid, exc_info=True)
  918. else:
  919. callback(pid, returncode, *args)
  920. class FastChildWatcher(BaseChildWatcher):
  921. """'Fast' child watcher implementation.
  922. This implementation reaps every terminated processes by calling
  923. os.waitpid(-1) directly, possibly breaking other code spawning processes
  924. and waiting for their termination.
  925. There is no noticeable overhead when handling a big number of children
  926. (O(1) each time a child terminates).
  927. """
  928. def __init__(self):
  929. super().__init__()
  930. self._lock = threading.Lock()
  931. self._zombies = {}
  932. self._forks = 0
  933. def close(self):
  934. self._callbacks.clear()
  935. self._zombies.clear()
  936. super().close()
  937. def __enter__(self):
  938. with self._lock:
  939. self._forks += 1
  940. return self
  941. def __exit__(self, a, b, c):
  942. with self._lock:
  943. self._forks -= 1
  944. if self._forks or not self._zombies:
  945. return
  946. collateral_victims = str(self._zombies)
  947. self._zombies.clear()
  948. logger.warning(
  949. "Caught subprocesses termination from unknown pids: %s",
  950. collateral_victims)
  951. def add_child_handler(self, pid, callback, *args):
  952. assert self._forks, "Must use the context manager"
  953. with self._lock:
  954. try:
  955. returncode = self._zombies.pop(pid)
  956. except KeyError:
  957. # The child is running.
  958. self._callbacks[pid] = callback, args
  959. return
  960. # The child is dead already. We can fire the callback.
  961. callback(pid, returncode, *args)
  962. def remove_child_handler(self, pid):
  963. try:
  964. del self._callbacks[pid]
  965. return True
  966. except KeyError:
  967. return False
  968. def _do_waitpid_all(self):
  969. # Because of signal coalescing, we must keep calling waitpid() as
  970. # long as we're able to reap a child.
  971. while True:
  972. try:
  973. pid, status = os.waitpid(-1, os.WNOHANG)
  974. except ChildProcessError:
  975. # No more child processes exist.
  976. return
  977. else:
  978. if pid == 0:
  979. # A child process is still alive.
  980. return
  981. returncode = waitstatus_to_exitcode(status)
  982. with self._lock:
  983. try:
  984. callback, args = self._callbacks.pop(pid)
  985. except KeyError:
  986. # unknown child
  987. if self._forks:
  988. # It may not be registered yet.
  989. self._zombies[pid] = returncode
  990. if self._loop.get_debug():
  991. logger.debug('unknown process %s exited '
  992. 'with returncode %s',
  993. pid, returncode)
  994. continue
  995. callback = None
  996. else:
  997. if self._loop.get_debug():
  998. logger.debug('process %s exited with returncode %s',
  999. pid, returncode)
  1000. if callback is None:
  1001. logger.warning(
  1002. "Caught subprocess termination from unknown pid: "
  1003. "%d -> %d", pid, returncode)
  1004. else:
  1005. callback(pid, returncode, *args)
  1006. class MultiLoopChildWatcher(AbstractChildWatcher):
  1007. """A watcher that doesn't require running loop in the main thread.
  1008. This implementation registers a SIGCHLD signal handler on
  1009. instantiation (which may conflict with other code that
  1010. install own handler for this signal).
  1011. The solution is safe but it has a significant overhead when
  1012. handling a big number of processes (*O(n)* each time a
  1013. SIGCHLD is received).
  1014. """
  1015. # Implementation note:
  1016. # The class keeps compatibility with AbstractChildWatcher ABC
  1017. # To achieve this it has empty attach_loop() method
  1018. # and doesn't accept explicit loop argument
  1019. # for add_child_handler()/remove_child_handler()
  1020. # but retrieves the current loop by get_running_loop()
  1021. def __init__(self):
  1022. self._callbacks = {}
  1023. self._saved_sighandler = None
  1024. def is_active(self):
  1025. return self._saved_sighandler is not None
  1026. def close(self):
  1027. self._callbacks.clear()
  1028. if self._saved_sighandler is None:
  1029. return
  1030. handler = signal.getsignal(signal.SIGCHLD)
  1031. if handler != self._sig_chld:
  1032. logger.warning("SIGCHLD handler was changed by outside code")
  1033. else:
  1034. signal.signal(signal.SIGCHLD, self._saved_sighandler)
  1035. self._saved_sighandler = None
  1036. def __enter__(self):
  1037. return self
  1038. def __exit__(self, exc_type, exc_val, exc_tb):
  1039. pass
  1040. def add_child_handler(self, pid, callback, *args):
  1041. loop = events.get_running_loop()
  1042. self._callbacks[pid] = (loop, callback, args)
  1043. # Prevent a race condition in case the child is already terminated.
  1044. self._do_waitpid(pid)
  1045. def remove_child_handler(self, pid):
  1046. try:
  1047. del self._callbacks[pid]
  1048. return True
  1049. except KeyError:
  1050. return False
  1051. def attach_loop(self, loop):
  1052. # Don't save the loop but initialize itself if called first time
  1053. # The reason to do it here is that attach_loop() is called from
  1054. # unix policy only for the main thread.
  1055. # Main thread is required for subscription on SIGCHLD signal
  1056. if self._saved_sighandler is not None:
  1057. return
  1058. self._saved_sighandler = signal.signal(signal.SIGCHLD, self._sig_chld)
  1059. if self._saved_sighandler is None:
  1060. logger.warning("Previous SIGCHLD handler was set by non-Python code, "
  1061. "restore to default handler on watcher close.")
  1062. self._saved_sighandler = signal.SIG_DFL
  1063. # Set SA_RESTART to limit EINTR occurrences.
  1064. signal.siginterrupt(signal.SIGCHLD, False)
  1065. def _do_waitpid_all(self):
  1066. for pid in list(self._callbacks):
  1067. self._do_waitpid(pid)
  1068. def _do_waitpid(self, expected_pid):
  1069. assert expected_pid > 0
  1070. try:
  1071. pid, status = os.waitpid(expected_pid, os.WNOHANG)
  1072. except ChildProcessError:
  1073. # The child process is already reaped
  1074. # (may happen if waitpid() is called elsewhere).
  1075. pid = expected_pid
  1076. returncode = 255
  1077. logger.warning(
  1078. "Unknown child process pid %d, will report returncode 255",
  1079. pid)
  1080. debug_log = False
  1081. else:
  1082. if pid == 0:
  1083. # The child process is still alive.
  1084. return
  1085. returncode = waitstatus_to_exitcode(status)
  1086. debug_log = True
  1087. try:
  1088. loop, callback, args = self._callbacks.pop(pid)
  1089. except KeyError: # pragma: no cover
  1090. # May happen if .remove_child_handler() is called
  1091. # after os.waitpid() returns.
  1092. logger.warning("Child watcher got an unexpected pid: %r",
  1093. pid, exc_info=True)
  1094. else:
  1095. if loop.is_closed():
  1096. logger.warning("Loop %r that handles pid %r is closed", loop, pid)
  1097. else:
  1098. if debug_log and loop.get_debug():
  1099. logger.debug('process %s exited with returncode %s',
  1100. expected_pid, returncode)
  1101. loop.call_soon_threadsafe(callback, pid, returncode, *args)
  1102. def _sig_chld(self, signum, frame):
  1103. try:
  1104. self._do_waitpid_all()
  1105. except (SystemExit, KeyboardInterrupt):
  1106. raise
  1107. except BaseException:
  1108. logger.warning('Unknown exception in SIGCHLD handler', exc_info=True)
  1109. class ThreadedChildWatcher(AbstractChildWatcher):
  1110. """Threaded child watcher implementation.
  1111. The watcher uses a thread per process
  1112. for waiting for the process finish.
  1113. It doesn't require subscription on POSIX signal
  1114. but a thread creation is not free.
  1115. The watcher has O(1) complexity, its performance doesn't depend
  1116. on amount of spawn processes.
  1117. """
  1118. def __init__(self):
  1119. self._pid_counter = itertools.count(0)
  1120. self._threads = {}
  1121. def is_active(self):
  1122. return True
  1123. def close(self):
  1124. self._join_threads()
  1125. def _join_threads(self):
  1126. """Internal: Join all non-daemon threads"""
  1127. threads = [thread for thread in list(self._threads.values())
  1128. if thread.is_alive() and not thread.daemon]
  1129. for thread in threads:
  1130. thread.join()
  1131. def __enter__(self):
  1132. return self
  1133. def __exit__(self, exc_type, exc_val, exc_tb):
  1134. pass
  1135. def __del__(self, _warn=warnings.warn):
  1136. threads = [thread for thread in list(self._threads.values())
  1137. if thread.is_alive()]
  1138. if threads:
  1139. _warn(f"{self.__class__} has registered but not finished child processes",
  1140. ResourceWarning,
  1141. source=self)
  1142. def add_child_handler(self, pid, callback, *args):
  1143. loop = events.get_running_loop()
  1144. thread = threading.Thread(target=self._do_waitpid,
  1145. name=f"waitpid-{next(self._pid_counter)}",
  1146. args=(loop, pid, callback, args),
  1147. daemon=True)
  1148. self._threads[pid] = thread
  1149. thread.start()
  1150. def remove_child_handler(self, pid):
  1151. # asyncio never calls remove_child_handler() !!!
  1152. # The method is no-op but is implemented because
  1153. # abstract base classe requires it
  1154. return True
  1155. def attach_loop(self, loop):
  1156. pass
  1157. def _do_waitpid(self, loop, expected_pid, callback, args):
  1158. assert expected_pid > 0
  1159. try:
  1160. pid, status = os.waitpid(expected_pid, 0)
  1161. except ChildProcessError:
  1162. # The child process is already reaped
  1163. # (may happen if waitpid() is called elsewhere).
  1164. pid = expected_pid
  1165. returncode = 255
  1166. logger.warning(
  1167. "Unknown child process pid %d, will report returncode 255",
  1168. pid)
  1169. else:
  1170. returncode = waitstatus_to_exitcode(status)
  1171. if loop.get_debug():
  1172. logger.debug('process %s exited with returncode %s',
  1173. expected_pid, returncode)
  1174. if loop.is_closed():
  1175. logger.warning("Loop %r that handles pid %r is closed", loop, pid)
  1176. else:
  1177. loop.call_soon_threadsafe(callback, pid, returncode, *args)
  1178. self._threads.pop(expected_pid)
  1179. class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
  1180. """UNIX event loop policy with a watcher for child processes."""
  1181. _loop_factory = _UnixSelectorEventLoop
  1182. def __init__(self):
  1183. super().__init__()
  1184. self._watcher = None
  1185. def _init_watcher(self):
  1186. with events._lock:
  1187. if self._watcher is None: # pragma: no branch
  1188. self._watcher = ThreadedChildWatcher()
  1189. if threading.current_thread() is threading.main_thread():
  1190. self._watcher.attach_loop(self._local._loop)
  1191. def set_event_loop(self, loop):
  1192. """Set the event loop.
  1193. As a side effect, if a child watcher was set before, then calling
  1194. .set_event_loop() from the main thread will call .attach_loop(loop) on
  1195. the child watcher.
  1196. """
  1197. super().set_event_loop(loop)
  1198. if (self._watcher is not None and
  1199. threading.current_thread() is threading.main_thread()):
  1200. self._watcher.attach_loop(loop)
  1201. def get_child_watcher(self):
  1202. """Get the watcher for child processes.
  1203. If not yet set, a ThreadedChildWatcher object is automatically created.
  1204. """
  1205. if self._watcher is None:
  1206. self._init_watcher()
  1207. return self._watcher
  1208. def set_child_watcher(self, watcher):
  1209. """Set the watcher for child processes."""
  1210. assert watcher is None or isinstance(watcher, AbstractChildWatcher)
  1211. if self._watcher is not None:
  1212. self._watcher.close()
  1213. self._watcher = watcher
  1214. SelectorEventLoop = _UnixSelectorEventLoop
  1215. DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy