tf.compat.v1.distribute.experimental.ParameterServerStrategy
Асинхронная стратегия распределения tf.distribute для сервера параметров с несколькими узлами.
Наследуется от: Strategy
tf.compat.v1.distribute.experimental.ParameterServerStrategy(
cluster_resolver=None
)
Эта стратегия требует двух ролей: рабочие узлы и серверы параметров. Переменные и обновления этих переменных будут назначены серверам параметров, а другие операции — рабочим узлам.
Когда на каждом рабочем узле имеется более одного графического процессора, операции будут дублироваться на всех графических процессорах. Хотя операции могут дублироваться, переменные — нет, и каждый рабочий узел разделяет общее представление о том, какому серверу параметров назначена переменная.
По умолчанию она использует TFConfigClusterResolver для обнаружения конфигураций для обучения с несколькими узлами. Для этого требуется переменная среды 'TF_CONFIG', и 'TF_CONFIG' должен содержать описание кластера.
Этот класс предполагает, что каждый рабочий узел независимо выполняет одинаковый код, но серверы параметров выполняют стандартный сервер. Это означает, что в то время как каждый рабочий узел синхронно вычисляет одно обновление градиента на всех графических процессорах, обновления между рабочими узлами выполняются асинхронно. Операции, которые выполняются только на первом реплике (например, инкрементирование глобального шага), будут выполняться на первом реплике каждого рабочего узла.
Ожидается, что вы вызовете call_for_each_replica(fn, ...) для любых операций, которые потенциально могут быть дублированы на репликах (т.е. нескольких графических процессорах), даже если имеется только процессор или один графический процессор. При определении fn, необходимо соблюдать особую осторожность:
1) Обычно не рекомендуется открывать область устройства в области действия стратегии. Область устройства (т.е. вызов tf.device) будет объединена с областью устройства или заменит её для операций, но не изменит устройство для переменных.
2) Также не рекомендуется открывать область коллокации (т.е. вызов tf.compat.v1.colocate_with) в области действия стратегии. Для коллокации переменных используйте strategy.extended.colocate_vars_with вместо этого. Коллокация операций может создать конфликты в назначении устройства.
Примечание: Эта стратегия работает только с API Estimator. Передайте экземпляр этой стратегии аргументуexperimental_distributeпри созданииRunConfig. Этот экземплярRunConfigдолжен быть передан экземпляруEstimator, на котором вызываетсяtrain_and_evaluate.
Пример:
strategy = tf.distribute.experimental.ParameterServerStrategy()
run_config = tf.estimator.RunConfig(
experimental_distribute.train_distribute=strategy)
estimator = tf.estimator.Estimator(config=run_config)
tf.estimator.train_and_evaluate(estimator,...)
| Аргументы | |
|---|---|
cluster_resolver | Дополнительный объект tf.distribute.cluster_resolver.ClusterResolver. По умолчанию — tf.distribute.cluster_resolver.TFConfigClusterResolver. |
| Атрибуты | |
|---|---|
cluster_resolver | Возвращает решатель кластера, связанный с данной стратегией. В общем случае, при использовании многоузловой стратегии Стратегии, которые намерены иметь связанный Одноузловые стратегии обычно не имеют Решатель кластера может быть полезен, когда пользователю необходимо получить доступ к такой информации, как описание кластера, тип задачи или идентификатор задачи. Например,
os.environ['TF_CONFIG'] = json.dumps({
'cluster': {
'worker': ["localhost:12345", "localhost:23456"],
'ps': ["localhost:34567"]
},
'task': {'type': 'worker', 'index': 0}
})
# This implicitly uses TF_CONFIG for the cluster and current task info.
strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
...
if strategy.cluster_resolver.task_type == 'worker':
# Perform something that's only applicable on workers. Since we set this
# as a worker above, this block will run on this particular instance.
elif strategy.cluster_resolver.task_type == 'ps':
# Perform something that's only applicable on parameter servers. Since we
# set this as a worker above, this block will not run on this particular
# instance.
Дополнительную информацию см. в документации API |
extended | tf.distribute.StrategyExtended с дополнительными методами. |
num_replicas_in_sync | Возвращает количество реплик, по которым агрегируются градиенты. |
Методы
distribute_datasets_from_function
distribute_datasets_from_function(
dataset_fn, options=None
)
Распределяет экземпляры tf.data.Dataset, созданные при вызовах dataset_fn.
Аргумент dataset_fn, который передают пользователи, — это функция ввода с аргументом tf.distribute.InputContext, которая возвращает экземпляр tf.data.Dataset. Ожидается, что возвращаемый из dataset_fn набор данных уже сгруппирован по размеру пакета на реплику (т.е. глобальный размер пакета, деленный на количество реплик в синхронизации), и разделен. tf.distribute.Strategy.distribute_datasets_from_function не группирует и не разделяет экземпляр tf.data.Dataset, возвращаемый функцией ввода. dataset_fn будет вызываться на процессоре CPU каждого из рабочих узлов, и каждый из них создает набор данных, где каждая реплика на этом рабочем узле будет извлекать одну партию ввода (т.е. если у рабочего узла две реплики, две партии будут извлекаться из Dataset на каждом шаге).
Этот метод можно использовать для нескольких целей. Во-первых, он позволяет указать собственную логику группирования и разделения. (В отличие от tf.distribute.experimental_distribute_dataset, которая выполняет группирование и разделение за вас.) Например, там, где experimental_distribute_dataset не может разделить входные файлы, этот метод может быть использован для ручного разделения набора данных (избегая медленного поведения по умолчанию в experimental_distribute_dataset). В случаях, когда набор данных бесконечный, это разделение можно выполнить, создав реплики наборов данных, отличающиеся только случайным числом.
Функция dataset_fn должна принимать экземпляр tf.distribute.InputContext, где можно получить доступ к информации о группировании и репликации ввода.
Вы можете использовать свойство element_spec возвращаемого этим API tf.distribute.DistributedDataset, чтобы запросить tf.TypeSpec элементов, возвращаемых итератором. Это можно использовать для установки свойства input_signature функции tf.function. Следуйте tf.distribute.DistributedDataset.element_spec, чтобы увидеть пример.
Примечание: Если вы используете TPUStrategy, порядок обработки данных рабочими узлами при использованииtf.distribute.Strategy.experimental_distribute_datasetилиtf.distribute.Strategy.distribute_datasets_from_functionне гарантируется. Это обычно необходимо, если вы используетеtf.distributeдля масштабирования предсказаний. Однако вы можете вставить индекс для каждого элемента в пакет и упорядочить выходы соответствующим образом. Обратитесь к этому фрагменту для примера упорядочения выходов.
Примечание: Трансформации состояний набора данных в настоящее время не поддерживаются сtf.distribute.experimental_distribute_datasetилиtf.distribute.distribute_datasets_from_function. Любые операции состояния, которые может иметь набор данных, в настоящее время игнорируются. Например, если ваш набор данных имеет преобразование состоянияmap_fn, которое используетtf.random.uniformдля вращения изображения, тогда у вас есть граф набора данных, зависящий от состояния (т.е. случайного числа) на локальной машине, где выполняется python-процесс.
Для получения дополнительной информации об использовании и свойствах этого метода обратитесь к учебнику по распределенному вводу. Если вас интересует обработка последней частичной партии, прочитайте эту секцию.
| Аргументы | |
|---|---|
dataset_fn | Функция, принимающая экземпляр tf.distribute.InputContext и возвращающая tf.data.Dataset. |
options | tf.distribute.InputOptions, используемые для управления параметрами распределения набора данных. |
| Возвращаемое значение | |
|---|---|
tf.distribute.DistributedDataset. |
experimental_distribute_dataset
experimental_distribute_dataset(
dataset, options=None
)
Создает tf.distribute.DistributedDataset из tf.data.Dataset.
Возвращаемый tf.distribute.DistributedDataset можно перебирать аналогично обычным наборам данных. ПРИМЕЧАНИЕ: пользователь не может добавить больше преобразований к tf.distribute.DistributedDataset. Вы можете только создать итератор или проверить tf.TypeSpec генерируемых данных. См. документацию API tf.distribute.DistributedDataset для получения дополнительной информации.
Следующий пример:
global_batch_size = 2
# Passing the devices is optional.
strategy = tf.distribute.MirroredStrategy(devices=["GPU:0", "GPU:1"])
# Create a dataset
dataset = tf.data.Dataset.range(4).batch(global_batch_size)
# Distribute that dataset
dist_dataset = strategy.experimental_distribute_dataset(dataset)
@tf.function
def replica_fn(input):
return input*2
result = []
# Iterate over the `tf.distribute.DistributedDataset`
for x in dist_dataset:
# process dataset elements
result.append(strategy.run(replica_fn, args=(x,)))
print(result)
[PerReplica:{
0: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([0])>,
1: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([2])>
}, PerReplica:{
0: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([4])>,
1: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([6])>
}]
Три ключевых действия, происходящих под капотом этого метода, это пайпинг, фрагментация и предварительная выборка.
В приведённом выше фрагменте кода dataset пайпится global_batch_size, и вызов experimental_distribute_dataset на нём перепайпивает dataset в новый размер пакета, равный глобальному размеру пакета, делённому на количество реплик в синхронизации. Мы проходим по нему с помощью питонической циклической конструкции. x — это tf.distribute.DistributedValues, содержащий данные для всех реплик, и каждая реплика получает данные нового размера пакета. tf.distribute.Strategy.run позаботится о подаче правильных данных на реплику в x правильным replica_fn на каждой реплике.
Фрагментация включает в себя автоматическую фрагментацию по нескольким рабочим узлам и внутри каждого рабочего узла. Во-первых, в распределённом обучении с несколькими рабочими узлами (т.е. когда вы используете tf.distribute.experimental.MultiWorkerMirroredStrategy или tf.distribute.TPUStrategy), автоматическая фрагментация набора данных по нескольким рабочим узлам означает, что каждому рабочему узлу назначается подмножество всего набора данных (если установлена правильная tf.data.experimental.AutoShardPolicy). Это делается для того, чтобы на каждом шаге глобальный размер пакета неперекрывающихся элементов набора данных обрабатывался каждым рабочим узлом. Автоматическая фрагментация имеет несколько различных вариантов, которые можно указать с помощью tf.data.experimental.DistributeOptions. Затем фрагментация внутри каждого рабочего узла означает, что метод разделит данные между всеми устройствами рабочего узла (если их более одного). Это произойдёт независимо от автоматической фрагментации по нескольким рабочим узлам.
Примечание: для автоматической фрагментации по нескольким рабочим узлам, режим по умолчанию —tf.data.experimental.AutoShardPolicy.AUTO. Этот режим попытается разбить входной набор данных по файлам, если набор данных создаётся из наборов данных-чтецов (например,tf.data.TFRecordDataset,tf.data.TextLineDatasetи т.д.) или в противном случае разбить набор данных по данным, где каждый из рабочих узлов будет читать весь набор данных и обрабатывать только назначенный ему фрагмент. Однако, если у вас меньше одного входного файла на рабочий узел, мы рекомендуем отключить автоматическую фрагментацию набора данных по рабочим узлам, установивtf.data.experimental.DistributeOptions.auto_shard_policyвtf.data.experimental.AutoShardPolicy.OFF.
По умолчанию этот метод добавляет преобразование предварительной выборки в конце предоставленного пользователем экземпляра tf.data.Dataset. Аргумент преобразования предварительной выборки, который равен buffer_size, равен количеству реплик в синхронизации.
Если вышеупомянутая логика разделения пакетов и фрагментации наборов данных нежелательна, используйте tf.distribute.Strategy.distribute_datasets_from_function вместо неё, так как он не выполняет автоматическое пайпирование или фрагментацию за вас.
Примечание: Если вы используете TPUStrategy, порядок обработки данных рабочими узлами при использованииtf.distribute.Strategy.experimental_distribute_datasetилиtf.distribute.Strategy.distribute_datasets_from_functionне гарантируется. Это обычно требуется, если вы используетеtf.distributeдля масштабирования прогнозирования. Однако вы можете вставить индекс для каждого элемента в пакете и упорядочить выходные данные соответственно. См. этот фрагмент для примера того, как упорядочить выходные данные.
Примечание: Трансформации наборов данных с состоянием в настоящее время не поддерживаются сtf.distribute.experimental_distribute_datasetилиtf.distribute.distribute_datasets_from_function. Любые операторы с состоянием, которые может иметь набор данных, в настоящее время игнорируются. Например, если ваш набор данных содержитmap_fnкоторый используетtf.random.uniformдля поворота изображения, то у вас есть граф набора данных, который зависит от состояния (т.е. случайного семени) на локальной машине, где выполняется процесс Python.
Для ознакомления с дополнительными вариантами использования и свойствами этого метода обратитесь к учебнику по распределённому вводу. Если вас интересует обработка последнего частичного пакета, прочитайте эту секцию.
| Аргументы | |
|---|---|
dataset | tf.data.Dataset, который будет фрагментирован по всем репликам в соответствии с правилами, указанными выше. |
options | tf.distribute.InputOptions, используемый для управления параметрами распределения этого набора данных. |
| Возвращает | |
|---|---|
tf.distribute.DistributedDataset. |
experimental_local_results
experimental_local_results(
value
)
Возвращает список всех локальных значений на реплику, содержащихся в value.
Примечание: Это возвращает только значения на рабочем узле, инициированном этим клиентом. При использованииtf.distribute.Strategy, такого какtf.distribute.experimental.MultiWorkerMirroredStrategy, каждый рабочий узел будет своим клиентом, и эта функция вернёт только значения, вычисленные на этом рабочем узле.
| Аргументы | |
|---|---|
value | Значение, возвращаемое experimental_run(), run(), or a variable created inscope`. |
| Возвращает | |
|---|---|
Кортеж значений, содержащихся в value, где i-й элемент соответствует i-й реплике. Если value представляет собой единственное значение, это возвращает (value,). |
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)
| Аргументы | |
|---|---|
numpy_input | Вложенный массив NumPy входных данных, который будет преобразован в набор данных. Обратите внимание, что списки массивов NumPy объединяются, так как это стандартное поведение tf.data.Dataset. |
session | (Только выполнение графика TensorFlow v1.x) Сессия, используемая для инициализации. |
| Возвращает | |
|---|---|
tf.data.Dataset, представляющий numpy_input. |
experimental_run
experimental_run(
fn, input_iterator=None
)
Выполняет операции в fn на каждой реплике с входными данными из input_iterator. (устарело)
При включённом режиме выполнения операций, выполняет операции, указанные в fn на каждой реплике. В противном случае создаёт граф для выполнения операций на каждой реплике.
Каждая реплика получит один, отличающийся вход из входных данных, предоставленных одним вызовом get_next на итераторе входных данных.
fn может вызвать tf.distribute.get_replica_context() для доступа к таким элементам, как replica_id_in_sync_group.
| Аргументы | |
|---|---|
fn | Функция для выполнения. Входные данные функции должны соответствовать выходам input_iterator.get_next(). Выход должен быть tf.nest Tensors. |
input_iterator | (Необязательно) Итератор входных данных, из которого берутся входные данные. |
| Возвращает | |
|---|---|
Объединённое возвращаемое значение fn по репликам. Структура возвращаемого значения такая же, как и у возвращаемого значения fn. Каждый элемент структуры может быть PerReplica (если значения не синхронизированы), Mirrored (если значения синхронизированы) или Tensor (если выполняется на одной реплике). |
make_dataset_iterator
make_dataset_iterator(
dataset
)
Создаёт итератор для входных данных, предоставленных через dataset.
Данные из заданного набора данных будут распределены равномерно по всем вычислительным репликам. Мы будем предполагать, что входной набор данных сгруппирован по глобальному размеру пакета. С этим предположением мы будем прилагать все усилия для разделения каждого пакета по всем репликам (один или несколько рабочих узлов). Если эти усилия окажутся безуспешными, будет выброшено исключение, и пользователь должен вместо этого использовать make_input_fn_iterator, которое предоставляет пользователю больший контроль и не пытается разделить пакет между репликами.
Пользователь также может использовать make_input_fn_iterator, если хочет настроить, какие входные данные передаются какой реплике/рабочему узлу и т.д.
| Аргументы | |
|---|---|
dataset | tf.data.Dataset, который будет распределён равномерно по всем репликам. |
| Возвращаемое значение | |
|---|---|
Объект tf.distribute.InputIterator, который возвращает входные данные для каждого шага вычисления. Пользователь должен вызвать initialize на возвращённом итераторе. |
make_input_fn_iterator
make_input_fn_iterator(
input_fn,
replication_mode=tf.distribute.InputReplicationMode.PER_WORKER
)
Возвращает итератор, разделённый по репликам, созданный из функции ввода.
Функция 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.
| Аргументы | |
|---|---|
input_fn | Функция, которая принимает объект tf.distribute.InputContext и возвращает tf.data.Dataset. |
replication_mode | Значение перечисления tf.distribute.InputReplicationMode. В настоящее время поддерживается только PER_WORKER, что означает, что будет один вызов input_fn на каждый рабочий узел. Реплики будут извлекать данные из локального tf.data.Dataset на своём рабочем узле. |
| Возвращаемое значение | |
|---|---|
Объект итератора, который сначала должен быть .initialize()-ен. Затем он может быть передан в strategy.experimental_run() или вы можете вызвать iterator.get_next() для получения следующего значения, которое следует передать в strategy.extended.call_for_each_replica(). |
reduce
reduce(
reduce_op, value, axis=None
)
Выполняет агрегацию value по репликам и возвращает результат на текущем устройстве.
strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
def step_fn():
i = tf.distribute.get_replica_context().replica_id_in_sync_group
return tf.identity(i)
per_replica_result = strategy.run(step_fn)
total = strategy.reduce("SUM", per_replica_result, axis=None)
total
<tf.Tensor: shape=(), dtype=int32, numpy=1>
Чтобы увидеть, как это будет выглядеть с несколькими репликами, рассмотрим тот же пример с MirroredStrategy с 2 графическими процессорами:
strategy = tf.distribute.MirroredStrategy(devices=["GPU:0", "GPU:1"])
def step_fn():
i = tf.distribute.get_replica_context().replica_id_in_sync_group
return tf.identity(i)
per_replica_result = strategy.run(step_fn)
# Check devices on which per replica result is:
strategy.experimental_local_results(per_replica_result)[0].device
# /job:localhost/replica:0/task:0/device:GPU:0
strategy.experimental_local_results(per_replica_result)[1].device
# /job:localhost/replica:0/task:0/device:GPU:1
total = strategy.reduce("SUM", per_replica_result, axis=None)
# Check device on which reduced result is:
total.device
# /job:localhost/replica:0/task:0/device:CPU:0
Этот API обычно используется для агрегирования результатов, возвращаемых разными репликами, для отчётности и т.д. Например, вычисленные по разным репликам значения потерь могут быть усреднены с помощью этого API перед выводом.
Примечание: Результат копируется на "текущее" устройство — обычно это процессор рабочего узла, на котором выполняется программа. ДляTPUStrategy, это первый хост TPU. Для многоклиентскойMultiWorkerMirroredStrategy, это процессор каждого рабочего узла.
Существует ряд различных API tf.distribute для агрегирования значений по репликам:
-
tf.distribute.ReplicaContext.all_reduce: Это отличается отStrategy.reduceтем, что предназначен для контекста реплики и не копирует результаты на устройство хоста.all_reduceобычно используется для сокращений внутри шага обучения, таких как градиенты. -
tf.distribute.StrategyExtended.reduce_toиtf.distribute.StrategyExtended.batch_reduce_to: Эти API являются более продвинутыми версиямиStrategy.reduce, так как они позволяют настраивать место назначения результата. Они также вызываются в контексте между репликами.
Каким должен быть ось?
Учитывая значение для каждой реплики, возвращённое функцией run, например, потерю на пример, пакет делится между всеми репликами. Эта функция позволяет агрегировать значения по репликам, а также по элементам пакета, если указан параметр axis соответственно.
Например, если у вас есть глобальный размер пакета 8 и 2 реплики, значения для примеров [0, 1, 2, 3] будут на реплике 0, а [4, 5, 6, 7] — на реплике 1. С помощью axis=None, reduce агрегирует только по репликам, возвращая [0+4, 1+5, 2+6, 3+7]. Это полезно, когда каждая реплика вычисляет скаляр или какое-то другое значение, у которого нет "размерности пакета" (например, градиент или потеря).
strategy.reduce("sum", per_replica_result, axis=None)
Иногда вам нужно агрегировать как по глобальному пакету, так и по всем репликам. Этого можно достичь, указав размер пакета в качестве axis, обычно axis=0. В этом случае возвращается скаляр 0+1+2+3+4+5+6+7.
strategy.reduce("sum", per_replica_result, axis=0)
Если есть последний неполный пакет, вам нужно указать ось, чтобы форма результата была согласованной между репликами. Так, если последний пакет имеет размер 6 и он разделён на [0, 1, 2, 3] и [4, 5], у вас возникнет несоответствие форм, если не указать axis=0. Если вы укажете tf.distribute.ReduceOp.MEAN, используя axis=0 будет использован правильный знаменатель 6. Сравните это с вычислением reduce_mean для получения скалярного значения на каждой реплике и этой функцией для усреднения этих средних значений, что будет учитывать некоторые значения 1/8 и другие 1/4.
| Аргументы | |
|---|---|
reduce_op | значение tf.distribute.ReduceOp, определяющее, как должны быть объединены значения. Разрешает использование строкового представления перечисления, например, "СУММА", "СРЕДНЕЕ". |
value | экземпляр tf.distribute.DistributedValues, например, возвращаемый Strategy.run, для объединения в один тензор. Также может быть обычным тензором при использовании с OneDeviceStrategy или стратегией по умолчанию. |
axis | определяет размерность, по которой нужно выполнить сокращение внутри тензора каждой реплики. Обычно следует устанавливать в размерность пакета или None для сокращения только по репликам (например, если тензор не имеет размерности пакета). |
| Возвращаемое значение | |
|---|---|
Tensor. |
run
run(
fn, args=(), kwargs=None, options=None
)
Вызывает fn для каждой реплики с указанными аргументами.
Этот метод является основным способом распределения вычислений с объектом tf.distribute. Он вызывает fn для каждой реплики. Если args или kwargs содержат tf.distribute.DistributedValues, например, те, которые создаются tf.distribute.DistributedDataset из tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.distribute_datasets_from_function, при выполнении fn на конкретной реплике, оно будет выполнено с компонентом tf.distribute.DistributedValues, соответствующим этой реплике.
fn вызывается в контексте реплики. fn может вызвать tf.distribute.get_replica_context() для доступа к членам, таким как all_reduce. Пожалуйста, обратитесь к документированию модуля tf.distribute для понятия контекста реплики.
Все аргументы в args или kwargs могут быть вложенной структурой тензоров, например, списком тензоров, в таком случае args и kwargs будут переданы в вызываемый fn для каждой реплики. Или args или kwargs могут быть tf.distribute.DistributedValues содержащими тензоры или составные тензоры, т.е. tf.compat.v1.TensorInfo.CompositeTensor, в таком случае каждый вызов fn получит компонент tf.distribute.DistributedValues, соответствующий его реплике. Обратите внимание, что произвольные значения Python, которые не относятся к перечисленным выше типам, не поддерживаются.
Пример использования:
- Входные данные тензора-константы.
strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
tensor_input = tf.constant(3.0)
@tf.function
def replica_fn(input):
return input*2.0
result = strategy.run(replica_fn, args=(tensor_input,))
result
PerReplica:{
0: <tf.Tensor: shape=(), dtype=float32, numpy=6.0>,
1: <tf.Tensor: shape=(), dtype=float32, numpy=6.0>
}
- Входные данные DistributedValues.
strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
@tf.function
def run():
def value_fn(value_context):
return value_context.num_replicas_in_sync
distributed_values = (
strategy.experimental_distribute_values_from_function(
value_fn))
def replica_fn2(input):
return input*2
return strategy.run(replica_fn2, args=(distributed_values,))
result = run()
result
<tf.Tensor: shape=(), dtype=int32, numpy=4>
- Использование
tf.distribute.ReplicaContextдля агрегирования значений по всем репликам.
strategy = tf.distribute.MirroredStrategy(["gpu:0", "gpu:1"])
@tf.function
def run():
def value_fn(value_context):
return tf.constant(value_context.replica_id_in_sync_group)
distributed_values = (
strategy.experimental_distribute_values_from_function(
value_fn))
def replica_fn(input):
return tf.distribute.get_replica_context().all_reduce("sum", input)
return strategy.run(replica_fn, args=(distributed_values,))
result = run()
result
PerReplica:{
0: <tf.Tensor: shape=(), dtype=int32, numpy=1>,
1: <tf.Tensor: shape=(), dtype=int32, numpy=1>
}
| Аргументы | |
|---|---|
fn | Функция, которая выполняется на каждой реплике. |
args | Необязательные позиционные аргументы для fn. Его элементы могут быть тензором, вложенной структурой тензоров или tf.distribute.DistributedValues. |
kwargs | Необязательные именованные аргументы для fn. Его элементы могут быть тензором, вложенной структурой тензоров или tf.distribute.DistributedValues. |
options | Необязательный экземпляр tf.distribute.RunOptions, определяющий опции для выполнения fn. |
| Возвращаемое значение | |
|---|---|
Объединенное значение fn по всем репликам. Структура возвращаемого значения такая же, как у возвращаемого значения от fn. Каждый элемент в структуре может быть tf.distribute.DistributedValues, объектами Tensor или Tensor (например, при выполнении на одной реплике). |
scope
scope()
Блок кода для установки текущей стратегии и распределения переменных.
Этот метод возвращает контекстный менеджер и используется следующим образом:
strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
# Variable created inside scope:
with strategy.scope():
mirrored_variable = tf.Variable(1.)
mirrored_variable
MirroredVariable:{
0: <tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>,
1: <tf.Variable 'Variable/replica_1:0' shape=() dtype=float32, numpy=1.0>
}
# Variable created outside scope:
regular_variable = tf.Variable(1.)
regular_variable
<tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>
Что происходит при входе в область Strategy.scope?
-
strategyустанавливается в глобальном контексте в качестве текущей стратегии. Внутри этой областиtf.distribute.get_strategy()теперь вернёт эту стратегию. За пределами этой области возвращается стратегия по умолчанию (без действий). - Вход в область также влечёт вход в "межрепликационный контекст". Подробное описание межрепликационного и репликационного контекстов см. в
tf.distribute.StrategyExtended. - Создание переменных внутри
scopeперехватывается стратегией. Каждая стратегия определяет, как она хочет повлиять на создание переменных. Синхронные стратегии, такие какMirroredStrategy,TPUStrategyиMultiWorkerMiroredStrategy, создают переменные, дублированные на каждой реплике, тогда какParameterServerStrategyсоздаёт переменные на серверах параметров. Это делается с использованием пользовательскогоtf.variable_creator_scope. - В некоторых стратегиях также может быть введена область контекста устройства по умолчанию: в
MultiWorkerMiroredStrategy, на каждом рабочем узле вводится область контекста устройства по умолчанию "/CPU:0".
Примечание: Вход в область не приводит к автоматическому распределению вычисления, за исключением случаев использования высокоуровневых фреймворков обучения, таких как Kerasmodel.fit. Если вы не используетеmodel.fit, вам необходимо использовать APIstrategy.runдля явного распределения вычисления. См. пример в обучающем пособии по настройке цикла обучения настраиваемому циклу обучения.
Что должно быть внутри и что снаружи области?
Существует ряд требований к тому, что должно происходить внутри области. Однако в тех случаях, где у нас есть информация о используемой стратегии, мы часто входим в область для пользователя, чтобы не заставлять его делать это явно (то есть вызов функций внутри или вне области допустим).
- Все, что создаёт переменные, которые должны быть распределенными переменными, должно вызываться внутри области
strategy.scope. Это можно сделать либо, напрямую вызвав функцию создания переменной внутри контекста области, либо, полагаясь на другой API, например,strategy.runилиkeras.Model.fit, который автоматически помещает вас в него. Любая переменная, созданная вне области, не будет распределена и может повлиять на производительность. Некоторые распространённые объекты, создающие переменные в TF, — это модели, оптимизаторы, метрики. Такие объекты всегда должны быть инициализированы внутри области, и любые функции, которые могут лениво создавать переменные (например,Model.__call__(), трассировкаtf.functionи т. д.), аналогично должны вызываться внутри области. Ещё одним источником создания переменных может быть восстановление контрольной точки — когда переменные создаются лениво. Обратите внимание, что любая переменная, созданная внутри стратегии, сохраняет информацию о стратегии. Так чтение и запись этих переменных вне областиstrategy.scopeтакже могут работать беспрепятственно, без необходимости вхождения пользователя в область. - Некоторые API стратегии (такие как
strategy.runиstrategy.reduce), которые требуют вхождения в область стратегии, автоматически входят в область, что означает, что при использовании этих API вам не нужно входить в область явно. - Когда
tf.keras.Modelсоздаётся внутриstrategy.scope, объект модели получает информацию о области. При последующем вызове методов высокоуровневого фреймворка обучения, таких какmodel.compile,model.fitи т. д., захваченная область автоматически входит, и связанная стратегия используется для распределения обучения и т. д. См. подробный пример в обучающем пособии по распределённому Keras по распределённому Keras. ПРЕДУПРЕЖДЕНИЕ: Простое выполнениеmodel(..)не приводит к автоматическому входу в захваченную область — только высокоуровневые API фреймворка обучения поддерживают это поведение:model.compile,model.fit,model.evaluate,model.predictиmodel.saveмогут вызываться внутри или вне области. - Следующее может быть как внутри, так и вне области:
- Создание наборов входных данных
- Определение
tf.functionэлементов, представляющих вашу итерацию обучения - Сохранение API, такие как
tf.saved_model.save. Загрузка создаёт переменные, поэтому она должна выполняться внутри области, если вы хотите обучить модель в распределённом режиме. - Сохранение контрольных точек. Как упоминалось выше —
checkpoint.restoreиногда может потребоваться находиться внутри области, если он создаёт переменные.
| Возвращаемое значение | |
|---|---|
| Менеджер контекста. |
update_config_proto
update_config_proto(
config_proto
)
Возвращает копию config_proto с изменениями, необходимыми для использования с данной стратегией.
Обновлённая конфигурация содержит информацию, необходимую для работы с стратегией, например, конфигурацию для запуска коллективных операций или фильтры устройств для повышения производительности распределённого обучения.
| Аргументы | |
|---|---|
config_proto | объект tf.ConfigProto. |
| Возвращаемое значение | |
|---|---|
Обновлённая копия config_proto. |
© 2022 The TensorFlow Authors. All rights reserved.
Licensed under the Creative Commons Attribution License 4.0.
Code samples licensed under the Apache 2.0 License.
https://www.tensorflow.org/versions/r2.9/api_docs/python/tf/compat/v1/distribute/experimental/ParameterServerStrategy