BVh% dZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ddl m Z dd l m Zdd lmZdd lmZdd lmZdd lmZddlmZdaej0aej4ZdZGddeZy)zIUtility for eagerly executing operations in parallel on multiple devices.N)_pywrap_parallel_device) device_util)context)composite_tensor) constant_op)ops)tensor) array_ops) variables)tpu_ops)nest)variable_utilsctj|jd}|$tdj |j|j |S)zz*ParallelDevice.__init__..CsLAK44Q7Ls!z/ParallelDevice requires at least one component.z{}/device:CUSTOM:{}N)tuple componentsrr_next_device_number_lockrhost_address_space_next_device_number_namerGetParallelDeviceCapsulesregister_custom_device _device_ids _device_scoper)selfr$ctxr device_infos r__init__zParallelDevice.__init__8sLLLDO ?? H II // C !)//0F0F0H0CEdjQ  2KK DOO%FK ""64::{CDD(,$**%s 3C44C=cL|D]U}t|tjtjt j fr=tdj|tj|j5tj|cdddS#1swYyxYw)z?Helper to pack plain-old-tensors, not structures or composites.zEvery component to pack onto the ParallelDevice must already be a tensor, got {}. Consider running `tf.constant` or `tf.convert_to_tensor` first on literal values.)inputsN) isinstance tensor_libTensorrCompositeTensorr Variablerrrrr(r tpu_replicated_input)r-tensorsr s r _pack_tensorzParallelDevice._pack_tensorTs   ..   ?VG_    DJJ :  ) ) 9:::s :BB#c|jt|t|jk7r7tdj t|t|jt j d5tj|}dddtj|jg|ddiS#1swY,xYw)a~Create a tensor on the parallel device from a sequence of tensors. Args: tensors: A list of tensors, one per device in `self.components`. The list can contain composite tensors and nests (lists, dicts, etc. supported by `tf.nest`) with the same structure for each device, but every component of nests must already be a `tf.Tensor` or composite. Passing `tf.Variable` objects reads their value, it does not share a mutable reference between the packed and unpacked forms. Returns: A tensor placed on the ParallelDevice. For nested structures, returns a single structure containing tensors placed on the ParallelDevice (same structure as each component of `tensors`). Raises: ValueError: If the length of `tensors` does not match the number of component devices, or if there are non-tensor inputs. zZCreating a parallel tensor requires one tensor per component. Got {} but was expecting {}.Nexpand_compositesT) _assert_eagerlenr$rrrrrconvert_variables_to_tensorsr map_structurer:)r-r9s rpackzParallelDevice.packgs*  7|s4??++  * 6#g,DOO 4 5 77 D E;;GDgE   d// 6' 604 66 EEs >B??Ccft|tjtjt j fstdj|tj|j5tj|t|jcdddS#1swYyxYw)z!Helper to unpack a single tensor.zExpected a tensor, got {}.) num_replicasN)r3r4r5rr6r r7rrrrr(r tpu_replicated_outputr>r$)r-parallel_tensors r_unpack_tensorzParallelDevice._unpack_tensors      , ,      3::?K LL DJJ >  * * DOO(<>>>>s 3*B''B0c |jtt|jDcgc]}g}}t j |j 5tj|}dddtj|dD]6}t||j|D]\}}|j|8|Dcgc]}tj||dc}Scc}w#1swYxYwcc}w)aUnpack a parallel tensor into its components. Args: parallel_tensor: A tensor, composite tensor, or `tf.nest` of such placed on the ParallelDevice. Passing `tf.Variable` objects reads their value, it does not share a mutable reference between the packed and unpacked forms. Returns: A list with the same length as `self.components` each with the same structure as `parallel_tensor`, containing component tensors. NT)r<)r=ranger>r$rrr(rr?r flattenziprFappendpack_sequence_as)r-rE_unpacked_componentsr accumulatorunpacked_tensorunpackeds rrzParallelDevice.unpacks ',S-A'BC!2CC DJJ &CC o,,$G,*- t226:+<, &+?+,, 0 1  ! !/848 : 11D 1s C/C4 D4C=c |jtj5g}t|jD]Z\}}tj |5|j tjtj|ddd\|j||_ddd|jS|jS#1swYxYw#1swY|jSxYw)a!A parallel tensor with scalar integers numbering component devices. Each device ID is placed on its corresponding device, in the same order as the `components` constructor argument. Returns: A parallel tensor containing 0 on the first device, 1 on the second, etc. N) r+r init_scope enumerater$rrKr identityrconstantrA)r-device_ids_listindexrs r device_idszParallelDevice.device_idss  >>  6 &t7 AME6zz&! A  " """;#7#7#>? A A A A  99_5 6   4   A A 6   s#3C8C  C C CC,cBtjs tdy)z$Verifies that tracing is not active.zParallelDevice is currently not supported inside `tf.function`. It can however run calls to a `tf.function` in parallel: with ParallelDevice() as p: f()N)rexecuting_eagerlyNotImplementedErrorr-s rr=zParallelDevice._assert_eagers%  $ $ &  / 00 'rc|j td|jtj|j |_|jj |S)zERuns ops in parallel, makes variables which save independent buffers.z;Re-entered a ParallelDevice scope without first exiting it.)r,AssertionErrorr=rrr( __enter__r]s rr`zParallelDevice.__enter__sX %  G IIDJJ/D  " KrcL|jj|||d|_yr)r,__exit__)r-typexctbs rrbzParallelDevice.__exit__s"S"-DrN)__name__ __module__ __qualname____doc__r0r:rArFrpropertyrYr=r`rbrrrr5sC7-8:& 6D >16  @0rr)ri threadingweakreftensorflow.pythonrtensorflow.python.distributertensorflow.python.eagerrtensorflow.python.frameworkrrrr r4tensorflow.python.opsr r tensorflow.python.tpu.opsr tensorflow.python.utilr rr'Lockr%WeakValueDictionaryrrobjectrrkrrrxsnP54+83+<++-'1)9>>+3335(sVsr