tf.contrib.distribute.CollectiveAllReduceStrategy

Стратегия распределения, использующая коллективные операции для all-reduce.

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

tf.contrib.distribute.CollectiveAllReduceStrategy(
    num_gpus_per_worker=0,
    communication=tf.distribute.experimental.CollectiveCommunication.AUTO
)

*** версия contrib ***

Она похожа на MirroredStrategy, но использует коллективные операции для сокращения.

Когда cluster_spec задаётся методом configure, она превращается в многоузловую версию, которая работает на нескольких узлах с репликацией между графами.

Примечание: configure будет вызываться API верхнего уровня, если выполнение происходит в распределённой среде.

Методы

experimental_distribute_dataset

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

experimental_distribute_dataset(
    dataset
)

Распределяет экземпляр tf.data.Dataset, предоставленный через dataset.

Возвращаемый распределённый набор данных можно итерировать так же, как и обычные наборы данных. ПРИМЕЧАНИЕ: В настоящее время пользователь не может добавить больше преобразований к распределённому набору данных.

Вот пример:

strategy = tf.distribute.MirroredStrategy()

# Create a dataset
dataset = dataset_ops.Dataset.TFRecordDataset([
  "/a/1.tfr", "/a/2.tfr", "/a/3.tfr", "/a/4.tfr"])

# Distribute that dataset
dist_dataset = strategy.experimental_distribute_dataset(dataset)
# Iterate over the distributed dataset
for x in dist_dataset:
  # process dataset elements
  strategy.experimental_run_v2(train_step, args=(x,))

Мы предположим, что входной набор данных сгруппирован по глобальному размеру пакета. При этом предположении мы постараемся разделить каждый пакет между всеми репликами (один или несколько рабочих узлов).

В многоузловой среде мы сначала попытаемся распределить набор данных, попытаемся определить, создан ли набор данных из ReaderDatasets (например, TFRecordDataset, TextLineDataset и т. д.), и если да, попытаемся разбить входные файлы. Обратите внимание, что должно быть как минимум один входной файл на рабочий узел. Если у вас меньше одного входного файла на узел, мы рекомендуем отключить распределение вашего набора данных с помощью метода ниже.

Если эта попытка не удастся (например, набор данных создан из Dataset.range), мы разделим набор данных равномерно в конце, добавив операцию .shard в конец потока обработки. Это заставит весь поток предобработки всех данных выполняться на каждом рабочем узле, и каждый рабочий узел будет выполнять избыточную работу. Мы выведем предупреждение, если этот метод фрагментации будет выбран. В этом случае рассмотрите использование experimental_distribute_datasets_from_function вместо него.

Вы можете отключить фрагментацию набора данных между рабочими узлами, используя параметр auto_shard в tf.data.experimental.DistributeOptions.

Внутри каждого рабочего узла мы также разделим данные между всеми устройствами рабочего узла (если их несколько), и это произойдёт даже если многоузловая фрагментация отключена с помощью метода выше.

Если логика разбиения пакета и фрагментации набора данных нежелательна, используйте experimental_distribute_datasets_from_function вместо неё, которая не выполняет автоматическое разделение или фрагментацию.

experimental_distribute_datasets_from_function

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

experimental_distribute_datasets_from_function(
    dataset_fn
)

Распределяет экземпляры tf.data.Dataset, созданные вызовами к dataset_fn.

dataset_fn будет вызываться один раз для каждого рабочего узла в стратегии. Каждая реплика на этом узле будет извлекать одну партию входных данных из локального Dataset (т. е., если у рабочего узла есть две реплики, в каждом шаге будут извлечены две партии входных данных из Dataset).

Этот метод можно использовать для нескольких целей. Например, там, где experimental_distribute_dataset не может разбить входные файлы, этот метод может использоваться для ручного разбиения набора данных (избегая медленного поведения по умолчанию в experimental_distribute_dataset). В случаях, когда набор данных бесконечен, это разбиение можно выполнить, создав реплики набора данных, которые отличаются только случайным начальным значением. experimental_distribute_dataset также иногда может не разделить пакет между репликами на рабочем узле. В этом случае этот метод можно использовать там, где это ограничение отсутствует.

dataset_fn должен принимать экземпляр tf.distribute.InputContext, где можно получить информацию о пакетировании и репликации входных данных:

def dataset_fn(input_context):
  batch_size = input_context.get_per_replica_batch_size(global_batch_size)
  d = tf.data.Dataset.from_tensors([[1.]]).repeat().batch(batch_size)
  return d.shard(
      input_context.num_input_pipelines, input_context.input_pipeline_id)

inputs = strategy.experimental_distribute_datasets_from_function(dataset_fn)

for batch in inputs:
  replica_results = strategy.experimental_run_v2(replica_fn, args=(batch,))

experimental_local_results

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

experimental_local_results(
    value
)

Возвращает список всех локальных значений по репликам, содержащихся в value.

Примечание: Это возвращает только значения на рабочем узле, инициированном этим клиентом. При использовании tf.distribute.Strategy, такого как tf.distribute.experimental.MultiWorkerMirroredStrategy, каждый рабочий узел будет своим клиентом, и эта функция будет возвращать только вычисленные на этом рабочем узле значения.

experimental_make_numpy_dataset

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

experimental_make_numpy_dataset(
    numpy_input, session=None
)

Создаёт tf.data.Dataset для входных данных, предоставленных через массив NumPy.

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

Обратите внимание, что, скорее всего, вам потребуется использовать tf.distribute.Strategy.experimental_distribute_dataset с возвращённым набором данных для его дальнейшего распределения с помощью стратегии.

Пример:

numpy_input = np.ones([10], dtype=np.float32)
dataset = strategy.experimental_make_numpy_dataset(numpy_input)
dist_dataset = strategy.experimental_distribute_dataset(dataset)

experimental_run

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

experimental_run(
    fn, input_iterator=None
)

Выполняет операции в fn на каждой реплике с входными данными из input_iterator.

УСТЕРЕЖДЕННО: Этот метод недоступен в TF 2.x. Пожалуйста, переключитесь на использование experimental_run_v2 вместо него.

При включённом выполнении eager, выполняет операции, заданные fn на каждой реплике. В противном случае строит граф для выполнения операций на каждой реплике.

Каждая реплика получит один, разный вход из входных данных, предоставленных одним вызовом get_next для итератора входных данных.

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

experimental_run_v2

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

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

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

Выполняет операции, указанные в fn, на каждой реплике. Если args или kwargs имеют значения "на реплику", такие как те, что получены с помощью "распределённого Dataset", при выполнении fn на конкретной реплике, оно будет выполнено с компонентом этих значений "на реплику", соответствующим этой реплике.

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

Все аргументы в args или kwargs должны быть либо вложенными тензорами, либо объектами "на реплику", содержащими тензоры или составные тензоры.

make_dataset_iterator

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

make_dataset_iterator(
    dataset
)

Создаёт итератор для входных данных, предоставленных через dataset.

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

Данные из заданного набора данных будут распределены равномерно по всем вычислительным репликам. Мы предположим, что входной набор данных сгруппирован по глобальному размеру пакета. При этом предположении мы сделаем всё возможное, чтобы разделить каждый пакет по всем репликам (один или несколько рабочих узлов). Если эта попытка окажется неудачной, будет выброшено исключение, и пользователь должен вместо этого использовать make_input_fn_iterator , который предоставляет больше контроля пользователю и не пытается разделить пакет между репликами.

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

make_input_fn_iterator

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

make_input_fn_iterator(
    input_fn, replication_mode=tf.distribute.InputReplicationMode.PER_WORKER
)

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

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

Функция input_fn должна принимать объект tf.distribute.InputContext, где можно получить информацию о группировке по пакетам и фрагментации входных данных:

def input_fn(input_context):
  batch_size = input_context.get_per_replica_batch_size(global_batch_size)
  d = tf.data.Dataset.from_tensors([[1.]]).repeat().batch(batch_size)
  return d.shard(input_context.num_input_pipelines,
                 input_context.input_pipeline_id)
with strategy.scope():
  iterator = strategy.make_input_fn_iterator(input_fn)
  replica_results = strategy.experimental_run(replica_fn, iterator)

tf.data.Dataset, возвращаемый input_fn, должен иметь размер пакета на реплику, который можно вычислить с помощью input_context.get_per_replica_batch_size.

reduce

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

reduce(
    reduce_op, value, axis=None
)

Суммирование value по репликам.

Учитывая значение на реплику, возвращённое experimental_run_v2, например, потерю на пример, пакет будет разделен по всем репликам. Эта функция позволяет агрегировать по репликам, и, необязательно, по элементам пакета. Например, если у вас глобальный размер пакета 8 и 2 реплики, значения для примеров [0, 1, 2, 3] будут на реплике 0, и [4, 5, 6, 7] — на реплике 1. По умолчанию reduce просто агрегирует по репликам, возвращая [0+4, 1+5, 2+6, 3+7]. Это полезно, когда каждая реплика вычисляет скаляр или какое-либо другое значение, не имеющее "размерности пакета" (например, градиент). Чаще всего вам нужно агрегировать по глобальному пакету, что вы можете сделать, указав размерность пакета как axis, обычно axis=0. В этом случае это вернёт скаляр 0+1+2+3+4+5+6+7.

Если есть последний частичный пакет, вам нужно будет указать ось, чтобы результирующая форма была согласованной между репликами. Так что, если последний пакет имеет размер 6 и он разделён на [0, 1, 2, 3] и [4, 5], у вас возникнет несоответствие форм, если вы не укажете axis=0. Если вы укажете tf.distribute.ReduceOp.MEAN, используя axis=0 будет использоваться правитель 6. В отличие от вычисления reduce_mean для получения скалярного значения на каждой реплике и этой функции для усреднения этих средних значений, которая будет взвешивать некоторые значения 1/8 и другие 1/4.

scope

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

scope()

Возвращает менеджер контекста, выбирающий эту стратегию в качестве текущей.

Внутри блока кода with strategy.scope():, эта нить будет использовать создатель переменных, установленный strategy, и войдёт в свой "межрепличный контекст".

update_config_proto

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

update_config_proto(
    config_proto
)

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

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

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