__all__ = (
'SelectorEventLoop',
- 'AbstractChildWatcher',
- 'PidfdChildWatcher',
- 'ThreadedChildWatcher',
'DefaultEventLoopPolicy',
'EventLoop',
)
super().__init__(selector)
self._signal_handlers = {}
self._unix_server_sockets = {}
+ if can_use_pidfd():
+ self._watcher = _PidfdChildWatcher()
+ else:
+ self._watcher = _ThreadedChildWatcher()
def close(self):
super().close()
async def _make_subprocess_transport(self, protocol, args, shell,
stdin, stdout, stderr, bufsize,
extra=None, **kwargs):
- with warnings.catch_warnings():
- warnings.simplefilter('ignore', DeprecationWarning)
- watcher = events.get_event_loop_policy()._watcher
-
- with watcher:
- if not watcher.is_active():
- # Check early.
- # Raising exception before process creation
- # prevents subprocess execution if the watcher
- # is not ready to handle it.
- raise RuntimeError("asyncio.get_child_watcher() is not activated, "
- "subprocess support is not installed.")
- waiter = self.create_future()
- transp = _UnixSubprocessTransport(self, protocol, args, shell,
- stdin, stdout, stderr, bufsize,
- waiter=waiter, extra=extra,
- **kwargs)
- watcher.add_child_handler(transp.get_pid(),
- self._child_watcher_callback, transp)
- try:
- await waiter
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException:
- transp.close()
- await transp._wait()
- raise
+ watcher = self._watcher
+ waiter = self.create_future()
+ transp = _UnixSubprocessTransport(self, protocol, args, shell,
+ stdin, stdout, stderr, bufsize,
+ waiter=waiter, extra=extra,
+ **kwargs)
+ watcher.add_child_handler(transp.get_pid(),
+ self._child_watcher_callback, transp)
+ try:
+ await waiter
+ except (SystemExit, KeyboardInterrupt):
+ raise
+ except BaseException:
+ transp.close()
+ await transp._wait()
+ raise
return transp
stdin_w.close()
-class AbstractChildWatcher:
- """Abstract base class for monitoring child processes.
-
- Objects derived from this class monitor a collection of subprocesses and
- report their termination or interruption by a signal.
-
- New callbacks are registered with .add_child_handler(). Starting a new
- process must be done within a 'with' block to allow the watcher to suspend
- its activity until the new process if fully registered (this is needed to
- prevent a race condition in some implementations).
-
- Example:
- with watcher:
- proc = subprocess.Popen("sleep 1")
- watcher.add_child_handler(proc.pid, callback)
-
- Notes:
- Implementations of this class must be thread-safe.
-
- Since child watcher objects may catch the SIGCHLD signal and call
- waitpid(-1), there should be only one active object per process.
- """
-
- def __init_subclass__(cls) -> None:
- if cls.__module__ != __name__:
- warnings._deprecated("AbstractChildWatcher",
- "{name!r} is deprecated as of Python 3.12 and will be "
- "removed in Python {remove}.",
- remove=(3, 14))
-
- def add_child_handler(self, pid, callback, *args):
- """Register a new child handler.
-
- Arrange for callback(pid, returncode, *args) to be called when
- process 'pid' terminates. Specifying another callback for the same
- process replaces the previous handler.
-
- Note: callback() must be thread-safe.
- """
- raise NotImplementedError()
-
- def remove_child_handler(self, pid):
- """Removes the handler for process 'pid'.
-
- The function returns True if the handler was successfully removed,
- False if there was nothing to remove."""
-
- raise NotImplementedError()
-
- def attach_loop(self, loop):
- """Attach the watcher to an event loop.
-
- If the watcher was previously attached to an event loop, then it is
- first detached before attaching to the new loop.
-
- Note: loop may be None.
- """
- raise NotImplementedError()
-
- def close(self):
- """Close the watcher.
-
- This must be called to make sure that any underlying resource is freed.
- """
- raise NotImplementedError()
-
- def is_active(self):
- """Return ``True`` if the watcher is active and is used by the event loop.
-
- Return True if the watcher is installed and ready to handle process exit
- notifications.
-
- """
- raise NotImplementedError()
-
- def __enter__(self):
- """Enter the watcher's context and allow starting new processes
-
- This function must return self"""
- raise NotImplementedError()
-
- def __exit__(self, a, b, c):
- """Exit the watcher's context"""
- raise NotImplementedError()
-
-
-class PidfdChildWatcher(AbstractChildWatcher):
+class _PidfdChildWatcher:
"""Child watcher implementation using Linux's pid file descriptors.
This child watcher polls process file descriptors (pidfds) to await child
recent (5.3+) kernels.
"""
- def __enter__(self):
- return self
-
- def __exit__(self, exc_type, exc_value, exc_traceback):
- pass
-
def is_active(self):
return True
- def close(self):
- pass
-
- def attach_loop(self, loop):
- pass
-
def add_child_handler(self, pid, callback, *args):
loop = events.get_running_loop()
pidfd = os.pidfd_open(pid)
os.close(pidfd)
callback(pid, returncode, *args)
- def remove_child_handler(self, pid):
- # asyncio never calls remove_child_handler() !!!
- # The method is no-op but is implemented because
- # abstract base classes require it.
- return True
-
-
-class ThreadedChildWatcher(AbstractChildWatcher):
+class _ThreadedChildWatcher:
"""Threaded child watcher implementation.
The watcher uses a thread per process
def is_active(self):
return True
- def close(self):
- pass
-
- def __enter__(self):
- return self
-
- def __exit__(self, exc_type, exc_val, exc_tb):
- pass
-
def __del__(self, _warn=warnings.warn):
threads = [thread for thread in list(self._threads.values())
if thread.is_alive()]
self._threads[pid] = thread
thread.start()
- def remove_child_handler(self, pid):
- # asyncio never calls remove_child_handler() !!!
- # The method is no-op but is implemented because
- # abstract base classes require it.
- return True
-
- def attach_loop(self, loop):
- pass
-
def _do_waitpid(self, loop, expected_pid, callback, args):
assert expected_pid > 0
class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
- """UNIX event loop policy with a watcher for child processes."""
+ """UNIX event loop policy"""
_loop_factory = _UnixSelectorEventLoop
- def __init__(self):
- super().__init__()
- if can_use_pidfd():
- self._watcher = PidfdChildWatcher()
- else:
- self._watcher = ThreadedChildWatcher()
-
- def set_event_loop(self, loop):
- """Set the event loop.
-
- As a side effect, if a child watcher was set before, then calling
- .set_event_loop() from the main thread will call .attach_loop(loop) on
- the child watcher.
- """
-
- super().set_event_loop(loop)
-
- if (self._watcher is not None and
- threading.current_thread() is threading.main_thread()):
- self._watcher.attach_loop(loop)
SelectorEventLoop = _UnixSelectorEventLoop
DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy
projects / thirdparty / Python / cpython.git / commitdiff
