unix_events.py
python
"""Цикл событий Selector для Unix с обработкой сигналов."""
import errno
import os
import signal
import socket
import stat
import subprocess
import sys
import threading
import warnings
from . import base_events
from . import base_subprocess
from . import compat
from . import constants
from . import coroutines
from . import events
from . import futures
from . import selector_events
from . import selectors
from . import transports
from .coroutines import coroutine
from .log import logger
__all__ = ['SelectorEventLoop',
'AbstractChildWatcher', 'SafeChildWatcher',
'FastChildWatcher', 'DefaultEventLoopPolicy',
]
if sys.platform == 'win32': # pragma: no cover
raise ImportError('Signals are not really supported on Windows')
def _sighandler_noop(signum, frame):
"""Заглушка обработчика сигнала."""
pass
try:
_fspath = os.fspath
except AttributeError:
# Python 3.5 или раньше
_fspath = lambda path: path
class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
"""Цикл событий Unix.
Добавляет обработку сигналов и поддержку доменных сокетов UNIX в SelectorEventLoop.
"""
def __init__(self, selector=None):
super().__init__(selector)
self._signal_handlers = {}
def _socketpair(self):
return socket.socketpair()
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 coroutines.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)
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('sig {} cannot be caught'.format(sig))
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('sig {} cannot be caught'.format(sig))
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('sig must be an int, not {!r}'.format(sig))
if not (1 <= sig < signal.NSIG):
raise ValueError(
'sig {} out of range(1, {})'.format(sig, signal.NSIG))
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)
@coroutine
def _make_subprocess_transport(self, protocol, args, shell,
stdin, stdout, stderr, bufsize,
extra=None, **kwargs):
with events.get_child_watcher() as 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:
yield from waiter
except Exception as exc:
# Обход ошибки CPython #23353: использование yield/yield-from в
# блоке except генератора не очищается должным образом
# sys.exc_info()
err = exc
else:
err = None
if err is not None:
transp.close()
yield from transp._wait()
raise err
return transp
def _child_watcher_callback(self, pid, returncode, transp):
self.call_soon_threadsafe(transp._process_exited, returncode)
@coroutine
def create_unix_connection(self, protocol_factory, path, *,
ssl=None, sock=None,
server_hostname=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 path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
try:
sock.setblocking(False)
yield from 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
not base_events._is_stream_socket(sock.type)):
raise ValueError(
'A UNIX Domain Stream Socket was expected, got {!r}'
.format(sock))
sock.setblocking(False)
transport, protocol = yield from self._create_connection_transport(
sock, protocol_factory, ssl, server_hostname)
return transport, protocol
@coroutine
def create_unix_server(self, protocol_factory, path=None, *,
sock=None, backlog=100, ssl=None):
if isinstance(ssl, bool):
raise TypeError('ssl argument must be an SSLContext or None')
if path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
path = _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 = 'Address {!r} is already in use'.format(path)
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
not base_events._is_stream_socket(sock.type)):
raise ValueError(
'A UNIX Domain Stream Socket was expected, got {!r}'
.format(sock))
server = base_events.Server(self, [sock])
sock.listen(backlog)
sock.setblocking(False)
self._start_serving(protocol_factory, sock, ssl, server)
return server
if hasattr(os, 'set_blocking'):
def _set_nonblocking(fd):
os.set_blocking(fd, False)
else:
import fcntl
def _set_nonblocking(fd):
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
flags = flags | os.O_NONBLOCK
fcntl.fcntl(fd, fcntl.F_SETFL, flags)
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
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.")
_set_nonblocking(self._fileno)
self._loop.call_soon(self._protocol.connection_made, self)
# начинать чтение только после вызова 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('fd=%s' % 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 '<%s>' % ' '.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):
self._loop._remove_reader(self._fileno)
def resume_reading(self):
self._loop._add_reader(self._fileno, self._read_ready)
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)
# В Python 3.3 и старше объекты с деструктором, являющиеся частью циклической ссылки,
# никогда не уничтожаются. В Python 3.4 это уже не так благодаря
# PEP 442.
if compat.PY34:
def __del__(self):
if self._pipe is not None:
warnings.warn("unclosed transport %r" % self, 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")
_set_nonblocking(self._fileno)
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('fd=%s' % 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('bufsize=%s' % bufsize)
elif self._pipe is not None:
info.append('open')
else:
info.append('closed')
return '<%s>' % ' '.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 Exception 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 Exception 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()
# В Python 3.3 и старше объекты с деструктором, являющиеся частью циклической ссылки,
# никогда не уничтожаются. В Python 3.4 это уже не так благодаря
# PEP 442.
if compat.PY34:
def __del__(self):
if self._pipe is not None:
warnings.warn("unclosed transport %r" % self, 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, base_events._FATAL_ERROR_IGNORE):
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
if hasattr(os, 'set_inheritable'):
# Python 3.4 и новее
_set_inheritable = os.set_inheritable
else:
import fcntl
def _set_inheritable(fd, inheritable):
cloexec_flag = getattr(fcntl, 'FD_CLOEXEC', 1)
old = fcntl.fcntl(fd, fcntl.F_GETFD)
if not inheritable:
fcntl.fcntl(fd, fcntl.F_SETFD, old | cloexec_flag)
else:
fcntl.fcntl(fd, fcntl.F_SETFD, old & ~cloexec_flag)
class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
stdin_w = None
if stdin == subprocess.PIPE:
# Используйте пару сокетов для stdin, так как не все платформы
# поддерживает выбор событий чтения на конце записи
# сокета (который мы используем для обнаружения закрытия
# другой конец). Особенно это необходимо на AIX и работает
# отлично на других платформах.
stdin, stdin_w = self._loop._socketpair()
# Пометьте записывающий конец канала stdin как ненаследуемый,
# требуется для close_fds=False в Python 3.3 и старше
# (Python 3.4 реализует PEP 446, socketpair возвращает
# ненаследуемые сокеты)
_set_inheritable(stdin_w.fileno(), False)
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)
class AbstractChildWatcher:
"""Абстрактный базовый класс для мониторинга дочерних процессов.
Объекты, производные от этого класса, следят за набором подпроцессов и
сообщают об их завершении или прерывании сигналом.
Новые колбэки регистрируются с помощью .add_child_handler(). Запуск нового
процесса должен выполняться внутри блока 'with', чтобы наблюдатель мог
приостановить свою активность до полной регистрации нового процесса (это
необходимо для предотвращения состояния гонки в некоторых реализациях).
Пример:
with watcher:
proc = subprocess.Popen("sleep 1")
watcher.add_child_handler(proc.pid, callback)
Примечания:
Реализации этого класса должны быть потокобезопасными.
Поскольку объекты наблюдателя дочерних процессов могут перехватывать
сигнал SIGCHLD и вызывать waitpid(-1), на процесс должен приходиться
только один активный объект.
"""
def add_child_handler(self, pid, callback, *args):
"""Зарегистрировать новый обработчик дочернего процесса.
Организовать вызов callback(pid, returncode, *args) при завершении
процесса с идентификатором 'pid'. Указание другого колбэка для того же
процесса заменяет предыдущий обработчик.
Примечание: callback() должен быть потокобезопасным.
"""
raise NotImplementedError()
def remove_child_handler(self, pid):
"""Удаляет обработчик для процесса 'pid'.
Функция возвращает True, если обработчик был успешно удалён, и False,
если удалять было нечего."""
raise NotImplementedError()
def attach_loop(self, loop):
"""Прикрепить наблюдатель к циклу событий.
Если наблюдатель ранее был прикреплён к циклу событий, то сначала он
отсоединяется, а затем прикрепляется к новому циклу.
Примечание: loop может быть None.
"""
raise NotImplementedError()
def close(self):
"""Закрыть наблюдатель.
Это необходимо вызывать, чтобы гарантировать освобождение всех
связанных ресурсов.
"""
raise NotImplementedError()
def __enter__(self):
"""Войти в контекст наблюдателя и разрешить запуск новых процессов
Эта функция должна возвращать self"""
raise NotImplementedError()
def __exit__(self, a, b, c):
"""Выйти из контекста наблюдателя"""
raise NotImplementedError()
class BaseChildWatcher(AbstractChildWatcher):
def __init__(self):
self._loop = None
self._callbacks = {}
def close(self):
self.attach_loop(None)
def _do_waitpid(self, expected_pid):
raise NotImplementedError()
def _do_waitpid_all(self):
raise NotImplementedError()
def attach_loop(self, loop):
assert loop is None or isinstance(loop, events.AbstractEventLoop)
if self._loop is not None and loop is None and self._callbacks:
warnings.warn(
'A loop is being detached '
'from a child watcher with pending handlers',
RuntimeWarning)
if self._loop is not None:
self._loop.remove_signal_handler(signal.SIGCHLD)
self._loop = loop
if loop is not None:
loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)
# Предотвратить состояние гонки на случай, если дочерний процесс завершился
# во время переключения.
self._do_waitpid_all()
def _sig_chld(self):
try:
self._do_waitpid_all()
except Exception as exc:
# self._loop всегда должен быть доступен здесь
# поскольку '_sig_chld' добавляется как обработчик сигнала
# в 'attach_loop'
self._loop.call_exception_handler({
'message': 'Unknown exception in SIGCHLD handler',
'exception': exc,
})
def _compute_returncode(self, status):
if os.WIFSIGNALED(status):
# Дочерний процесс завершился из-за сигнала.
return -os.WTERMSIG(status)
elif os.WIFEXITED(status):
# Дочерний процесс завершился (например, sys.exit()).
return os.WEXITSTATUS(status)
else:
# Дочерний процесс завершился, но мы не понимаем его статус.
# Этого не должно происходить, но если это произошло, давайте просто
# вернём этот статус; возможно, это поможет в отладке.
return status
class SafeChildWatcher(BaseChildWatcher):
"""«Безопасная» реализация наблюдателя за дочерними процессами.
Эта реализация избегает нарушения работы другого кода, порождающего процессы, путём
явного опроса каждого процесса в обработчике SIGCHLD вместо вызова
os.waitpid(-1).
Это безопасное решение, но оно создаёт значительные накладные расходы при работе с
большим количеством дочерних процессов (O(n) при каждом возникновении SIGCHLD).
"""
def close(self):
self._callbacks.clear()
super().close()
def __enter__(self):
return self
def __exit__(self, a, b, c):
pass
def add_child_handler(self, pid, callback, *args):
if self._loop is None:
raise RuntimeError(
"Cannot add child handler, "
"the child watcher does not have a loop attached")
self._callbacks[pid] = (callback, args)
# Предотвращает состояние гонки на случай, если дочерний процесс уже завершился.
self._do_waitpid(pid)
def remove_child_handler(self, pid):
try:
del self._callbacks[pid]
return True
except KeyError:
return False
def _do_waitpid_all(self):
for pid in list(self._callbacks):
self._do_waitpid(pid)
def _do_waitpid(self, expected_pid):
assert expected_pid > 0
try:
pid, status = os.waitpid(expected_pid, os.WNOHANG)
except ChildProcessError:
# Дочерний процесс уже обработан
# (может произойти, если waitpid() вызывается в другом месте).
pid = expected_pid
returncode = 255
logger.warning(
"Unknown child process pid %d, will report returncode 255",
pid)
else:
if pid == 0:
# Дочерний процесс ещё жив.
return
returncode = self._compute_returncode(status)
if self._loop.get_debug():
logger.debug('process %s exited with returncode %s',
expected_pid, returncode)
try:
callback, args = self._callbacks.pop(pid)
except KeyError: # pragma: no cover
# Может произойти, если .remove_child_handler() вызывается
# после того, как os.waitpid() вернул результат.
if self._loop.get_debug():
logger.warning("Child watcher got an unexpected pid: %r",
pid, exc_info=True)
else:
callback(pid, returncode, *args)
class FastChildWatcher(BaseChildWatcher):
"""«Быстрая» реализация наблюдателя за дочерними процессами.
Эта реализация перехватывает все завершённые процессы путём вызова
os.waitpid(-1) напрямую, что может нарушить работу другого кода, порождающего процессы
и ожидающего их завершения.
Нет заметных накладных расходов при обработке большого количества дочерних процессов
(O(1) при каждом завершении дочернего процесса).
"""
def __init__(self):
super().__init__()
self._lock = threading.Lock()
self._zombies = {}
self._forks = 0
def close(self):
self._callbacks.clear()
self._zombies.clear()
super().close()
def __enter__(self):
with self._lock:
self._forks += 1
return self
def __exit__(self, a, b, c):
with self._lock:
self._forks -= 1
if self._forks or not self._zombies:
return
collateral_victims = str(self._zombies)
self._zombies.clear()
logger.warning(
"Caught subprocesses termination from unknown pids: %s",
collateral_victims)
def add_child_handler(self, pid, callback, *args):
assert self._forks, "Must use the context manager"
if self._loop is None:
raise RuntimeError(
"Cannot add child handler, "
"the child watcher does not have a loop attached")
with self._lock:
try:
returncode = self._zombies.pop(pid)
except KeyError:
# Дочерний процесс выполняется.
self._callbacks[pid] = callback, args
return
# Дочерний процесс уже мёртв. Можно вызвать колбэк.
callback(pid, returncode, *args)
def remove_child_handler(self, pid):
try:
del self._callbacks[pid]
return True
except KeyError:
return False
def _do_waitpid_all(self):
# Из-за объединения сигналов мы должны продолжать вызывать waitpid(),
# пока есть возможность перехватить дочерний процесс.
while True:
try:
pid, status = os.waitpid(-1, os.WNOHANG)
except ChildProcessError:
# Больше не существует дочерних процессов.
return
else:
if pid == 0:
# Дочерний процесс ещё жив.
return
returncode = self._compute_returncode(status)
with self._lock:
try:
callback, args = self._callbacks.pop(pid)
except KeyError:
# неизвестный дочерний процесс
if self._forks:
# Возможно, он ещё не зарегистрирован.
self._zombies[pid] = returncode
if self._loop.get_debug():
logger.debug('unknown process %s exited '
'with returncode %s',
pid, returncode)
continue
callback = None
else:
if self._loop.get_debug():
logger.debug('process %s exited with returncode %s',
pid, returncode)
if callback is None:
logger.warning(
"Caught subprocess termination from unknown pid: "
"%d -> %d", pid, returncode)
else:
callback(pid, returncode, *args)
class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
"""Политика цикла событий UNIX с наблюдателем за дочерними процессами."""
_loop_factory = _UnixSelectorEventLoop
def __init__(self):
super().__init__()
self._watcher = None
def _init_watcher(self):
with events._lock:
if self._watcher is None: # pragma: no branch
self._watcher = SafeChildWatcher()
if isinstance(threading.current_thread(),
threading._MainThread):
self._watcher.attach_loop(self._local._loop)
def set_event_loop(self, loop):
"""Установить цикл событий.
Побочный эффект: если до этого был установлен наблюдатель за дочерними процессами, то вызов
.set_event_loop() из главного потока вызовет .attach_loop(loop) у
наблюдателя.
"""
super().set_event_loop(loop)
if self._watcher is not None and \
isinstance(threading.current_thread(), threading._MainThread):
self._watcher.attach_loop(loop)
def get_child_watcher(self):
"""Получить наблюдатель для дочерних процессов.
Если ещё не установлен, автоматически создаётся объект SafeChildWatcher.
"""
if self._watcher is None:
self._init_watcher()
return self._watcher
def set_child_watcher(self, watcher):
"""Устанавливает наблюдатель за дочерними процессами."""
assert watcher is None or isinstance(watcher, AbstractChildWatcher)
if self._watcher is not None:
self._watcher.close()
self._watcher = watcher
SelectorEventLoop = _UnixSelectorEventLoop
DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy