tf.train.Coordinator

Координатор потоков.

tf.train.Coordinator(
    clean_stop_exception_types=None
)

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

Использование:

# Create a coordinator.
coord = Coordinator()
# Start a number of threads, passing the coordinator to each of them.
...start thread 1...(coord, ...)
...start thread N...(coord, ...)
# Wait for all the threads to terminate.
coord.join(threads)

Любой из потоков может вызвать coord.request_stop(), чтобы запросить остановку всех потоков. Чтобы обработать этот запрос, каждый поток должен регулярно проверять coord.should_stop(). coord.should_stop() возвращает True сразу после вызова coord.request_stop().

Типичный поток, работающий с координатором, выполняет примерно следующее:

while not coord.should_stop():
  ...do some work...

Обработка исключений:

Поток может сообщить координатору об исключении в рамках вызова request_stop(). Исключение будет повторно вызвано при вызове coord.join().

Код потока:

try:
  while not coord.should_stop():
    ...do some work...
except Exception as e:
  coord.request_stop(e)

Основной код:

try:
  ...
  coord = Coordinator()
  # Start a number of threads, passing the coordinator to each of them.
  ...start thread 1...(coord, ...)
  ...start thread N...(coord, ...)
  # Wait for all the threads to terminate.
  coord.join(threads)
except Exception as e:
  ...exception that was passed to coord.request_stop()

Чтобы упростить реализацию потока, Coordinator предоставляет обработчик контекста stop_on_exception(), который автоматически запрашивает остановку при возникновении исключения. С помощью этого обработчика приведенный выше код потока можно записать так:

with coord.stop_on_exception():
  while not coord.should_stop():
    ...do some work...

Льготный период для остановки:

После вызова потоком coord.request_stop() другим потокам предоставляется фиксированное время для остановки; оно называется «льготным периодом остановки» и по умолчанию составляет 2 минуты. Если по истечении льготного периода какой-либо поток все еще работает, coord.join() вызывает RuntimeError с указанием отставших потоков.

try:
  ...
  coord = Coordinator()
  # Start a number of threads, passing the coordinator to each of them.
  ...start thread 1...(coord, ...)
  ...start thread N...(coord, ...)
  # Wait for all the threads to terminate, give them 10s grace period
  coord.join(threads, stop_grace_period_secs=10)
except RuntimeError:
  ...one of the threads took more than 10s to stop after request_stop()
  ...was called.
except Exception:
  ...exception that was passed to coord.request_stop()

Методы

clear_stop

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

clear_stop()

Сбрасывает флаг остановки.

После этого вызовы should_stop() будут возвращать False.

join

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

join(
    threads=None, stop_grace_period_secs=120, ignore_live_threads=False
)

Ожидает завершения потоков.

Этот вызов блокируется до завершения набора потоков. Набор потоков представляет собой объединение потоков, переданных в аргумент threads, и списка потоков, зарегистрированных в координаторе вызовом Coordinator.register_thread().

После остановки потоков, если в request_stop было передано exc_info, это исключение будет вызвано повторно.

Обработка льготного периода: после вызова request_stop() потокам дается 'stop_grace_period_secs' секунд на завершение. Если по истечении этого времени какой-либо из них все еще работает, возникает RuntimeError. Обратите внимание: если в request_stop() было передано exc_info, оно вызывается вместо этого RuntimeError.

raise_requested_exception

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

raise_requested_exception()

Если исключение было передано в request_stop, вызывает его.

register_thread

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

register_thread(
    thread
)

Регистрирует поток для присоединения.

request_stop

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

request_stop(
    ex=None
)

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

После этого вызовы should_stop() будут возвращать True.

Примечание: Передаваемое исключение должно находиться в контексте обработки исключения (то есть try: ... except Exception as ex: ...), а не быть вновь созданным.

should_stop

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

should_stop()

Проверяет, была ли запрошена остановка.

stop_on_exception

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

@contextlib.contextmanager
stop_on_exception()

Менеджер контекста, запрашивающий остановку при возникновении исключения.

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

Этот обработчик контекста упрощает обработку исключений. Используйте его следующим образом:

with coord.stop_on_exception():
  # Any exception raised in the body of the with
  # clause is reported to the coordinator before terminating
  # the execution of the body.
  ...body...

Это полностью эквивалентно немного более длинному коду:

try:
  ...body...
except:
  coord.request_stop(sys.exc_info())

wait_for_stop

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

wait_for_stop(
    timeout=None
)

Ожидает команды остановки для Coordinator.