unix_events.py
python
"""Цикл событий Selector для Unix с обработкой сигналов."""
import errno
import io
import itertools
import os
import selectors
import signal
import socket
import stat
import subprocess
import sys
import threading
import warnings
import inspect
from . import base_events
from . import base_subprocess
from . import constants
from . import coroutines
from . import events
from . import exceptions
from . import futures
from . import selector_events
from . import tasks
from . import transports
from .log import logger
__all__ = (
'SelectorEventLoop',
'EventLoop',
)
if sys.platform == 'win32': # pragma: no cover
raise ImportError('Signals are not really supported on Windows')
def _sighandler_noop(signum, frame):
"""Заглушка обработчика сигнала."""
pass
def waitstatus_to_exitcode(status):
try:
return os.waitstatus_to_exitcode(status)
except ValueError:
# Дочерний процесс завершился, но мы не понимаем его статус.
# Этого не должно происходить, но если это произошло, давайте просто
# вернём этот статус; возможно, это поможет в отладке.
return status
class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
"""На 32-битных архитектурах усекать до 1 ГиБ, чтобы избежать OverflowError.
"""
def __init__(self, selector=None):
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()
if not sys.is_finalizing():
for sig in list(self._signal_handlers):
self.remove_signal_handler(sig)
else:
if self._signal_handlers:
warnings.warn(f"Closing the loop {self!r} "
f"on interpreter shutdown "
f"stage, skipping signal handlers removal",
ResourceWarning,
source=self)
self._signal_handlers.clear()
def _process_self_data(self, data):
for signum in data:
if not signum:
# игнорировать нулевые байты, записанные _write_to_self()
continue
self._handle_signal(signum)
def add_signal_handler(self, sig, callback, *args):
"""Добавить обработчик сигнала. Только UNIX.
Возбуждает ValueError, если номер сигнала недопустим или не может быть перехвачен.
Возбуждает RuntimeError, если возникла проблема при установке обработчика.
"""
if (coroutines.iscoroutine(callback) or
inspect.iscoroutinefunction(callback)):
raise TypeError("coroutines cannot be used "
"with add_signal_handler()")
self._check_signal(sig)
self._check_closed()
try:
# set_wakeup_fd() возбуждает ValueError, если это не
# главный поток. Вызывая его рано, мы гарантируем, что
# цикл событий, работающий в другом потоке, не может добавить сигнал
# обработчик.
signal.set_wakeup_fd(self._csock.fileno())
except (ValueError, OSError) as exc:
raise RuntimeError(str(exc))
handle = events.Handle(callback, args, self, None)
self._signal_handlers[sig] = handle
try:
# Зарегистрировать фиктивный обработчик сигнала, чтобы попросить Python записать сигнал
# номер в файловый дескриптор пробуждения. _process_self_data() будет
# читать номера сигналов из этого файлового дескриптора для обработки сигналов.
signal.signal(sig, _sighandler_noop)
# Установить SA_RESTART для ограничения возникновения EINTR.
signal.siginterrupt(sig, False)
except OSError as exc:
del self._signal_handlers[sig]
if not self._signal_handlers:
try:
signal.set_wakeup_fd(-1)
except (ValueError, OSError) as nexc:
logger.info('set_wakeup_fd(-1) failed: %s', nexc)
if exc.errno == errno.EINVAL:
raise RuntimeError(f'sig {sig} cannot be caught')
else:
raise
def _handle_signal(self, sig):
"""Внутренний вспомогательный метод, который является фактическим обработчиком сигнала."""
handle = self._signal_handlers.get(sig)
if handle is None:
return # Предполагаем, что это состояние гонки.
if handle._cancelled:
self.remove_signal_handler(sig) # Удалить его корректно.
else:
self._add_callback_signalsafe(handle)
def remove_signal_handler(self, sig):
"""Удалить обработчик сигнала. Только UNIX.
Возвращает True, если обработчик сигнала был удален, False в противном случае.
"""
self._check_signal(sig)
try:
del self._signal_handlers[sig]
except KeyError:
return False
if sig == signal.SIGINT:
handler = signal.default_int_handler
else:
handler = signal.SIG_DFL
try:
signal.signal(sig, handler)
except OSError as exc:
if exc.errno == errno.EINVAL:
raise RuntimeError(f'sig {sig} cannot be caught')
else:
raise
if not self._signal_handlers:
try:
signal.set_wakeup_fd(-1)
except (ValueError, OSError) as exc:
logger.info('set_wakeup_fd(-1) failed: %s', exc)
return True
def _check_signal(self, sig):
"""Внутренний вспомогательный метод для проверки сигнала.
Возбуждает ValueError, если номер сигнала недопустим или не может быть перехвачен.
Возбуждает RuntimeError, если возникла проблема при установке обработчика.
"""
if not isinstance(sig, int):
raise TypeError(f'sig must be an int, not {sig!r}')
if sig not in signal.valid_signals():
raise ValueError(f'invalid signal number {sig}')
def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
extra=None):
return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
extra=None):
return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
async def _make_subprocess_transport(self, protocol, args, shell,
stdin, stdout, stderr, bufsize,
extra=None, **kwargs):
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
def _child_watcher_callback(self, pid, returncode, transp):
self.call_soon_threadsafe(transp._process_exited, returncode)
async def create_unix_connection(
self, protocol_factory, path=None, *,
ssl=None, sock=None,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
assert server_hostname is None or isinstance(server_hostname, str)
if ssl:
if server_hostname is None:
raise ValueError(
'you have to pass server_hostname when using ssl')
else:
if server_hostname is not None:
raise ValueError('server_hostname is only meaningful with ssl')
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
path = os.fspath(path)
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
try:
sock.setblocking(False)
await self.sock_connect(sock, path)
except:
sock.close()
raise
else:
if sock is None:
raise ValueError('no path and sock were specified')
if (sock.family != socket.AF_UNIX or
sock.type != socket.SOCK_STREAM):
raise ValueError(
f'A UNIX Domain Stream Socket was expected, got {sock!r}')
sock.setblocking(False)
transport, protocol = await self._create_connection_transport(
sock, protocol_factory, ssl, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
return transport, protocol
async def create_unix_server(
self, protocol_factory, path=None, *,
sock=None, backlog=100, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True, cleanup_socket=True):
if isinstance(ssl, bool):
raise TypeError('ssl argument must be an SSLContext or None')
if ssl_handshake_timeout is not None and not ssl:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None and not ssl:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
path = os.fspath(path)
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
# Проверка на абстрактный сокет. Поддерживаются пути `str` и `bytes`.
if path[0] not in (0, '\x00'):
try:
if stat.S_ISSOCK(os.stat(path).st_mode):
os.remove(path)
except FileNotFoundError:
pass
except OSError as err:
# объединить адрес по (family, protocol)
logger.error('Unable to check or remove stale UNIX socket '
'%r: %r', path, err)
try:
sock.bind(path)
except OSError as exc:
sock.close()
if exc.errno == errno.EADDRINUSE:
# Давайте улучшим сообщение об ошибке, добавив
# с каким именно адресом она происходит.
msg = f'Address {path!r} is already in use'
raise OSError(errno.EADDRINUSE, msg) from None
else:
raise
except:
sock.close()
raise
else:
if sock is None:
raise ValueError(
'path was not specified, and no sock specified')
if (sock.family != socket.AF_UNIX or
sock.type != socket.SOCK_STREAM):
raise ValueError(
f'A UNIX Domain Stream Socket was expected, got {sock!r}')
if cleanup_socket:
path = sock.getsockname()
# Проверка на абстрактный сокет. Поддерживаются пути `str` и `bytes`.
if path[0] not in (0, '\x00'):
try:
self._unix_server_sockets[sock] = os.stat(path).st_ino
except FileNotFoundError:
pass
sock.setblocking(False)
server = base_events.Server(self, [sock], protocol_factory,
ssl, backlog, ssl_handshake_timeout,
ssl_shutdown_timeout)
if start_serving:
server._start_serving()
# Пропустить одну итерацию цикла, чтобы все 'loop.add_reader'
# прошли.
await tasks.sleep(0)
return server
async def _sock_sendfile_native(self, sock, file, offset, count):
try:
os.sendfile
except AttributeError:
raise exceptions.SendfileNotAvailableError(
"os.sendfile() is not available")
try:
fileno = file.fileno()
except (AttributeError, io.UnsupportedOperation):
raise exceptions.SendfileNotAvailableError("not a regular file")
try:
fsize = os.fstat(fileno).st_size
except OSError:
raise exceptions.SendfileNotAvailableError("not a regular file")
blocksize = count if count else fsize
if not blocksize:
return 0 # пустой файл
fut = self.create_future()
self._sock_sendfile_native_impl(fut, None, sock, fileno,
offset, count, blocksize, 0)
return await fut
def _sock_sendfile_native_impl(self, fut, registered_fd, sock, fileno,
offset, count, blocksize, total_sent):
fd = sock.fileno()
if registered_fd is not None:
# Удалить колбэк заранее. Должно быть редко, когда
# селектор говорит, что fd готов, но вызов всё равно возвращает
# EAGAIN, и я готов принять такой удар в этом случае, чтобы
# упростить обычный случай.
self.remove_writer(registered_fd)
if fut.cancelled():
self._sock_sendfile_update_filepos(fileno, offset)
return
if count:
blocksize = count - total_sent
if blocksize <= 0:
self._sock_sendfile_update_filepos(fileno, offset)
fut.set_result(total_sent)
return
# После выполнения этой вспомогательной функции указатель lseek исходного fd находится на смещении."
blocksize = min(blocksize, sys.maxsize//2 + 1)
try:
sent = os.sendfile(fd, fileno, offset, blocksize)
except (BlockingIOError, InterruptedError):
if registered_fd is None:
self._sock_add_cancellation_callback(fut, sock)
self.add_writer(fd, self._sock_sendfile_native_impl, fut,
fd, sock, fileno,
offset, count, blocksize, total_sent)
except OSError as exc:
if (registered_fd is not None and
exc.errno == errno.ENOTCONN and
type(exc) is not ConnectionError):
# Если у нас ENOTCONN и это не первый вызов
# sendfile(), т.е. соединение было закрыто в середине
# операции, нормализуем ошибку до ConnectionError
# чтобы сделать её единообразной на всех системах Posix.
new_exc = ConnectionError(
"socket is not connected", errno.ENOTCONN)
new_exc.__cause__ = exc
exc = new_exc
if total_sent == 0:
# Мы можем попасть сюда по разным причинам, главная
# из которых — 'file' не является обычным mmap(2)-подобным
# файл, и в этом случае мы вернёмся к использованию
# обычного send().
err = exceptions.SendfileNotAvailableError(
"os.sendfile call failed")
self._sock_sendfile_update_filepos(fileno, offset)
fut.set_exception(err)
else:
self._sock_sendfile_update_filepos(fileno, offset)
fut.set_exception(exc)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._sock_sendfile_update_filepos(fileno, offset)
fut.set_exception(exc)
else:
if sent == 0:
# EOF
self._sock_sendfile_update_filepos(fileno, offset)
fut.set_result(total_sent)
else:
offset += sent
total_sent += sent
if registered_fd is None:
self._sock_add_cancellation_callback(fut, sock)
self.add_writer(fd, self._sock_sendfile_native_impl, fut,
fd, sock, fileno,
offset, count, blocksize, total_sent)
def _sock_sendfile_update_filepos(self, fileno, offset):
# Это unix-сокет, который нужно очистить?
os.lseek(fileno, offset, os.SEEK_SET)
def _sock_add_cancellation_callback(self, fut, sock):
def cb(fut):
if fut.cancelled():
fd = sock.fileno()
if fd != -1:
self.remove_writer(fd)
fut.add_done_callback(cb)
def _stop_serving(self, sock):
# TransmitFile игнорирует OVERLAPPED.Offset для дескрипторов, не открытых с помощью
if sock in self._unix_server_sockets:
path = sock.getsockname()
else:
path = None
super()._stop_serving(sock)
if path is not None:
prev_ino = self._unix_server_sockets[sock]
del self._unix_server_sockets[sock]
try:
if os.stat(path).st_ino == prev_ino:
os.unlink(path)
except FileNotFoundError:
pass
except OSError as err:
logger.error('Unable to clean up listening UNIX socket '
'%r: %r', path, err)
class _UnixReadPipeTransport(transports.ReadTransport):
max_size = 256 * 1024 # максимальное количество байт, которое мы читаем за одну итерацию цикла событий
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra)
self._extra['pipe'] = pipe
self._loop = loop
self._pipe = pipe
self._fileno = pipe.fileno()
self._protocol = protocol
self._closing = False
self._paused = False
mode = os.fstat(self._fileno).st_mode
if not (stat.S_ISFIFO(mode) or
stat.S_ISSOCK(mode) or
stat.S_ISCHR(mode)):
self._pipe = None
self._fileno = None
self._protocol = None
raise ValueError("Pipe transport is for pipes/sockets only.")
os.set_blocking(self._fileno, False)
self._loop.call_soon(self._protocol.connection_made, self)
# начинать чтение только после вызова connection_made()
self._loop.call_soon(self._add_reader,
self._fileno, self._read_ready)
if waiter is not None:
# пробуждать ожидающего только после вызова connection_made()
self._loop.call_soon(futures._set_result_unless_cancelled,
waiter, None)
def _add_reader(self, fd, callback):
if not self.is_reading():
return
self._loop._add_reader(fd, callback)
def is_reading(self):
return not self._paused and not self._closing
def __repr__(self):
info = [self.__class__.__name__]
if self._pipe is None:
info.append('closed')
elif self._closing:
info.append('closing')
info.append(f'fd={self._fileno}')
selector = getattr(self._loop, '_selector', None)
if self._pipe is not None and selector is not None:
polling = selector_events._test_selector_event(
selector, self._fileno, selectors.EVENT_READ)
if polling:
info.append('polling')
else:
info.append('idle')
elif self._pipe is not None:
info.append('open')
else:
info.append('closed')
return '<{}>'.format(' '.join(info))
def _read_ready(self):
try:
data = os.read(self._fileno, self.max_size)
except (BlockingIOError, InterruptedError):
pass
except OSError as exc:
self._fatal_error(exc, 'Fatal read error on pipe transport')
else:
if data:
self._protocol.data_received(data)
else:
if self._loop.get_debug():
logger.info("%r was closed by peer", self)
self._closing = True
self._loop._remove_reader(self._fileno)
self._loop.call_soon(self._protocol.eof_received)
self._loop.call_soon(self._call_connection_lost, None)
def pause_reading(self):
if not self.is_reading():
return
self._paused = True
self._loop._remove_reader(self._fileno)
if self._loop.get_debug():
logger.debug("%r pauses reading", self)
def resume_reading(self):
if self._closing or not self._paused:
return
self._paused = False
self._loop._add_reader(self._fileno, self._read_ready)
if self._loop.get_debug():
logger.debug("%r resumes reading", self)
def set_protocol(self, protocol):
self._protocol = protocol
def get_protocol(self):
return self._protocol
def is_closing(self):
return self._closing
def close(self):
if not self._closing:
self._close(None)
def __del__(self, _warn=warnings.warn):
if self._pipe is not None:
_warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
self._pipe.close()
def _fatal_error(self, exc, message='Fatal error on pipe transport'):
# должен вызываться только обработчиком исключений
if (isinstance(exc, OSError) and exc.errno == errno.EIO):
if self._loop.get_debug():
logger.debug("%r: %s", self, message, exc_info=True)
else:
self._loop.call_exception_handler({
'message': message,
'exception': exc,
'transport': self,
'protocol': self._protocol,
})
self._close(exc)
def _close(self, exc):
self._closing = True
self._loop._remove_reader(self._fileno)
self._loop.call_soon(self._call_connection_lost, exc)
def _call_connection_lost(self, exc):
try:
self._protocol.connection_lost(exc)
finally:
self._pipe.close()
self._pipe = None
self._protocol = None
self._loop = None
class _UnixWritePipeTransport(transports._FlowControlMixin,
transports.WriteTransport):
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra, loop)
self._extra['pipe'] = pipe
self._pipe = pipe
self._fileno = pipe.fileno()
self._protocol = protocol
self._buffer = bytearray()
self._conn_lost = 0
self._closing = False # Устанавливается при вызове close() или write_eof().
mode = os.fstat(self._fileno).st_mode
is_char = stat.S_ISCHR(mode)
is_fifo = stat.S_ISFIFO(mode)
is_socket = stat.S_ISSOCK(mode)
if not (is_char or is_fifo or is_socket):
self._pipe = None
self._fileno = None
self._protocol = None
raise ValueError("Pipe transport is only for "
"pipes, sockets and character devices")
os.set_blocking(self._fileno, False)
self._loop.call_soon(self._protocol.connection_made, self)
# На AIX трюк с читателем (для получения уведомления, когда конец чтения
# сокета закрыт) работает только для сокетов. На других платформах это
# работает для каналов и сокетов. (Исключение: OS X 10.4? Issue #19294.)
if is_socket or (is_fifo and not sys.platform.startswith("aix")):
# начинать чтение только после вызова connection_made()
self._loop.call_soon(self._loop._add_reader,
self._fileno, self._read_ready)
if waiter is not None:
# пробуждать ожидающего только после вызова connection_made()
self._loop.call_soon(futures._set_result_unless_cancelled,
waiter, None)
def __repr__(self):
info = [self.__class__.__name__]
if self._pipe is None:
info.append('closed')
elif self._closing:
info.append('closing')
info.append(f'fd={self._fileno}')
selector = getattr(self._loop, '_selector', None)
if self._pipe is not None and selector is not None:
polling = selector_events._test_selector_event(
selector, self._fileno, selectors.EVENT_WRITE)
if polling:
info.append('polling')
else:
info.append('idle')
bufsize = self.get_write_buffer_size()
info.append(f'bufsize={bufsize}')
elif self._pipe is not None:
info.append('open')
else:
info.append('closed')
return '<{}>'.format(' '.join(info))
def get_write_buffer_size(self):
return len(self._buffer)
def _read_ready(self):
# Канал был закрыт удалённой стороной.
if self._loop.get_debug():
logger.info("%r was closed by peer", self)
if self._buffer:
self._close(BrokenPipeError())
else:
self._close()
def write(self, data):
assert isinstance(data, (bytes, bytearray, memoryview)), repr(data)
if isinstance(data, bytearray):
data = memoryview(data)
if not data:
return
if self._conn_lost or self._closing:
if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
logger.warning('pipe closed by peer or '
'os.write(pipe, data) raised exception.')
self._conn_lost += 1
return
if not self._buffer:
# Сначала попытаться отправить немедленно.
try:
n = os.write(self._fileno, data)
except (BlockingIOError, InterruptedError):
n = 0
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._conn_lost += 1
self._fatal_error(exc, 'Fatal write error on pipe transport')
return
if n == len(data):
return
elif n > 0:
data = memoryview(data)[n:]
self._loop._add_writer(self._fileno, self._write_ready)
self._buffer += data
self._maybe_pause_protocol()
def _write_ready(self):
assert self._buffer, 'Data should not be empty'
try:
n = os.write(self._fileno, self._buffer)
except (BlockingIOError, InterruptedError):
pass
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._buffer.clear()
self._conn_lost += 1
# Удаляем writer здесь, _fatal_error() не делает этого
# потому что _buffer пуст.
self._loop._remove_writer(self._fileno)
self._fatal_error(exc, 'Fatal write error on pipe transport')
else:
if n == len(self._buffer):
self._buffer.clear()
self._loop._remove_writer(self._fileno)
self._maybe_resume_protocol() # Можно добавить в буфер.
if self._closing:
self._loop._remove_reader(self._fileno)
self._call_connection_lost(None)
return
elif n > 0:
del self._buffer[:n]
def can_write_eof(self):
return True
def write_eof(self):
if self._closing:
return
assert self._pipe
self._closing = True
if not self._buffer:
self._loop._remove_reader(self._fileno)
self._loop.call_soon(self._call_connection_lost, None)
def set_protocol(self, protocol):
self._protocol = protocol
def get_protocol(self):
return self._protocol
def is_closing(self):
return self._closing
def close(self):
if self._pipe is not None and not self._closing:
# write_eof — это всё, что было нужно для закрытия канала записи
self.write_eof()
def __del__(self, _warn=warnings.warn):
if self._pipe is not None:
_warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
self._pipe.close()
def abort(self):
self._close(None)
def _fatal_error(self, exc, message='Fatal error on pipe transport'):
# должен вызываться только обработчиком исключений
if isinstance(exc, OSError):
if self._loop.get_debug():
logger.debug("%r: %s", self, message, exc_info=True)
else:
self._loop.call_exception_handler({
'message': message,
'exception': exc,
'transport': self,
'protocol': self._protocol,
})
self._close(exc)
def _close(self, exc=None):
self._closing = True
if self._buffer:
self._loop._remove_writer(self._fileno)
self._buffer.clear()
self._loop._remove_reader(self._fileno)
self._loop.call_soon(self._call_connection_lost, exc)
def _call_connection_lost(self, exc):
try:
self._protocol.connection_lost(exc)
finally:
self._pipe.close()
self._pipe = None
self._protocol = None
self._loop = None
class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
stdin_w = None
if (stdin == subprocess.PIPE
and (sys.platform.startswith('aix') or sys.platform == 'cygwin')):
# Используйте сокетную пару для stdin на AIX, так как он не
# поддерживает выбор событий чтения на конце записи
# сокета (который мы используем для обнаружения закрытия
# другого конца).
stdin, stdin_w = socket.socketpair()
try:
self._proc = subprocess.Popen(
args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
universal_newlines=False, bufsize=bufsize, **kwargs)
if stdin_w is not None:
stdin.close()
self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
stdin_w = None
finally:
if stdin_w is not None:
stdin.close()
stdin_w.close()
class _PidfdChildWatcher:
"""Реализация наблюдателя дочерних процессов с использованием файловых
дескрипторов pid в Linux.
Этот наблюдатель дочерних процессов опрашивает файловые дескрипторы
процессов (pidfds) для ожидания завершения дочерних процессов. В некоторых
аспектах PidfdChildWatcher является "золотой серединой" среди реализаций
наблюдателей дочерних процессов. Он не требует сигналов или потоков, не
вмешивается в процессы, запущенные вне цикла событий, и масштабируется
линейно в зависимости от количества подпроцессов, запущенных циклом
событий. Основной недостаток в том, что pidfds специфичны для Linux и
работают только на новых (5.3+) ядрах.
"""
def add_child_handler(self, pid, callback, *args):
loop = events.get_running_loop()
pidfd = os.pidfd_open(pid)
loop._add_reader(pidfd, self._do_wait, pid, pidfd, callback, args)
def _do_wait(self, pid, pidfd, callback, args):
loop = events.get_running_loop()
loop._remove_reader(pidfd)
try:
_, status = os.waitpid(pid, 0)
except ChildProcessError:
# Дочерний процесс уже обработан
# (может произойти, если waitpid() вызывается в другом месте).
returncode = 255
logger.warning(
"child process pid %d exit status already read: "
" will report returncode 255",
pid)
else:
returncode = waitstatus_to_exitcode(status)
finally:
os.close(pidfd)
callback(pid, returncode, *args)
class _ThreadedChildWatcher:
"""Реализация потокового наблюдателя за дочерними процессами.
Наблюдатель использует отдельный поток на процесс
для ожидания завершения процесса.
Он не требует подписки на сигнал POSIX,
но создание потоков не бесплатно.
Наблюдатель имеет сложность O(1), его производительность не зависит
от количества порождённых процессов.
"""
def __init__(self):
self._pid_counter = itertools.count(0)
self._threads = {}
def __del__(self, _warn=warnings.warn):
threads = [thread for thread in list(self._threads.values())
if thread.is_alive()]
if threads:
_warn(f"{self.__class__} has registered but not finished child processes",
ResourceWarning,
source=self)
def add_child_handler(self, pid, callback, *args):
loop = events.get_running_loop()
thread = threading.Thread(target=self._do_waitpid,
name=f"asyncio-waitpid-{next(self._pid_counter)}",
args=(loop, pid, callback, args),
daemon=True)
self._threads[pid] = thread
thread.start()
def _do_waitpid(self, loop, expected_pid, callback, args):
assert expected_pid > 0
try:
pid, status = os.waitpid(expected_pid, 0)
except ChildProcessError:
# Дочерний процесс уже обработан
# (может произойти, если waitpid() вызывается в другом месте).
pid = expected_pid
returncode = 255
logger.warning(
"Unknown child process pid %d, will report returncode 255",
pid)
else:
returncode = waitstatus_to_exitcode(status)
if loop.get_debug():
logger.debug('process %s exited with returncode %s',
expected_pid, returncode)
if loop.is_closed():
logger.warning("Loop %r that handles pid %r is closed", loop, pid)
else:
loop.call_soon_threadsafe(callback, pid, returncode, *args)
self._threads.pop(expected_pid)
def can_use_pidfd():
if not hasattr(os, 'pidfd_open'):
return False
try:
pid = os.getpid()
os.close(os.pidfd_open(pid, 0))
except OSError:
# заблокировано политикой безопасности, например SECCOMP
return False
return True
SelectorEventLoop = _UnixSelectorEventLoop
EventLoop = SelectorEventLoop