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().

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

Обработка периода ожидания: Когда request_stop() вызывается, потокам дается 'stop_grace_period_secs' секунд на завершение. Если какой-либо из них все еще жив после истечения этого периода, поднимается RuntimeError. Обратите внимание, что если исключение exc_info передано в request_stop(), то оно поднимается вместо 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
)

Ожидание сигнала о прекращении работы координатора.