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

Для упрощения реализации потоков координатор предоставляет обработчик контекста 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
)

Ожидание, пока координатор не получит команду на остановку.