FM / data

/usr/local/lib64/python3.6/site-packages/torch/utils/data/

/usr/local/lib64/python3.6/site-packages/torch/utils/data

Name	Size	Mode	Actions
communication/	-	0755	rm
datapipes/	-	0755	rm
_utils/	-	0755	rm
__pycache__/	-	0755	rm
backward_compatibility.py	254	0644	edit dl rm
dataloader.py	67104	0644	edit dl rm
dataloader_experimental.py	6995	0644	edit dl rm
dataset.py	15001	0644	edit dl rm
distributed.py	5689	0644	edit dl rm
graph.py	1115	0644	edit dl rm
sampler.py	10226	0644	edit dl rm
sharding.py	1079	0644	edit dl rm
_decorator.py	7699	0644	edit dl rm
_typing.py	14436	0644	edit dl rm
__init__.py	2070	0644	edit dl rm

Edit: /usr/local/lib64/python3.6/site-packages/torch/utils/data/sharding.py (1079B)

import torch.utils.data.graph

def apply_sharding(datapipe, num_of_instances, instance_id):
    graph = torch.utils.data.graph.traverse(datapipe)

def traverse_graph(graph):
        results = set()
        for datapipe, sub_graph in graph.items():
            results.add(datapipe)
            sub_items = traverse_graph(sub_graph)
            for item in sub_items:
                results.add(item)
        return results

all_pipes = traverse_graph(graph)
    already_applied_to = None
    for pipe in all_pipes:
        if hasattr(pipe, 'is_shardable'):
            if pipe.is_shardable():
                if hasattr(pipe, 'apply_sharding'):
                    if already_applied_to is not None:
                        raise RuntimeError('This implementation of sharding can be only applied once per instance of DataPipeline.',
                                           'Already applied to', already_applied_to, 'while trying to apply to', pipe)
                    pipe.apply_sharding(num_of_instances, instance_id)
                    already_applied_to = pipe

import torch.utils.data.graph def apply_sharding(datapipe, num_of_instances, instance_id): graph = torch.utils.data.graph.traverse(datapipe) def traverse_graph(graph): results = set() for datapipe, sub_graph in graph.items(): results.add(datapipe) sub_items = traverse_graph(sub_graph) for item in sub_items: results.add(item) return results all_pipes = traverse_graph(graph) already_applied_to = None for pipe in all_pipes: if hasattr(pipe, 'is_shardable'): if pipe.is_shardable(): if hasattr(pipe, 'apply_sharding'): if already_applied_to is not None: raise RuntimeError('This implementation of sharding can be only applied once per instance of DataPipeline.', 'Already applied to', already_applied_to, 'while trying to apply to', pipe) pipe.apply_sharding(num_of_instances, instance_id) already_applied_to = pipe