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 будут атомарными.

ПРИМЕЧАНИЯ

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

При использовании нескольких критических секций для одних и тех же ресурсов нет гарантии эксклюзивного доступа к этим ресурсам. Это поведение по умолчанию запрещено (но см. аргумент 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))