o ?e@@s dZddlZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ddl m Z dd l m Z dd lmZdd lmZdd lmZd gZdZdZGdddeddZddZddZddZeZddZejddZed Gdd d Z dS)z,Critical Section object and execution logic.N)context)dtypes)ops) array_ops)control_flow_ops)gen_resource_variable_ops)tensor_array_ops)nest)object_identity) tf_exportCriticalSectionZcritical_sectionsZcritical_section_executionsc@seZdZdZdS)_ExecutionSignaturezGA class storing an `ExecuteInCriticalResource` op and associated attrs.N)__name__ __module__ __qualname____doc__rrk/home/www/facesmatcher.com/pyenv/lib/python3.10/site-packages/tensorflow/python/ops/critical_section_ops.pyr )sr ophandle resourcesexclusive_resource_accesscCsHt|tjr |St|tjrt|St r|durdSt |S)zEIdentity op that recognizes `TensorArray`, `Operation`, and `Tensor`.N) isinstancerZ TensorArrayidentityrZ Operationrgrouprexecuting_eagerlyr)xrrr _identity1s    rcCs|jpt|SN)device_get_colocationrrrr_get_device_or_colocation=sr#c Cs&z|dWSttfyYdSw)z&Get colocation symbol from op, if any._classN)get_attr ValueErrorAttributeErrorr"rrrr!As  r!cCs(ztjWStygt_tjYSwr)_CRITICAL_SECTION_STACKvaluer'rrrr_get_critical_section_stackLs   r*c cst}||vrtd|d||zdVW|}||kr,td|d|dS|}||kr@td|d|w)aPush a CriticalSection._signature to the thread-local stack. If the signature is already on the stack, raise an error because it means we're trying to execute inside the same locked CriticalSection, which will create a deadlock. Args: signature: Tuple of the type `CriticalSection._signature`. Uniquely identifies a CriticalSection by its `shared_name`, `container`, and device. Yields: An empty value. The context is guaranteed to run without deadlock. Raises: ValueError: If the signature is already on the stack. RuntimeError: If another thread or function modifies the current stack entry during the yield. z0Attempting to lock a CriticalSection (signature=zK) in which we are already running. This is illegal and may cause deadlocks.Nz8CriticalSection stack inconsistency: expected signature z but received )r*r&appendpop RuntimeError) signaturestackZreceived_signaturerrr_push_critical_section_stackTs8  r0c@sTeZdZdZ  dddZddZeddZdd d Zd d Z ddZ ddZ dS)r a# Critical section. A `CriticalSection` object is a resource in the graph which executes subgraphs in **serial** order. A common example of a subgraph one may wish to run exclusively is the one given by the following function: ```python 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) ``` Here, a snapshot of `v` is captured in `value`; and then `v` is updated. The snapshot value is returned. If multiple workers or threads all execute `count` in parallel, there is no guarantee that access to the variable `v` is atomic at any point within any thread's calculation of `count`. In fact, even implementing an atomic counter that guarantees that the user will see each value `0, 1, ...,` is currently impossible. The solution is to ensure any access to the underlying resource `v` is only processed through a critical section: ```python cs = CriticalSection() f1 = cs.execute(count) f2 = cs.execute(count) output = f1 + f2 session.run(output) ``` The functions `f1` and `f2` will be executed serially, and updates to `v` will be atomic. **NOTES** All resource objects, including the critical section and any captured variables of functions executed on that critical section, will be colocated to the same device (host and cpu/gpu). When using multiple critical sections on the same resources, there is no guarantee of exclusive access to those resources. This behavior is disallowed by default (but see the kwarg `exclusive_resource_access`). For example, running the same function in two separate critical sections will not ensure serial execution: ```python 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 ``` NcCsFt|r|durtd|d|d|rtd|||dS)zCreates a critical section.NzArguments critical_section_def=z and shared_name=z9 are mutually exclusive. Please only specify one of them.z1Argument `critical_section_def` is not supported.)rZensure_initializedr&_init_from_args)selfname shared_nameZcritical_section_defZ import_scoperrr__init__s   zCriticalSection.__init__c Cst|dgD}t0tj}|dur|}|durd}tj|||d|_||p.t|jt |jf|_ Wdn1s?wYWdn1sNwYt s_t t|dSdS)z:Initialize the CriticalSection from constructor arguments.r N)r4 containerr3)r name_scopeZ init_scopeget_default_graph _containerrZmutex_v2_handleidr# _signaturerradd_to_collectionsCRITICAL_SECTIONS)r2r3r4r7rrrr1s(   zCriticalSection._init_from_argscCs |jjjSr)r;rr3r2rrrr3s zCriticalSection.nameTc s|t|dg+tjktj}tsYt j 2t }t |g |}Wdn1s9wYt t |}Wdn1sSwYnt |g |}Wdn1smwYWdn1s|wYts||jtdd|D}tfdd|Drtdjd||d dt|D} t | /tj t|} Wdn1swYt|tt|}Wdn1swYt | gtt|} Wdn 1swYts*t|jjt ||d } t!t"| | WdS1s7wYdS) aExecute function `fn()` inside the critical section. `fn` should not accept any arguments. To add extra arguments to when calling `fn` in the critical section, create a lambda: ```python critical_section.execute(lambda: fn(*my_args, **my_kwargs)) ``` Args: fn: The function to execute. Must return at least one tensor. exclusive_resource_access: Whether the resources required by `fn` should be exclusive to this `CriticalSection`. Default: `True`. You may want to set this to `False` if you will be accessing a resource in read-only mode in two different CriticalSections. name: The name to use when creating the execute operation. Returns: The tensors returned from `fn()`. Raises: ValueError: If `fn` attempts to lock this `CriticalSection` in any nested or lazy way that may cause a deadlock. ValueError: If `exclusive_resource_access == True` and another `CriticalSection` has an execution requesting the same resources as `fn``. Note, even if `exclusive_resource_access` is `True`, if another execution in another `CriticalSection` was created without `exclusive_resource_access=True`, a `ValueError` will be raised. Zcritical_section_executeNcSs(g|]}|jD] }|jtjkr|qqSr)inputsZdtyperresource.0rinput_rrr !s z+CriticalSection.execute..c3s|]}|VqdSr)_is_self_handlerDrr@rr +z*CriticalSection.execute..zPAttempting to lock a CriticalSection in which we are already running (signature=z+). This is illegal and may cause deadlocks.cSsg|]}t|qSr)rrHrrrrF4sr)#rr8r0r=rZ mutex_lockr;rrr9_lockZget_operationsZcontrol_dependenciesset difference!_add_control_dependencies_to_lockrr ZObjectIdentitySetanyr&#_check_multiple_access_to_resourcesr flattenZ colocate_withZconsume_mutex_lockZpack_sequence_asrtupleZ map_structurerr listr>CRITICAL_SECTION_EXECUTIONS) r2fnrr3lockZ existing_opsr created_opscaptured_resourcesZr_flatZensure_lock_existsZoutputsr.rr@rexecutesr           &zCriticalSection.executecCstdd|D}|dd|Dtdd|D}|D] }||jdq|jD] }||jdq+|jD] }||jjdq8||jd|}|sRdSt j |}| |dS)z=To avoid deadlocks, all args must be executed before lock_op.cSsg|] }|jD]}|jqqSr)rArrCrrrrFTszECriticalSection._add_control_dependencies_to_lock..css |] }|jD]}|VqqdSr)control_inputs)rDrZinput_oprrrrIUs  zDCriticalSection._add_control_dependencies_to_lock..css|]}|j|fVqdSr)_id)rDrrrrrI\rJN) rLupdatedictr,r\r[rArvaluesrrZ_add_control_input)r2rXZlock_opall_argsZ all_args_dictrrErrrrNQs"    z1CriticalSection._add_control_dependencies_to_lockcCspt|tjr ||juS|jjdko7|jdo7|jd|jjdko7|jj|jjjkp7t|jt|jjkS)zs>             $