tf.compat.v1.train.Supervisor

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

tf.compat.v1.train.Supervisor(
    graph=None,
    ready_op=USE_DEFAULT,
    ready_for_local_init_op=USE_DEFAULT,
    is_chief=True,
    init_op=USE_DEFAULT,
    init_feed_dict=None,
    local_init_op=USE_DEFAULT,
    logdir=None,
    summary_op=USE_DEFAULT,
    saver=USE_DEFAULT,
    global_step=USE_DEFAULT,
    save_summaries_secs=120,
    save_model_secs=600,
    recovery_wait_secs=30,
    stop_grace_secs=120,
    checkpoint_basename='model.ckpt',
    session_manager=None,
    summary_writer=USE_DEFAULT,
    init_fn=None,
    local_init_run_options=None
)

Этот класс устарел. Вместо него используйте tf.compat.v1.train.MonitoredTrainingSession.

Supervisor — это небольшая оболочка вокруг Coordinator, Saver и SessionManager, которая берет на себя выполнение общих задач программ обучения TensorFlow.

Использование в одной программе

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that will checkpoint the model in '/tmp/mydir'.
  sv = Supervisor(logdir='/tmp/mydir')
  # Get a TensorFlow session managed by the supervisor.
  with sv.managed_session(FLAGS.master) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)

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

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

Supervisor получает уведомление о любом исключении, вызванном одной из служб. После вызова исключения should_stop() возвращает True. В этом случае цикл обучения тоже должен остановиться. Поэтому в цикле обучения нужно проверять sv.should_stop().

Исключения, указывающие на то, что входные данные для обучения исчерпаны, tf.errors.OutOfRangeError, также приводят к тому, что sv.should_stop() возвращает True, но не вызываются повторно из блока with: они указывают на нормальное завершение.

Использование для нескольких реплик

Для обучения с репликами одну и ту же программу развертывают в Cluster. Одна из задач должна быть назначена главной: эта задача отвечает за инициализацию, создание контрольных точек, сводки и восстановление. Остальные задачи используют службы, предоставляемые главной задачей.

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

# Choose a task as the chief. This could be based on server_def.task_index,
# or job_def.name, or job_def.tasks. It's entirely up to the end user.
# But there can be only one *chief*.
is_chief = (server_def.task_index == 0)
server = tf.distribute.Server(server_def)

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that uses log directory on a shared file system.
  # Indicate if you are the 'chief'
  sv = Supervisor(logdir='/shared_directory/...', is_chief=is_chief)
  # Get a Session in a TensorFlow server on the cluster.
  with sv.managed_session(server.target) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)

В главной задаче Supervisor работает точно так же, как в первом примере выше. В остальных задачах sv.managed_session() ожидает инициализации модели, прежде чем вернуть сеанс коду обучения. Для инициализации модели не главные задачи зависят от главной задачи.

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

Примечание: Эта измененная программа по-прежнему корректно работает как одна программа. Одна программа назначает себя главной.

Какую строку master использовать

Независимо от того, запускаете ли вы программу на своем компьютере или в кластере, для флага --master можно использовать следующие значения:

• Указание '' запрашивает внутрипроцессный сеанс, не использующий RPC.

• Указание 'local' запрашивает сеанс, использующий интерфейс RPC «Master» для выполнения программ TensorFlow. Подробнее см. tf.train.Server.create_local_server.

• Указание 'grpc://hostname:port' запрашивает сеанс, использующий интерфейс RPC для подключения к определенному узлу, а также позволяющий внутрипроцессному master обращаться к удаленным рабочим процессам TensorFlow. Часто уместно передать server.target (для некоторого tf.distribute.Server с именем `server).

Расширенное использование

Запуск дополнительных служб

managed_session() запускает службы Checkpoint и Summary (потоки). Если требуется запустить дополнительные службы, их можно запустить в блоке, управляемом managed_session().

Пример: запуск потока для вывода значений функции потерь. Нужно, чтобы этот поток выполнялся каждые 60 секунд, поэтому запустим его с помощью sv.loop().

...
sv = Supervisor(logdir='/tmp/mydir')
with sv.managed_session(FLAGS.master) as sess:
  sv.loop(60, print_loss, (sess, ))
  while not sv.should_stop():
    sess.run(my_train_op)

Запуск меньшего числа служб

managed_session() запускает потоки «summary» и «checkpoint», которые используют либо переданные конструктору необязательные summary_op и saver, либо значения по умолчанию, автоматически созданные supervisor. Если требуется использовать собственную логику создания сводок и контрольных точек, отключите эти службы, передав None параметрам summary_op и saver.

Пример: создание сводок вручную каждые 100 шагов в главной задаче.

# Create a Supervisor with no automatic summaries.
sv = Supervisor(logdir='/tmp/mydir', is_chief=is_chief, summary_op=None)
# As summary_op was None, managed_session() does not start the
# summary thread.
with sv.managed_session(FLAGS.master) as sess:
  for step in range(1000000):
    if sv.should_stop():
      break
    if is_chief and step % 100 == 0:
      # Create the summary every 100 chief steps.
      sv.summary_computed(sess, sess.run(my_summary_op))
    else:
      # Train normally
      sess.run(my_train_op)

Пользовательская инициализация модели

managed_session() поддерживает только инициализацию модели путем запуска init_op или восстановления из последней контрольной точки. Если требуется особая инициализация, см. описание того, как указать local_init_op при создании supervisor. Можно также напрямую использовать SessionManager для создания сеанса и проверки возможности автоматической инициализации.

Методы

Loop

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

Loop(
    timer_interval_secs, target, args=None, kwargs=None
)

Запускает LooperThread, периодически вызывающий функцию.

Если timer_interval_secs равно None, поток многократно вызывает target(*args, **kwargs). В противном случае он вызывает функцию каждые timer_interval_secs секунд. Поток завершается после запроса остановки.

Запущенный поток добавляется в список потоков, управляемых supervisor, поэтому передавать его методу stop() не требуется.

PrepareSession

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

PrepareSession(
    master='',
    config=None,
    wait_for_checkpoint=False,
    max_wait_secs=7200,
    start_standard_services=True
)

Проверяет, готова ли модель к использованию.

Создает сеанс на 'master', восстанавливая или инициализируя модель по мере необходимости, либо ожидает готовности сеанса. Если выполняется в качестве главной задачи и для start_standard_service задано значение True, также вызывает диспетчер сеансов для запуска стандартных служб.

RequestStop

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

RequestStop(
    ex=None
)

Запрашивает у координатора остановку потоков.

См. Coordinator.request_stop().

ShouldStop

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

ShouldStop()

Проверяет, получил ли координатор указание остановиться.

См. Coordinator.should_stop().

StartQueueRunners

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

StartQueueRunners(
    sess, queue_runners=None
)

Запускает потоки для QueueRunners.

Обратите внимание, что обработчики очередей, собранные в графе по ключу QUEUE_RUNNERS, уже запускаются автоматически при создании сеанса с помощью supervisor. Поэтому явный вызов этого метода не требуется, если не нужно запускать обработчики очередей, не собранные в графе.

Очереди несовместимы с eager execution. Для получения данных при включенном eager execution используйте API tf.data.

StartStandardServices

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

StartStandardServices(
    sess
)

Запускает стандартные службы для 'sess'.

Службы запускаются в фоновом режиме. Набор запускаемых служб зависит от параметров конструктора и может включать:

• Поток Summary, вычисляющий сводки каждые save_summaries_secs.
• Поток Checkpoint, сохраняющий модель каждые save_model_secs.
• Поток StepCounter, измеряющий время шага.

Stop

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

Stop(
    threads=None, close_summary_writer=True, ignore_live_threads=False
)

Останавливает службы и координатор.

Сеанс при этом не закрывается.

StopOnException

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

StopOnException()

Обработчик контекста, останавливающий supervisor при возникновении исключения.

См. Coordinator.stop_on_exception().

SummaryComputed

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

SummaryComputed(
    sess, summary, global_step=None
)

Указывает, что сводка была вычислена.

WaitForStop

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

WaitForStop()

Блокирует выполнение, ожидая остановки координатора.

loop

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

loop(
    timer_interval_secs, target, args=None, kwargs=None
)

Запускает LooperThread, периодически вызывающий функцию.

Если timer_interval_secs равно None, поток многократно вызывает target(*args, **kwargs). В противном случае он вызывает функцию каждые timer_interval_secs секунд. Поток завершается после запроса остановки.

Запущенный поток добавляется в список потоков, управляемых supervisor, поэтому передавать его методу stop() не требуется.

managed_session

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

@contextlib.contextmanager
managed_session(
    master='',
    config=None,
    start_standard_services=True,
    close_summary_writer=True
)

Возвращает менеджер контекста для управляемого сеанса.

Этот менеджер контекста создает сеанс и автоматически восстанавливает его при необходимости. При необходимости он запускает стандартные службы для работы с контрольными точками и сводками. Он отслеживает исключения, возникающие в блоке with или в службах, и при необходимости останавливает supervisor.

Менеджер контекста обычно используется следующим образом:

def train():
  sv = tf.compat.v1.train.Supervisor(...)
  with sv.managed_session(<master>) as sess:
    for step in range(..):
      if sv.should_stop():
        break
      sess.run(<my training op>)
      ...do other things needed at each training step...

Исключение, возникшее в блоке with или в одном из потоков служб, повторно вызывается при выходе из блока. Это происходит после остановки всех потоков и закрытия сеанса. Например, исключение AbortedError, возникающее при вытеснении одного из рабочих процессов распределенной модели, повторно вызывается при выходе из блока.

Если при вытеснении нужно повторить цикл обучения, это можно сделать следующим образом:

def main(...):
  while True
    try:
      train()
    except tf.errors.Aborted:
      pass

В особом случае исключения, используемые для управления потоком выполнения, например OutOfRangeError, сообщающее об исчерпании входных очередей, не вызываются повторно из блока with: они указывают на штатное завершение цикла обучения и считаются нормальным завершением.

prepare_or_wait_for_session

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

prepare_or_wait_for_session(
    master='',
    config=None,
    wait_for_checkpoint=False,
    max_wait_secs=7200,
    start_standard_services=True
)

Убедитесь, что модель готова к использованию.

Создайте сеанс на 'master', восстановив или инициализировав модель при необходимости, либо дождитесь готовности сеанса. Если вы работаете в роли главного узла и для start_standard_service установлено значение True, также вызовите диспетчер сеансов, чтобы запустить стандартные службы.

request_stop

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

request_stop(
    ex=None
)

Запросить у координатора остановку потоков.

См. Coordinator.request_stop().

should_stop

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

should_stop()

Проверить, получил ли координатор команду остановиться.

См. Coordinator.should_stop().

start_queue_runners

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

start_queue_runners(
    sess, queue_runners=None
)

Запустить потоки для QueueRunners.

Обратите внимание, что обработчики очередей, собранные в графе с ключом QUEUE_RUNNERS, запускаются автоматически при создании сеанса с помощью супервизора. Поэтому явно вызывать этот метод не нужно, если только не требуется запустить обработчики очередей, которые не были собраны.

Очереди несовместимы с eager execution. Чтобы загружать данные при включенном eager execution, используйте API tf.data.

start_standard_services

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

start_standard_services(
    sess
)

Запустить стандартные службы для 'sess'.

Службы запускаются в фоновом режиме. Запускаемые службы зависят от параметров конструктора и могут включать:

• Поток Summary, вычисляющий сводки каждые save_summaries_secs.
• Поток Checkpoint, сохраняющий модель каждые save_model_secs.
• Поток StepCounter, измеряющий время выполнения шага.

stop

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

stop(
    threads=None, close_summary_writer=True, ignore_live_threads=False
)

Остановить службы и координатор.

Этот метод не закрывает сеанс.

stop_on_exception

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

stop_on_exception()

Обработчик контекста для остановки супервизора при возникновении исключения.

См. Coordinator.stop_on_exception().

summary_computed

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

summary_computed(
    sess, summary, global_step=None
)

Указать, что сводка была вычислена.

wait_for_stop

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

wait_for_stop()

Заблокировать выполнение до остановки координатора.