Q/j ddlmZddlmZmZmZddlmZddlm Z ddgZ GddeZ Gdd e Z ed Gd de Z y ) )IntEnum)DictSizedTuple)functional_datapipe) IterDataPipeSHARDING_PRIORITIESShardingFilterIterDataPipeceZdZdZdZdZy)r N)__name__ __module__ __qualname__DEFAULT DISTRIBUTEDMULTIPROCESSINGk/root/repos/pocs/video-evals/.venv/lib/python3.12/site-packages/torch/utils/data/datapipes/iter/sharding.pyr r sGKOrc eZdZdededefdZy)_ShardingIterDataPipenum_of_instances instance_idsharding_groupctN)NotImplementedErrorselfrrrs rapply_shardingz$_ShardingIterDataPipe.apply_shardings "!rN)rrrintr r"rrrrrs#""", "rrsharding_filtercNeZdZdZd defdZejfdZdZ dZ dZ y) r ao Wrapper that allows DataPipe to be sharded (functional name: ``sharding_filter``). After ``apply_sharding`` is called, each instance of the DataPipe (on different workers) will have every `n`-th element of the original DataPipe, where `n` equals to the number of instances. Args: source_datapipe: Iterable DataPipe that will be sharded Nsource_datapipecj||_||_i|_d|_d|_|j yNr r)r&sharding_group_filtergroupsrr_update_num_of_instances)r!r&r)s r__init__z#ShardingFilterIterDataPipe.__init__+s5.%:"24 ! %%'rc`||k\rtd|d|d|tjk(r|j d|_d|_|D]<\}}|xj|j |zz c_|xj |zc_>yr()sortedr*keysr)appendreverserr)r!sorted_sharding_groupskeygroup_num_of_instancesgroup_instance_ids rr+z3ShardingFilterIterDataPipe._update_num_of_instancesGs!#$++**,- @C))1SD? @ &&( !9O < 5 "$5    5 58I I I   ! !%; ; ! rKsc%%E<   ' "L"&'DU!6DU(DUr