tf.distribute.experimental.coordinator.ClusterCoordinator

Объект для планирования и координации выполнения удалённых функций.

tf.distribute.experimental.coordinator.ClusterCoordinator(
    strategy
)

Используется в ноутбуках

Этот класс используется для создания отказоустойчивых ресурсов и отправки функций на удалённые серверы TensorFlow.

В настоящее время этот класс не поддерживает автономное использование. Его следует использовать вместе со стратегией tf.distribute, предназначенной для работы с ним. В настоящее время класс ClusterCoordinator работает только с tf.distribute.experimental.ParameterServerStrategy.

API schedule/join

Наиболее важными API этого класса являются schedule и join. API schedule является неблокирующим: он помещает tf.function в очередь и сразу возвращает RemoteValue. Функции из очереди отправляются удалённым рабочим в фоновых потоках, а их RemoteValue заполняются асинхронно. Поскольку для schedule не требуется назначение рабочего, переданная tf.function может выполняться на любом доступном рабочем. Если рабочий, на котором она выполняется, становится недоступен до завершения функции, она будет перенесена на другой рабочий. По этой причине, а также потому, что выполнение функции не является атомарным, одна и та же функция может выполниться несколько раз.

Обработка сбоев задач

При использовании этого класса вместе с tf.distribute.experimental.ParameterServerStrategy обеспечивается встроенная отказоустойчивость при сбоях рабочих. Это означает, что если по какой-либо причине координатор не может связаться с некоторыми рабочими, обучение продолжится на оставшихся рабочих. После восстановления отказавший рабочий будет снова задействован для выполнения функций, когда на нём будут повторно созданы наборы данных, созданные с помощью create_per_worker_dataset.

При сбое сервера параметров вызов schedule, join или done приводит к возникновению исключения tf.errors.UnavailableError. В этом случае, помимо восстановления отказавшего сервера параметров, пользователям следует перезапустить координатор, чтобы он повторно подключился к рабочим и серверам параметров, заново создал переменные и загрузил контрольные точки. Если координатор даст сбой, после его восстановления программа автоматически подключится к рабочим и серверам параметров и продолжит работу с контрольной точки.

Поэтому в программе пользователя необходимо периодически сохранять файл контрольной точки и восстанавливать состояние из него при запуске программы. Если контрольная точка создана для tf.keras.optimizers.Optimizer, то после восстановления из контрольной точки его свойство iterations приблизительно указывает количество выполненных шагов. Это можно использовать для определения количества эпох и шагов, необходимых до завершения обучения.

Пример использования этого API см. в документации tf.distribute.experimental.ParameterServerStrategy.

В настоящее время этот API находится в разработке, поэтому он и его реализация могут измениться.

Методы

create_per_worker_dataset

Посмотреть исходный код

create_per_worker_dataset(
    dataset_fn
)

Создаёт набор данных на каждом рабочем.

Создаёт наборы данных на рабочих на основе входных данных, которыми может быть tf.data.Dataset, tf.distribute.DistributedDataset или функция, возвращающая набор данных, и возвращает объект, представляющий коллекцию этих отдельных наборов данных. Вызов iter для такой коллекции наборов данных возвращает tf.distribute.experimental.coordinator.PerWorkerValues — коллекцию итераторов, размещённых на соответствующих рабочих.

Вызов next для PerWorkerValues итератора не поддерживается. Итератор предназначен для передачи в качестве аргумента в tf.distribute.experimental.coordinator.ClusterCoordinator.schedule. Когда запланированная функция будет выполняться рабочим, она получит отдельный итератор, соответствующий этому рабочему. Метод next можно вызвать для итератора внутри запланированной функции, если итератор является входным аргументом функции.

В настоящее время метод schedule предполагает, что все рабочие одинаковы, а значит, наборы данных на разных рабочих тоже одинаковы, за исключением случаев, когда они могут перемешиваться по-разному, если содержат операцию dataset.shuffle и для неё не задано начальное значение генератора случайных чисел. Поэтому мы также рекомендуем выполнять наборы данных бесконечно и планировать конечное число шагов, а не полагаться на OutOfRangeError набора данных.

Пример:

strategy = tf.distribute.experimental.ParameterServerStrategy(
    cluster_resolver=...)
coordinator = tf.distribute.experimental.coordinator.ClusterCoordinator(
    strategy=strategy)

@tf.function
def worker_fn(iterator):
  return next(iterator)

def per_worker_dataset_fn():
  return strategy.distribute_datasets_from_function(
      lambda x: tf.data.Dataset.from_tensor_slices([3] * 3))

per_worker_dataset = coordinator.create_per_worker_dataset(
    per_worker_dataset_fn)
per_worker_iter = iter(per_worker_dataset)
remote_value = coordinator.schedule(worker_fn, args=(per_worker_iter,))
assert remote_value.fetch() == 3

done

Посмотреть исходный код

done()

Возвращает значение, указывающее, завершили ли выполнение все запланированные функции.

Если ранее запланированная функция вызывает ошибку, done завершится с ошибкой, вызвав одно из таких исключений.

Если done возвращает True или вызывает исключение, это гарантирует, что ни одна функция в данный момент не выполняется.

fetch

Посмотреть исходный код

fetch(
    val
)

Блокирующий вызов для получения результатов удалённых значений.

Это обёртка для tf.distribute.experimental.coordinator.RemoteValue.fetch для структуры RemoteValue; она возвращает результаты выполнения RemoteValue. Если они ещё не готовы, вызов ожидает их готовности, блокируя вызывающий поток.

Пример:

strategy = ...
coordinator = tf.distribute.experimental.coordinator.ClusterCoordinator(
    strategy)

def dataset_fn():
  return tf.data.Dataset.from_tensor_slices([1, 1, 1])

with strategy.scope():
  v = tf.Variable(initial_value=0)

@tf.function
def worker_fn(iterator):
  def replica_fn(x):
    v.assign_add(x)
    return v.read_value()
  return strategy.run(replica_fn, args=(next(iterator),))

distributed_dataset = coordinator.create_per_worker_dataset(dataset_fn)
distributed_iterator = iter(distributed_dataset)
result = coordinator.schedule(worker_fn, args=(distributed_iterator,))
assert coordinator.fetch(result) == 1

join

Посмотреть исходный код

join()

Блокирует выполнение до завершения всех запланированных функций.

Если ранее запланированная функция вызывает ошибку, join завершается с ошибкой, вызывая одно из таких исключений, и очищает все собранные к этому моменту ошибки. В этом случае некоторые из ранее запланированных функций могли не выполниться. Пользователи могут вызвать fetch для возвращённого tf.distribute.experimental.coordinator.RemoteValue, чтобы проверить, выполнились ли они, завершились с ошибкой или были отменены. Если некоторые отменённые функции нужно запланировать повторно, пользователям следует снова вызвать schedule с этой функцией.

Если join возвращает управление или вызывает исключение, это гарантирует, что ни одна функция в данный момент не выполняется.

schedule

Посмотреть исходный код

schedule(
    fn, args=None, kwargs=None
)

Планирует отправку fn рабочему для асинхронного выполнения.

Этот метод является неблокирующим: он помещает fn в очередь для последующего выполнения и сразу возвращает объект tf.distribute.experimental.coordinator.RemoteValue. Для ожидания завершения выполнения функции и получения её результата с удалённого рабочего можно вызвать для этого объекта fetch. Чтобы дождаться завершения всех запланированных функций, вызовите tf.distribute.experimental.coordinator.ClusterCoordinator.join.

Метод schedule гарантирует, что fn будет выполнена на рабочем как минимум один раз; если соответствующий рабочий даст сбой в процессе выполнения, функция может выполниться несколько раз. Поскольку рабочий может дать сбой в любой момент выполнения функции, она может выполниться частично. Однако tf.distribute.experimental.coordinator.ClusterCoordinator гарантирует, что в таких случаях функция в конечном итоге будет выполнена на любом доступном рабочем.

Если ранее запланированная функция вызывает ошибку, schedule вызовет одно из таких исключений и очистит все собранные к этому моменту ошибки. В этом случае некоторые из ранее запланированных функций могли не выполниться. Пользователь может вызвать fetch для возвращённого tf.distribute.experimental.coordinator.RemoteValue, чтобы проверить, выполнились ли они, завершились с ошибкой или были отменены, и при необходимости повторно запланировать соответствующую функцию.

Если schedule вызывает исключение, это гарантирует, что ни одна функция в данный момент не выполняется.

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

args и kwargs — это аргументы, передаваемые в fn при выполнении fn на рабочем. Они могут быть объектами tf.distribute.experimental.coordinator.PerWorkerValues; в этом случае аргумент будет заменён соответствующим компонентом на целевом рабочем. Аргументы, не являющиеся объектами tf.distribute.experimental.coordinator.PerWorkerValues, передаются в fn без изменений. В настоящее время tf.distribute.experimental.coordinator.RemoteValue не поддерживается в качестве входного аргумента args или kwargs.