tf.CriticalSection

Критическая секция.

tf.CriticalSection(
    name=None, shared_name=None, critical_section_def=None, import_scope=None
)

Объект CriticalSection — это ресурс в графе, который выполняет подграфы в последовательном порядке. Распространённый пример подграфа, который требуется выполнять исключительно, — подграф, задаваемый следующей функцией:

v = resource_variable_ops.ResourceVariable(0.0, name="v")

def count():
  value = v.read_value()
  with tf.control_dependencies([value]):
    with tf.control_dependencies([v.assign_add(1)]):
      return tf.identity(value)

Здесь снимок v сохраняется в value, после чего обновляется v. Возвращается значение снимка.

Если несколько рабочих процессов или потоков параллельно выполняют count, не гарантируется, что доступ к переменной v будет атомарным в какой-либо момент вычисления count в любом потоке. Более того, в настоящее время невозможно даже реализовать атомарный счётчик, гарантирующий, что пользователь увидит каждое значение 0, 1, ...,.

Решение состоит в том, чтобы обеспечить обработку любого доступа к базовому ресурсу v только через критическую секцию:

cs = CriticalSection()
f1 = cs.execute(count)
f2 = cs.execute(count)
output = f1 + f2
session.run(output)

Функции f1 и f2 будут выполняться последовательно, а обновления v будут атомарными.

ПРИМЕЧАНИЯ

Все объекты ресурсов, включая критическую секцию и любые захваченные переменные функций, выполняемых в этой критической секции, будут размещены на одном устройстве (host и cpu/gpu).

При использовании нескольких критических секций для одних и тех же ресурсов эксклюзивный доступ к этим ресурсам не гарантируется. По умолчанию такое поведение запрещено (см. аргумент kwarg exclusive_resource_access).

Например, выполнение одной и той же функции в двух отдельных критических секциях не обеспечит последовательное выполнение:

v = tf.compat.v1.get_variable("v", initializer=0.0, use_resource=True)
def accumulate(up):
  x = v.read_value()
  with tf.control_dependencies([x]):
    with tf.control_dependencies([v.assign_add(up)]):
      return tf.identity(x)
ex1 = CriticalSection().execute(
  accumulate, 1.0, exclusive_resource_access=False)
ex2 = CriticalSection().execute(
  accumulate, 1.0, exclusive_resource_access=False)
bad_sum = ex1 + ex2
sess.run(v.initializer)
sess.run(bad_sum)  # May return 0.0

Методы

execute

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

execute(
    fn, exclusive_resource_access=True, name=None
)

Выполняет функцию fn() внутри критической секции.

Функция fn не должна принимать аргументы. Чтобы передать дополнительные аргументы при вызове fn в критической секции, создайте лямбда-функцию:

critical_section.execute(lambda: fn(*my_args, **my_kwargs))