Согласование участников

Динамическое rendezvous

torch.distributed.elastic.rendezvous.dynamic_rendezvous.create_handler(store, backend, params) [исходный код]

Создать новый DynamicRendezvousHandler с указанными параметрами.

Параметры:

• store (Store) – Хранилище C10d, которое возвращается в составе rendezvous.
• backend (RendezvousBackend) – Бэкенд для хранения состояния rendezvous.

Тип возвращаемого значения:

DynamicRendezvousHandler

Параметр	Описание
join_timeout	Общее время в секундах, за которое ожидается завершение rendezvous. По умолчанию — 600 секунд.
last_call_timeout	Дополнительное время ожидания в секундах перед завершением rendezvous после достижения минимального числа узлов. По умолчанию — 30 секунд.
close_timeout	Время в секундах, за которое ожидается закрытие rendezvous после вызова RendezvousHandler.set_closed() или RendezvousHandler.shutdown(). По умолчанию — 30 секунд.
heartbeat	Время в секундах, за которое ожидается завершение сигнала keep-alive

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.DynamicRendezvousHandler [исходный код]

Представляет обработчик, который устанавливает rendezvous между набором узлов.

classmethod from_backend(run_id, store, backend, min_nodes, max_nodes, local_addr=None, timeout=None, keep_alive_interval=5, keep_alive_max_attempt=3) [исходный код]

Создать новый DynamicRendezvousHandler.

Параметры:

• run_id (str) – Идентификатор запуска rendezvous.
• store (Store) – Хранилище C10d, которое возвращается в составе rendezvous.
• backend (RendezvousBackend) – Бэкенд для хранения состояния rendezvous.
• min_nodes (int) – Минимальное число узлов, допускаемых к rendezvous.
• max_nodes (int) – Максимальное число узлов, допускаемых к rendezvous.
• local_addr (str | None) – Адрес локального узла.
• timeout (RendezvousTimeout | None) – Конфигурация тайм-аутов rendezvous.
• keep_alive_interval (int) – Время ожидания узла перед отправкой сигнала heartbeat, чтобы оставаться активным в rendezvous.
• keep_alive_max_attempt (int) – Максимальное число неудачных попыток отправки сигнала heartbeat, после которого узел считается неактивным.

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousBackend [исходный код]

Представляет бэкенд, который хранит состояние rendezvous.

abstract get_state() [исходный код]

Получить состояние rendezvous.

Возвращает:

Кортеж из закодированного состояния rendezvous и его токена ограждения либо None, если состояние не найдено в бэкенде.

Вызывает исключения:

• RendezvousConnectionError – Не удалось установить соединение с бэкендом.
• RendezvousStateError – Состояние rendezvous повреждено.

Тип возвращаемого значения:

tuple[bytes, Any] | None

abstract property name: str

Получить имя бэкенда.

abstract set_state(state, token=None) [исходный код]

Задать состояние rendezvous.

Новое состояние rendezvous задаётся условно:

• Если указанный token совпадает с токеном ограждения, сохранённым в бэкенде, состояние будет обновлено. Новое состояние вместе с его токеном ограждения будет возвращено вызывающей стороне.
• Если указанный token не совпадает с токеном ограждения, сохранённым в бэкенде, состояние не будет обновлено; вместо этого вызывающей стороне будет возвращено существующее состояние вместе с его токеном ограждения.
• Если указанный token равен None, новое состояние будет задано только в том случае, если в бэкенде нет существующего состояния. Вызывающей стороне будет возвращено либо новое состояние, либо существующее состояние вместе с его токеном ограждения.

Параметры:

• state (bytes) – Закодированное состояние rendezvous.
• token (Any | None) – Необязательный токен ограждения, полученный при предыдущем вызове get_state() или set_state().

Возвращает:

Кортеж из сериализованного состояния rendezvous, его токена ограждения и логического значения, указывающего, была ли успешной попытка задать состояние.

Вызывает исключения:

• RendezvousConnectionError – Не удалось установить соединение с бэкендом.
• RendezvousStateError – Состояние rendezvous повреждено.

Тип возвращаемого значения:

tuple[bytes, Any, bool] | None

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousTimeout(join=None, last_call=None, close=None, heartbeat=None) [исходный код]

Содержит конфигурацию тайм-аутов rendezvous.

Параметры:

• join (timedelta | None) – Время, за которое ожидается завершение rendezvous.
• last_call (timedelta | None) – Дополнительное время ожидания перед завершением rendezvous после достижения минимального числа участников.
• close (timedelta | None) – Время, за которое ожидается закрытие rendezvous после вызова RendezvousHandler.set_closed() или RendezvousHandler.shutdown().
• heartbeat (timedelta | None) – Время, за которое ожидается завершение сигнала heartbeat keep-alive.

property close: timedelta

Получить тайм-аут закрытия.

property heartbeat: timedelta

Получить тайм-аут сигнала heartbeat keep-alive.

property join: timedelta

Получить тайм-аут подключения.

property last_call: timedelta

Получить тайм-аут последнего вызова.

Rendezvous Etcd (устаревший)

class torch.distributed.elastic.rendezvous.etcd_rendezvous.EtcdRendezvousHandler(rdzv_impl, local_addr) [исходный код]

Реализует интерфейс torch.distributed.elastic.rendezvous.RendezvousHandler на основе torch.distributed.elastic.rendezvous.etcd_rendezvous.EtcdRendezvous. EtcdRendezvousHandler использует URL для настройки типа rendezvous и передачи специфичных для реализации параметров модулю rendezvous. Базовый URL конфигурации rendezvous etcd выглядит следующим образом

etcd://<etcd_address>:<port>/<job_id>?min_workers=<min_workers>&max_workers=<max_workers>  # noqa: W605

-- example --

etcd://localhost:2379/1234?min_workers=1&max_workers=3

Приведённый выше URL интерпретируется следующим образом:

1. Используется обработчик rendezvous, зарегистрированный для схемы etcd
2. Используемая конечная точка etcd — localhost:2379
3. job_id == 1234 используется в качестве префикса в etcd (это позволяет нескольким заданиям совместно использовать один сервер etcd при условии, что job_ids гарантированно уникальны). Обратите внимание, что идентификатор задания может быть любой строкой (например, не обязательно числом), если он уникален.
4. min_workers=1 и max_workers=3 задают диапазон размера группы — Torch Distributed Elastic начинает выполнение задания, если размер кластера не меньше min_workers, и допускает в кластер до max_workers участников.

Ниже приведён полный список параметров, которые можно передать rendezvous etcd:

Параметр	Описание
min_workers	минимальное число рабочих процессов, необходимое для действительного rendezvous
max_workers	максимальное число допускаемых рабочих процессов
timeout	общий тайм-аут, в пределах которого ожидается успешное выполнение next_rendezvous (по умолчанию 600 с)
last_call_timeout	дополнительное время ожидания («последний вызов») после достижения минимального числа рабочих процессов (по умолчанию 30 с)
etcd_prefix	префикс пути (от корня etcd), внутри которого будут созданы все узлы etcd (по умолчанию /torchelastic/p2p)

Хранилище Etcd

EtcdStore — это тип экземпляра C10d Store, возвращаемый next_rendezvous() при использовании etcd в качестве бэкенда для rendezvous.

class torch.distributed.elastic.rendezvous.etcd_store.EtcdStore(etcd_client, etcd_store_prefix, timeout=None) [исходный код]

Реализует интерфейс хранилища c10, используя экземпляр etcd для rendezvous.

Это объект хранилища, возвращаемый EtcdRendezvous.

add(key, num) [исходный код]

Атомарно увеличивает значение на целое число.

Целое число представлено в виде строки в десятичной системе счисления. Если ключ отсутствует, будет использоваться значение по умолчанию 0.

Возвращает:

новое (увеличенное) значение

Тип возвращаемого значения:

int

check(keys) [исходный код]

Проверяет, присутствуют ли сразу все ключи (без ожидания).

Тип возвращаемого значения:

bool

get(key) [исходный код]

Получает значение по ключу, при необходимости блокируя выполнение в ожидании.

Если ключ отсутствует, будет ожидать его публикации не более timeout или до момента публикации ключа.

Возвращает:

значение (bytes)

Вызывает исключение:

LookupError — если ключ так и не опубликован по истечении времени ожидания –

Тип возвращаемого значения:

bytes

set(key, value) [исходный код]

Записывает пару ключ/значение в EtcdStore.

И ключ, и значение могут иметь тип Python str или bytes.

wait(keys, override_timeout=None) [исходный код]

Ожидает публикации всех ключей или истечения времени ожидания.

Вызывает исключение:

LookupError — если время ожидания истекло –

Сервер Etcd

EtcdServer — это вспомогательный класс, упрощающий запуск и остановку сервера etcd в отдельном процессе. Он полезен для тестирования или развертываний на одном узле (с несколькими рабочими процессами), когда неудобно вручную настраивать отдельный сервер etcd.

class torch.distributed.elastic.rendezvous.etcd_server.EtcdServer(data_dir=None) [исходный код]

Примечание

Проверено на сервере etcd v3.4.3.

Запускает и останавливает локальный автономный сервер etcd на случайном свободном порту. Полезен для запусков на одном узле с несколькими рабочими процессами или для тестирования, когда удобнее использовать вспомогательный сервер etcd, чем настраивать отдельный сервер.

Этот класс регистрирует обработчик завершения, который останавливает дочерний процесс etcd при выходе. Этот обработчик завершения НЕ заменяет вызов метода stop().

Для поиска исполняемого файла etcd используется следующий механизм резервного поиска:

1. Используется переменная окружения TORCHELASTIC_ETCD_BINARY_PATH
2. Используется <this file root>/bin/etcd, если он существует
3. Используется etcd из PATH

Использование

server = EtcdServer("/usr/bin/etcd", 2379, "/tmp/default.etcd")
server.start()
client = server.get_client()
# use client
server.stop()

Параметры:

etcd_binary_path – путь к исполняемому файлу сервера etcd (см. выше резервный путь)