Repository: pytorch
Issue: [Multiprocess Training & Datapipe] the message is sent "bind: Address already in use" after executing several epochs. 
Labels: oncall: distributed

### 🐛 Describe the bug

There is a failure case：
```
import torch
import torch.distributed as dist
from torch.utils.data import DataLoader
import torch.utils.data.datapipes as dp

if __name__ == "__main__":
    batch_size = 4
    world_size = int(os.environ.get("WORLD_SIZE", 1))
    rank = int(os.environ.get("RANK", 0))
    dist.init_process_group(backend='nccl', init_method='env://', world_size=world_size, rank=rank)
 
    idp = dp.iter.IterableWrapper(range(300)).map(lambda x: x + 1).shuffle()
    dataloader = DataLoader(idp, num_workers=2, shuffle=True, batch_size=4)
 
    for ite in range(1000):
        for itd in dataloader:
            print(f"rank {rank}, epoch {ite}, {itd.shape}")
```

Run command,

```
python -m torch.distributed.run --nproc_per_node 8 --master_port 2326 test.py
```

Error Info,
```
rank 13, epoch 160, torch.Size([4])
rank 13, epoch 160, torch.Size([4])
rank 13, epoch 160, torch.Size([4])
rank 13, epoch 160, torch.Size([4])
rank 13, epoch 160, torch.Size([4])
Traceback (most recent call last):
  File "/workspace/dataloader.py", line 29, in <module>
    for itd in dataloader:
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 435, in __iter__
    return self._get_iterator()
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 381, in _get_iterator
    return _MultiProcessingDataLoaderIter(self)
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 995, in __init__
    super(_MultiProcessingDataLoaderIter, self).__init__(loader)
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 567, in __init__
    self._pg = dist.new_group(backend="gloo")
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/distributed/distributed_c10d.py", line 3335, in new_group
    pg = _new_process_group_helper(
  File "/torch/venv3/pytorch/lib/python3.10/site-packages/torch/distributed/distributed_c10d.py", line 862, in _new_process_group_helper
    pg = ProcessGroupGloo(prefix_store, group_rank, group_size, timeout=timeout)
RuntimeError: [/torch/pytorch/third_party/gloo/gloo/transport/tcp/pair.cc:176] bind: Address already in use
```

My questions:
   1. The above code may be wrong. DataLoader does not have `world_size` and `rank` and cannot divide the data. If the above code is wrong, please tell me how to correct it.
   2. If the usage of DataPipe is correct, please tell me the reason for the error.
   3. why does _BaseDataLoaderIter need to [new process group](https://github.com/pytorch/pytorch/blob/faad8ecb26e710e883c3ae3d418ef7919f86ec37/torch/utils/data/dataloader.py#L571) for each batch  of DataLoader.
   
 


### Versions

I test the demo by torch1.13.1 and torch2.1.

cc @mrshenli @pritamdamania87 @zhaojuanmao @satgera @rohan-varma @gqchen @aazzolini @osalpekar @jiayisuse @H-Huang @kwen2501 @awgu @penguinwu @fegin @XilunWu @wanchaol @fduwjj @wz337 @tianyu-l @wconstab @yf225