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
)

Ожидание сигнала остановки от координатора.