AVhOldZddlZddlZddlmZmZmZmZmZddl m Z ddl m Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddlmZddlmZddlm Z ddl!m"Z#ddl$m%Z%ddl&m'Z'ejPGddZ)GddejTZ+GddejXZ-dee j\deej^d e0fd!Z1 d1d"e j\d#ee2d$ee3fd%Z4d"e j\d&ej^d$ejjfd'Z6d(eee3e#jnfd)ed*e jpd+e3fd,Z9e'd-g.Gd/d0ejtZ;y)2aAPIs to deal with input datasets efficiently in DTensor. When using tf.data with DTensor, the `DTensorDataset` API can be used to efficiently handle loading the input data and correctly packing it to the corresponding devices. This API is intended to work with unbatched data and can be used for both data and model parallel setups. Example usage: >>> # 1-D mesh with 4 devices >>> mesh = dtensor.Mesh(dim_names=['batch'], ...) >>> layout = dtensor.Layout.batch_sharded(mesh, 'batch', rank=1) >>> dataset = tf.data.Dataset.range(256) >>> d_dataset = dtensor.DTensorDataset( ... dataset=dataset, ... global_batch_size=16, ... mesh=mesh, ... layouts=layout, ... batch_dim='batch') >>> d_iter = iter(d_dataset) >>> # Each batch is a length 16 tensor sharded across 4 devices >>> batch_0_dtensor = next(d_iter) >>> batch_0_dtensor >>> batch_1_dtensor = next(d_iter) >>> batch_1_dtensor For multi-client setups, `DTensorDataset` interacts with tf.data service to correctly distribute the dataset among the participating clients. DTensor works with tf.data service in co-located mode where each worker is running alongside the DTensor client (the Tensorflow Python process). The `TFDataServiceConfig` dataclass can be filled with information about the tf.data service cluster, and passed to `DTensorDataset` to enable distribution. N)AnyListOptionalSequenceTuple)api)configlayout)data_service_ops) distribute) dataset_ops) iterator_ops)context) constant_op)dtypes)errors)ops)tensor) tensor_shape) tensor_spec) array_ops)math_ops)data)nest) tf_exportc&eZdZUdZeed<eed<y)TFDataServiceConfiga0Specifies the tf.data service configuration to use. Attributes: dispatcher_address: a string specifying the address of the tf.data service dispatcher server. job_name: a non-empty string identifying the shared job that will be created on tf.data service to process this dataset. dispatcher_addressjob_nameN)__name__ __module__ __qualname____doc__str__annotations__T/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/dtensor/python/input_util.pyrrXs -r(rcteZdZdZdeej dejde ffd Z dZ e dZ xZS)_DTensorIteratoraAn iterator for a tf.data.Dataset distributed using DTensor. DTensorIterator encapsulates multiple underlying dataset iterators. It handles retrieving the tensors to be placed on each underlying device and then uses the 'pack' operation to create and return a DTensor. Thus users need only interact with a single DTensorIterator to automatically distribute dataset tensors onto devices. dtensor_componentsglobal_element_speclayoutsc|\|_||_||_tjd||_t |||y)aInitializes a distributed iterator for DTensor datasets. This iterator encapsulates tf.data iterators for the underlying devices, and treats it as a packed DTensor of iterator resource tensors. Args: dtensor_components: a tuple containing the underlying iterator resources packed into a DTensor. This is expected to be a tuple with a single element. global_element_spec: the underlying dataset's element spec from a global view. layouts: a structure of DTensor layouts to be applied to the elements returned by the underlying iterators. This can be a single layout or (possibly nested) tuples or dictionaries of layouts, and the structure must match the structure of the iterator elements. c"|jSN) to_stringr s r)z+_DTensorIterator.__init__..sv'')r() components element_specN)_iterator_resource_dtensor_global_element_spec_layoutsr map_structure _layouts_strsuper__init__)selfr,r-r. __class__s r)r<z_DTensorIterator.__init__qsR,);%T$ 3DDM**)74D G%4GIr(cF |j}tjtjt j ||j}tj|S#tj$r"}tjrt||d}~wwxYwr1) _next_internalr async_waitrr9r copy_to_meshr8rOutOfRangeErrorexecuting_eagerly StopIteration)r= host_elem device_elemes r)__next__z_DTensorIterator.__next__s%%'i &&  It}}6k    ! !  " " $" sA(A++B >BB cBt|j|jSr1)_DTensorIteratorSpecr7r:r=s r) _type_specz_DTensorIterator._type_specs  9 94;L;L MMr()r!r"r#r$rrTensorr TensorSpecrr<rIpropertyrM __classcell__r>s@r)r+r+gsWI .I'11I I>& N Nr(r+ceZdZdZddgZdej deffd Ze dZ dZ e d Z d Z d Zed ZxZS) rKz*Type specification for `_DTensorIterator`.r7r:r- layouts_strc@t||||_||_yr1)r;r<r7r:)r=r-rTr>s r)r<z_DTensorIteratorSpec.__init__s! G() 3D#Dr(ctSr1)r+rLs r) value_typez_DTensorIteratorSpec.value_types r(c2|j|jfSr1r7r:rLs r) _serializez_DTensorIteratorSpec._serializes  % %t'8'8 99r(cLtjgtjfSr1)rrOrresourcerLs r)_component_specsz%_DTensorIteratorSpec._component_specss  " "2v 7 99r(c|jfSr1)r6)r=values r)_to_componentsz#_DTensorIteratorSpec._to_componentss  , , ..r(ctjtjj|j }t ||j|S)Nr,r-r.)rr9 layout_libLayout from_stringr:r+r7)r=r4r.s r)_from_componentsz%_DTensorIteratorSpec._from_componentssE  %%t'8'8:G % 55 r(c<||j|jSr1rY)clsr_s r) from_valuez_DTensorIteratorSpec.from_values u))5+=+= >>r()r!r"r#r$ __slots__rrOrr<rPrWrZr]r`rf classmethodrirQrRs@r)rKrKsv2%~6)$!,!7!7$FI$   : : :/??r(rKflattened_layoutsflattened_elem_specdataset_already_batchedc|s td|dj}t||D]\}}|jjtd|z|jj}|s|dz }|j|k7rtd|||jfz|st|j d}|td|z|jj d|k7std||fzy) aChecks that the dataset's layouts and element specs are compatible. Args: flattened_layouts: the flattened list of layouts used to distribute the dataset. flattened_elem_spec: the flattened list of element specs used in the dataset's components. dataset_already_batched: whether the dataset to be validated is already batched. Raises: ValueError: if the dataset's inputs are incompatible. z?Expected input element spec of at least one element, was empty.rNz:Dataset element shape must have a valid rank, got spec %s.zExpected layout with rank %d for element spec %s, got layout %s. Check that the dataset is not batched before passing to DTensorDataset.ziSize of batch dimension of element spec %s is None. Ensure drop_remainder=True when batching the dataset.zKSize of batch dimension of element spec %s does not match expected size %d.) ValueErrorshapezipranksharding_specsas_list)rlrmrnfirst_elem_shaper elem_spec expected_rankbatch_dim_sizes r)_validate_inputr{s?  I KK)+1102EFAfi#  F   OO((M "qm {{m#   )V%:%: ; < == '//1!4n  >AJ KL L  "1 % 7!%.$? @A A9Ar(r batch_dimreturncg}|jD]R}||tjfvr|jd)|j|jj |T|S)aComputes a list of the number of shards in each dimension of the layout. The shard counts are used to slice each dataset element. The batch dimension's count is overridden to 1 since we only consider how many shards to make locally (within each local replica). Sharding across clients is handled by either tf.data.Dataset's shard transformation (in the single-client case) or tf.data service's distribute function (in the multi-client case). Args: layout: the layout to compute the shard counts for. batch_dim: the name of the batch dimension of the layout, if present. Returns: A list of shard counts, one element per dimension of the layout. rp)rurc UNSHARDEDappendmeshdim_size)r r| shard_countsspecs r) _shard_countsrsd",##6d ://00!&++..t45 6 r(rxcg}|jjD]}dg}t|jddD]i\}}|tj k(s||k7r|j d0|j |j||jj|zk|j |tj|tjS)aComputes a utility matrix to derive device-based slice offsets. This function builds a matrix of shape `[mesh.rank, layout.rank]` for each dataset element. This matrix can be used to slice the DTensor components returned by the iterator according to the local device that component is to be placed on. This can be done by multiplying the device offsets of shape `[1, mesh.rank]` with this index matrix to get a `[1, layout.rank]` shape tensor containing the slice offsets. Note: the index on the batch dim is always 0 since sharding on the batch dimension is handled by either tf.data.Dataset's shard transformation (in the single-client case) or tf.data service's distribute function (in the multi-client case). If there is no sharding on the batch dimension (or any other dimension), the slice index remains 0. Args: layout: the layout of the dataset element. elem_spec: the spec of the dataset element. Returns: The index matrix as a tensor. rrpNdtype) r dim_names enumeraterurcrrrrrrconstantrint32)r rxmatrixdimrow layout_idxrs r) _index_matrixrs0 & [[ " "c #C%f&;&;AB&?@M D %% % 1  9??:.&++2F2Fs2KKL M  MM#   fFLL 99r(datasetsr.rnum_local_devices_per_replicacjj}d}g}|D]h\}}t|D]U} tj||5|j || } |j t| ddd|dz }Wj|t|k7rtd|dt|dtjfd|} |D cgc]} | j} } tj| tj j#jd }tj$j'|tj(| |S#1swYxYwcc} w) aCreates a DTensor iterator resource for the per-replica datasets. Given a list of replica ID to tf.data.Dataset mappings, this function creates iterators for each device and then packs the underlying iterator resource tensors into a single DTensor. This resource tensor is used by the IteratorGetNext op to retrieve the next element in the dataset. Args: datasets: a list of tuples of each unique local replica ID to the dataset object whose elements will be placed on the devices corresponding to that replica. layouts: a structure of DTensor layouts to be applied to the elements returned by the underlying iterators. This can be a single layout or (possibly nested) tuples or dictionaries of layouts, and the structure must match the structure of the iterator elements. mesh: the DTensor mesh to place the iterator batches on. num_local_devices_per_replica: the number of devices in each data-parallel replica. Returns: A DTensor of the underlying iterator resource tensors. r) num_shardsindexNrpzWThe `datasets` argument does not have the correct number of underlying datasets, found z but expected .c`tj|jjSr1)rcrdru host_mesh)lrs r)r3z1_pack_iterator_resource_dtensor..rs! !!!"2"2DNN4DEr()rrt)r local_devicesranger device_v2shardriterlenrqrr9_iterator_resourcerpackrcrd replicated_dtensor_deviceset_iterator_element_layoutsflatten)rr.rrhost_mesh_devices device_idx iterators_datasetidxdevice_dataset host_layoutsititerator_resourcesd_iterator_resources ` r)_pack_iterator_resource_dtensorrDs6nn&446*)ja23 ==*:6 7/ 4C'An-./Aoj 3())  ''1l3 ! " #1 & '' ##EwP, 9BB"--BB""(8q"AC444<< 57 1//"Cs.E(E4(E1 z#experimental.dtensor.DTensorDataset)v1ceZdZdZddddddej dejdede d e d e e d e e d e e ffd ZdZdZdZedZxZS)DTensorDatasetzA dataset of DTensors. DTensorDataset encapsulates a `tf.data.Dataset` whose elements are automatically packed and returned as DTensors based on a given mesh and layouts. FN)rnr|prefetchtf_data_service_configrrr.global_batch_sizernr|rrc t||tj|| t d||_||_||_||_||_ tj|j|tj|} tj|j} |r|j||_t!t"j%|j'D cgc]} | | c} |_| D]%} || j*dk7st-d|d| nd|_dg|_t/| | |||jz} |s|j1| d |_n<| dj4j7d}|| k7rt-d || fz||_g}t9j:|g}tj|j2jD]`}t=j>tAjB||j4dd|jD|jF }|jI|btjJ|j||_&tOjP|jS|jz}tU|j(|_+|jY|jVz|_-||jZz|_.tOj^|j\z|jZz|_0| D cgc]} tc| |c} |_2tg| | D cgc]\} }ti| |c}} |_5ycc} wcc} wcc}} w) ai Creates a DTensorDataset. DTensorDataset automatically handles distribution of the dataset elements to each client's devices. It can be used to create an iterator that returns DTensors of the input data on each iteration. DTensorDataset works best with unbatched datasets. It takes the mesh and the provided layouts to automatically calculate how to batch the input locally for each replica. If the provided dataset is already batched according to the per-replica batch size, then `dataset_already_batched` must be set and DTensorDataset will check that the batch size is consistent with the intended `global_batch_size` using the layout information. Each replica receives a separate slice of the global batch, thus the per-replica batch size can be computed as the global batch size divided by the number of model replicas. For a DTensor mesh, the number of replicas is equal to the size of the mesh's batch dimension. Note: `tf.experimental.dtensor.DTensorDataset` instances do *not* implement the full interface of `tf.data.Dataset`. It only supports two usages we will mention below: iteration and `element_spec`. We don't support any other APIs to transform or inspect the dataset. TODO(b/223275517): add support for input datasets that are already batched to the global batch size. Args: dataset: a `tf.data.Dataset` object. mesh: the DTensor mesh to place the dataset batches on. layouts: a structure of DTensor layouts to be applied to the input dataset values. This can be a single layout or (possibly nested) tuples or dictionaries of layouts, and the structure must match the structure of the dataset. Either all or none of the layouts should be sharded on the batch dimension; having only a subset of layouts batch sharded will not work and raises a ValueError. global_batch_size: the desired global batch size. dataset_already_batched: must be set only if the dataset is already batched to the per-replica batch size. The batched dataset must have `drop_remainder=True` set since DTensor requires static shapes for slicing the input tensors. batch_dim: the mesh dimension on which the input's batch dimension is sharded. Set to None if the input layouts do not shard on the batch dimension. prefetch: number of batches to prefetch using Dataset.prefetch. tf_data_service_config: if operating in multi-client mode, this config specifies the tf.data service configuration to use. Raises: ValueError: on any of the following situations, 1. if the structures and ranks of layouts and the dataset do not match. 2. if the shapes in the dataset's spec are not fully defined. 3. if batch_dim is specified and all layouts are not batch-sharded. 4. if per_replica_batch_size is specified for an already batched Dataset but it does not match the expected per-replica size based on the provided mesh. TypeError: if type of structures of layouts and the dataset do not match. NzJMulti-client DTensorDataset is currently not supported. Check b/271162918.rz batch_dim z; was specified but at least one layout did not contain it: rp)rnT)drop_remainderz`per_replica_batch_size does not matched expected size based on the mesh, got %d but expected %d.)rrrname)6r;r<r to_variantNotImplementedError_meshr8 _batch_dim _prefetch_tf_data_service_configrassert_same_structurer5rrnum_global_replicaslistdictfromkeyslocal_device_locations_local_replica_idsrurqr{batch_batched_datasetrrrvras_shaperrOoperatorconcatrrrpack_sequence_asr7r num_global_devices device_typer_num_local_replicasnum_local_devices_num_local_devices_per_replica_num_clients_per_replica client_id_partition_offsetr_all_shard_countsrsr_index_matrices)r=rrr.rrnr|rrrlrmlocr expected_batch_sizeper_replica_batch_sizeflattened_global_elem_specbatch_tensor_shaperx new_elem_specnum_global_devices_per_replicar>s r)r<zDTensorDataset.__init__sH GWk44W=>)   !!DJDMDODN#9D w33W= W-,,w';';<!%y!9d $ --)-)D)D)FG#s9~G I!Jd&8& --a0 0$-v78 88"#d!"d 79 ,t/G/GG "%mm d,4d 315;;CCEaH #6 61 #%8 9 :; ;&d"$%..0A/BC\\$"7"7"D"DE7 !,, 2IOOAB4GH~~m!'' 6 7 !% 5 58!:D&,%>%> &#77&8""4#:#:;D*.*@*@+ ! !+"D' '$*M*MM !%..043P3PP!%!D!DED 8I-3 fi(D "%%68K!L FI fi(DAHzs' M; 1NNc .g}|j}|ja|jdkDr7|j||j}tj j }n5tj j}ntj j}|jg|jt j||jj|jjdtjd}t!|j"D]\}}tj$||j&||j(}|j||j*}|j-|}|j.(|j1|j.|j*z}|j3||ft5||j6|j8|j*}t;|f|j<|j6S)NrprLOCAL)processing_modeservicer target_workers) num_workersr num_replicas)rr.rrrb)rrr _repeat_batchr ShardingPolicyDATAFILEOFFrapplyr rr r rrr_AutoShardDatasetrrr _partitionrrrrr8rr+r7)r=r local_datasetsharding_policylocal_replica_idx replica_idrrs r)__iter__zDTensorDataset.__iter__(s79H))M " & & ***=+/+H+HJ *99>> +99>>)77;;o ##/#))  % %-22EE66??@&BRBRBTAUV$  &'m*343J3J)K-%:,, ..!// 1g""7D,O,OPg(g  #"" NNT@@ @Booz7+,+-0: ZZ&*&I&I K /1 55  r(c>dk(r|Sfd}|j|S)Nrpc`tjj|jSr1)r DatasetV2 from_tensorsrepeat)xrepeatss r)rz,DTensorDataset._repeat_batch..repeatrs%  " " / / 2 9 9' BBr()flat_map)r=rrrs ` r)rzDTensorDataset._repeat_batchms(!| nC   F ##r(cjdk(rjdk(r|Sfd}|j}|j|}|S)z?Slices each dataset element on any sharded non-batch dimension.rprctj|}g}tj| jzt j }| jz } jj|}tj|d}t| j jD]\}}}tj||} tj| d} tj |t j |z} |j#tj$|| | tj&||S)Nr)rp)r)out_type)beginsize)rrrcastrrrrrcoordsrreshapersrrmatmulshape_v2rslicer) rrflattened_batchflattened_output norm_indexrelementr idx_matrixindexesstartrr=s r) slice_batchz.DTensorDataset._partition..slice_batch~s U+o== $55 5V\\KjD***jzz  ,f  1f/2?373I3I373G3G0I> +'<//&*5!!'51!! fll,/;< OOG5t < >> " "5*: ;;r()rrrmap)r=rrenumerated_datasetpartitioned_datasets` r)rzDTensorDataset._partitionwsP **a/D4J4Ja4O n<,!**,,00= r(c|jSr1)r7rLs r)r5zDTensorDataset.element_specs  $ $$r()r!r"r#r$ data_typesrrcMeshrintboolrr%rr<rrrrPr5rQrRs@r)rrs05*.)-GK]",,] __] ] #& ] )- ]#3-]"#](00C'D]~CJ$B % %r(rr1)r's,\77),:D>23+3..+.43+*6'6    >N|11>NB#?<44#?L4Ax 0A0A'B4A)1+2H2H)I4A-14Ap.2*++%c]6:3i4":*++":(33":8> ":J95j222349 9 //9$' 9x 0R8Y%[??Y%9Y%r(