tf.compat.v1.distribute.StrategyExtended

Дополнительные API для алгоритмов, которые должны учитывать распределение.

Наследуется от: StrategyExtended

tf.compat.v1.distribute.StrategyExtended(
    container_strategy
)

Примечание: Для большинства случаев использования tf.distribute.Strategy, вызывать эти методы не требуется, так как библиотеки TensorFlow (например, оптимизаторы) уже вызывают эти методы при необходимости от вашего имени.

Некоторые распространенные примеры использования функций на этой странице:

• Локализация

tf.distribute.DistributedValues может иметь ту же локализацию, что и распределённая переменная, что приводит к зеркальному значению, находящемуся на тех же устройствах, что и переменная (в отличие от вычислительных устройств). Такие значения могут быть переданы в вызов tf.distribute.StrategyExtended.update для обновления значения переменной. Вы можете использовать tf.distribute.StrategyExtended.colocate_vars_with для присвоения переменной той же локализации, что и другой переменной. Вы можете преобразовать значение «PerReplica» в локализацию переменной, используя tf.distribute.StrategyExtended.reduce_to или tf.distribute.StrategyExtended.batch_reduce_to.

• Как обновить распределённую переменную

Распределённая переменная — это переменные, созданные на нескольких устройствах. Как обсуждалось в словаре терминов, зеркальная переменная и переменная SyncOnRead являются двумя примерами. Стандартный шаблон для обновления распределённых переменных следующий:

1. В своей функции, передаваемой в tf.distribute.Strategy.run, вычислите список пар (обновление, переменная). Например, обновление может быть градиентом потери относительно переменной.
2. Переключитесь на межрепличный режим, вызвав tf.distribute.get_replica_context().merge_call() с обновлениями и переменными в качестве аргументов.
3. Вызовите tf.distribute.StrategyExtended.reduce_to(VariableAggregation.SUM, t, v) (для одной переменной) или tf.distribute.StrategyExtended.batch_reduce_to (для списка переменных), чтобы суммировать обновления.
4. Вызовите tf.distribute.StrategyExtended.update(v) для каждой переменной, чтобы обновить её значение.

Шаги 2—4 автоматически выполняются классом tf.keras.optimizers.Optimizer, если вы вызываете его метод tf.keras.optimizers.Optimizer.apply_gradients в контексте реплики.

Фактически, более высокоуровневое решение для обновления распределённой переменной заключается в вызове assign для переменной, как вы делаете это для обычной tf.Variable. Вы можете вызвать метод как в контексте реплики, так и в межрепличном контексте. Для зеркальной переменной вызов assign в контексте реплики требует указания типа aggregation в конструкторе переменной. В этом случае переключение контекстов и синхронизация, описанные в шагах 2—4, обрабатываются за вас. Если вы вызываете assign для зеркальной переменной в межрепличном контексте, вы можете назначить только одно значение или значения из другой зеркальной переменной или зеркального tf.distribute.DistributedValues. Для переменной SyncOnRead в контексте реплики вы можете просто вызвать assign, и агрегация не выполняется неявно. В межрепличном контексте вы можете назначить только одно значение переменной SyncOnRead. Пример — восстановление из контрольной точки: если тип aggregation переменной равен tf.VariableAggregation.SUM, предполагается, что значения реплик были добавлены до создания контрольной точки, поэтому при восстановлении значение делится на количество реплик, а затем назначается каждой реплике; если тип aggregation равен tf.VariableAggregation.MEAN, значение назначается каждой реплике непосредственно.

Методы

batch_reduce_to

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

batch_reduce_to(
    reduce_op, value_destination_pairs, options=None
)

Объединяет несколько вызовов reduce_to в один для более быстрого выполнения.

Аналогично reduce_to, но принимает список пар (значение, пункты назначения). Это более эффективно, чем уменьшение каждого значения по отдельности.

Этот API в настоящее время можно вызывать только в межрепличном контексте. Другие варианты для уменьшения значений между репликами:

• tf.distribute.StrategyExtended.reduce_to: не-поблочная версия этого API.
• tf.distribute.ReplicaContext.all_reduce: аналог этого API в контексте реплики. Поддерживает как поблочное, так и непоблочное all-reduce.
• tf.distribute.Strategy.reduce: более удобный метод для уменьшения до хоста в межрепличном контексте.

См. reduce_to для получения дополнительной информации.

@tf.function def step_fn(var):

def merge_fn(strategy, value, var): # Всередине выполняется all-reduce значения. Обратите внимание, что value здесь — # tf.distribute.DistributedValues. reduced = strategy.extended.batch_reduce_to( tf.distribute.ReduceOp.SUM, [(value, var)])[0] strategy.extended.update(var, lambda var, value: var.assign(value), args=(reduced,))

value = tf.identity(1.) tf.distribute.get_replica_context().merge_call(merge_fn, args=(value, var))

def run(strategy): with strategy.scope(): v = tf.Variable(0.) strategy.run(step_fn, args=(v,)) return v

run(tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])) MirroredVariable:{ 0: , 1: } run(tf.distribute.experimental.CentralStorageStrategy( compute_devices=["GPU:0", "GPU:1"], parameter_device="CPU:0")) run(tf.distribute.OneDeviceStrategy("GPU:0"))

broadcast_to

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

broadcast_to(
    tensor, destinations
)

Зеркалирование тензора на одном устройстве на все устройства рабочих узлов.

call_for_each_replica

Просмотр исходного кода

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

Выполнить fn один раз на каждой реплике.

fn может вызвать tf.get_replica_context() для доступа к методам, таким как replica_id_in_sync_group и merge_call().

merge_call() используется для связи между репликами и повторного входа в контекст между репликами. Все реплики приостанавливают свою работу после вызова merge_call(). После этого выполняется функция merge_fn. Результаты затем распаковываются и возвращаются каждой вызывающей реплике. После этого выполнение возобновляется до завершения fn или обнаружения другого вызова merge_call(). Пример:

# Called once in "cross-replica" context.
def merge_fn(distribution, three_plus_replica_id):
  # sum the values across replicas
  return sum(distribution.experimental_local_results(three_plus_replica_id))

# Called once per replica in `distribution`, in a "replica" context.
def fn(three):
  replica_ctx = tf.get_replica_context()
  v = three + replica_ctx.replica_id_in_sync_group
  # Computes the sum of the `v` values across all replicas.
  s = replica_ctx.merge_call(merge_fn, args=(v,))
  return s + v

with distribution.scope():
  # in "cross-replica" context
  ...
  merged_results = distribution.run(fn, args=[3])
  # merged_results has the values from every replica execution of `fn`.
  # This statement prints a list:
  print(distribution.experimental_local_results(merged_results))

colocate_vars_with

Просмотр исходного кода

colocate_vars_with(
    colocate_with_variable
)

Контекст, определяющий устройства, на которых будут создаваться переменные.

Внутри этого контекста не следует добавлять операции в граф, он используется только при создании переменных (некоторые реализации работают путём изменения создания переменных, другие используют контекст tf.compat.v1.colocate_with()).

Использовать можно только внутри self.scope().

Пример использования:

with strategy.scope():
  var1 = tf.Variable(...)
  with strategy.extended.colocate_vars_with(var1):
    # var2 and var3 will be created on the same device(s) as var1
    var2 = tf.Variable(...)
    var3 = tf.Variable(...)

  def fn(v1, v2, v3):
    # operates on v1 from var1, v2 from var2, and v3 from var3

  # `fn` runs on every device `var1` is on, `var2` and `var3` will be there
  # too.
  strategy.extended.update(var1, fn, args=(var2, var3))

experimental_make_numpy_dataset

Просмотр исходного кода

experimental_make_numpy_dataset(
    numpy_input, session=None
)

Создаёт набор данных на основе массива NumPy.

Это позволяет избежать добавления numpy_input как большой константы в граф и копирует данные на машину или машины, которые будут обрабатывать входные данные.

experimental_run_steps_on_iterator

Просмотр исходного кода

experimental_run_steps_on_iterator(
    fn, iterator, iterations=1, initial_loop_values=None
)

Выполнить fn с входными данными из iterator в течение iterations раз.

Этот метод может использоваться для многократного выполнения функции шага обучения с использованием входных данных из набора данных.

non_slot_devices

Просмотр исходного кода

non_slot_devices(
    var_list
)

Устройство(я) для переменных без слотов.

Этот метод возвращает устройства без слотов, где размещаются переменные без слотов. Пользователи могут создавать переменные без слотов на этих устройствах, используя блок:

with tf.distribute.StrategyExtended.colocate_vars_with(tf.distribute.StrategyExtended.non_slot_devices(...)):
  ...

read_var

Просмотр исходного кода

read_var(
    v
)

Считывает значение переменной.

Возвращает агрегированное значение локальной переменной реплики или (только для чтения) значение любой другой переменной.

reduce_to

Просмотр исходного кода

reduce_to(
    reduce_op, value, destinations, options=None
)

Объединить (например, суммировать или усреднять) значения по репликам.

reduce_to агрегирует tf.distribute.DistributedValues и распределённые переменные. Поддерживает как плотные значения, так и tf.IndexedSlices.

В настоящее время к этому API можно обращаться только в контексте между репликами. Другие варианты для уменьшения значений по репликам:

• tf.distribute.StrategyExtended.batch_reduce_to: пакетная версия этого API.
• tf.distribute.ReplicaContext.all_reduce: аналог этого API в контексте реплики. Поддерживает как пакетное, так и непоследовательное сведение.
• tf.distribute.Strategy.reduce: более удобный метод сведения к хосту в контексте между репликами.

destinations указывает, куда следует свести значение, например, "GPU:0". Также можно передать Tensor, и места назначения будут являться устройством этого тензора. Для сведения, передавайте то же самое value и destinations.

Может использоваться в tf.distribute.ReplicaContext.merge_call для написания кода, работающего со всеми tf.distribute.Strategy.

@tf.function def step_fn(var):

def merge_fn(strategy, value, var): # Всеобъемлющее уменьшение значения. Обратите внимание, что value здесь # tf.distribute.DistributedValues. reduced = strategy.extended.reduce_to(tf.distribute.ReduceOp.SUM, value, destinations=var) strategy.extended.update(var, lambda var, value: var.assign(value), args=(reduced,))

value = tf.identity(1.) tf.distribute.get_replica_context().merge_call(merge_fn, args=(value, var))

def run(strategy): with strategy.scope(): v = tf.Variable(0.) strategy.run(step_fn, args=(v,)) return v

run(tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])) MirroredVariable:{ 0: , 1: } run(tf.distribute.experimental.CentralStorageStrategy( compute_devices=["GPU:0", "GPU:1"], parameter_device="CPU:0")) run(tf.distribute.OneDeviceStrategy("GPU:0"))

update

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

update(
    var, fn, args=(), kwargs=None, group=True
)

Выполните fn для обновления var, используя входные данные, зеркально отображённые на тех же устройствах.

tf.distribute.StrategyExtended.update принимает распределенную переменную var для обновления, функцию обновления fn и args и kwargs для fn. Она применяет fn к каждой компоненте переменной var и передает соответствующие значения из args и kwargs. Ни args, ни kwargs не могут содержать значения для каждой реплики. Если они содержат зеркальные значения, они будут развернуты перед вызовом fn. Например, fn может быть assign_add, а args может быть зеркальными DistributedValues, где каждый компонент содержит значение, которое нужно добавить к этой зеркальной переменной var. Вызов update вызовет assign_add для каждой компоненты переменной var со соответствующим тензорным значением на этом устройстве.

Пример использования:

strategy = tf.distribute.MirroredStrategy(['GPU:0', 'GPU:1']) # With 2
devices
with strategy.scope():
  v = tf.Variable(5.0, aggregation=tf.VariableAggregation.SUM)
def update_fn(v):
  return v.assign(1.0)
result = strategy.extended.update(v, update_fn)
# result is
# Mirrored:{
#  0: tf.Tensor(1.0, shape=(), dtype=float32),
#  1: tf.Tensor(1.0, shape=(), dtype=float32)
# }

Если var зеркально отображается на нескольких устройствах, этот метод реализует следующую логику:

results = {}
for device, v in var:
  with tf.device(device):
    # args and kwargs will be unwrapped if they are mirrored.
    results[device] = fn(v, *args, **kwargs)
return merged(results)

В противном случае этот метод возвращает fn(var, *args, **kwargs), сокооперируемое с var.

update_non_slot

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

update_non_slot(
    colocate_with, fn, args=(), kwargs=None, group=True
)

Выполняет fn(*args, **kwargs) на устройствах colocate_with.

Используется для обновления переменных, не являющихся слотами.

value_container

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

value_container(
    value
)

Возвращает контейнер, к которому принадлежит эта реплицированная value.

variable_created_in_scope

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

variable_created_in_scope(
    v
)

Проверяет, была ли v создана во время активности этого контекста стратегии.

Переменные, созданные внутри контекста стратегии, "принадлежат" ей:

strategy = tf.distribute.MirroredStrategy()
with strategy.scope():
  v = tf.Variable(1.)
strategy.extended.variable_created_in_scope(v)
True

Переменные, созданные вне стратегии, не принадлежат ей:

strategy = tf.distribute.MirroredStrategy()
v = tf.Variable(1.)
strategy.extended.variable_created_in_scope(v)
False