Repository: pytorch
Issue: RuntimeError: Operation not supported  
Labels: oncall: distributed,module: c10d,Stale

## 🐛 Bug

<!-- A clear and concise description of what the bug is. -->
I ran my code on LibOS, and got the error blow.

raceback (most recent call last):
  File "data_distributed_parallel_trainer.py", line 151, in <module>
    run(args.world_size, args.rank, args.batch_size, args.max_epochs, args.scale)
  File "data_distributed_parallel_trainer.py", line 54, in run
    dist.init_process_group(
  File "/usr/local/lib/python3.8/dist-packages/torch/distributed/distributed_c10d.py", line 534, in init_process_group
    store, rank, world_size = next(rendezvous_iterator)
  File "/usr/local/lib/python3.8/dist-packages/torch/distributed/rendezvous.py", line 184, in _tcp_rendezvous_handler
    store = _create_c10d_store(result.hostname, result.port, rank, world_size, timeout)
  File "/usr/local/lib/python3.8/dist-packages/torch/distributed/rendezvous.py", line 158, in _create_c10d_store
    return TCPStore(
RuntimeError: Operation not supported
[P1:T1:python3.8] debug: ---- shim_exit_group (returning 1)
[P1:T1:python3.8] debug: clearing POSIX locks for pid 1
[P1:T1:python3.8] debug: sync client shutdown: closing handles
[P1:T1:python3.8] debug: sync client shutdown: waiting for confirmation
[P1:T1:python3.8] debug: sync client shutdown: finished
[P1:shim] debug: IPC worker: exiting worker thread
[P1:T1:python3.8] debug: process 1 exited with status 1


## To Reproduce

Steps to reproduce the behavior:
Run the code blow on LibOS.

<!-- If you have a code sample, error messages, stack traces, please provide it here as well -->
`import os
import argparse
import torch
import torch.optim as optim
from tqdm import tqdm
import torch.nn.functional as F
import torch.nn as nn
from time import time_ns

import torch.backends.cudnn as cudnn
import torch.distributed as dist
import torch.multiprocessing as mp

from densenet import densenet121, SimpleNet
from dataset import load_dataset
from state import save_checkpoint, load_checkpoint


def train(model, optimizer, train_loader):
    model.train()
    i=0
    for (x, y) in train_loader:
        #x = x.cuda(non_blocking=True)
        #y = y.cuda(non_blocking=True)
        optimizer.zero_grad()

        out = model(x)
        out = F.log_softmax(out, dim=1)
        loss = F.nll_loss(out, y)

        loss.backward()
        optimizer.step()


def test(model, val_loader):
    model.eval()
    correct = 0.
    with tqdm(total=len(val_loader.dataset)) as progress_bar:
        with torch.no_grad():
            for (x, y) in val_loader:
                #x = x.cuda(non_blocking=True)
                #y = y.cuda(non_blocking=True)

                out = model(x)
                pred = F.log_softmax(out, dim=1).max(1)[1]

                correct += pred.eq(y).sum().item()
                progress_bar.update(x.size(0))
    accuracy = (100. * correct) / len(val_loader.dataset)
    return accuracy


def run(world_size, rank_id, batch_size, epoch, scale):
    dist.init_process_group(
        backend='gloo',
        init_method='tcp://' + os.environ['MASTER_ADDR'] + ":" + os.environ['MASTER_PORT'],
        world_size=world_size,
        rank=rank_id
    )
    #torch.cuda.set_device(device_id)

    train_set, val_set = load_dataset('CIFAR10', './data', scale)
    train_sampler = torch.utils.data.distributed.DistributedSampler(
        train_set,
        num_replicas=3,
        rank=rank_id,
        shuffle=True
    )
    train_loader = torch.utils.data.DataLoader(dataset=train_set, batch_size=batch_size,
                                               shuffle=False, num_workers=3, pin_memory=True, sampler=train_sampler)
    val_loader = torch.utils.data.DataLoader(dataset=val_set, batch_size=32, shuffle=False, num_workers=2, pin_memory=True)

    model = densenet121(pretrained=True, num_classes=10, memory_efficient=True)#.cuda(device_id)
    #model = SimpleNet()
    optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9, nesterov=True)
    model = nn.parallel.DistributedDataParallel(model)#, device_ids=[device_id])
    state = load_checkpoint('./checkpoints/CIFAR10.pt', model, optimizer)

    cudnn.benchmark = True

    start_epoch = state.epoch + 1
    for epoch in range(start_epoch, epoch):
        t0 = time_ns()
        print(f"Device {rank_id}")
        train(model, optimizer, train_loader)

        t1 = time_ns()
        delta = (t1 - t0) / (10 ** 9)
        print(f"Device {rank_id} - Train time: {delta} sec")

        if rank_id == 0:
            accuracy = test(model, val_loader)
            print(f"Accuracy: {accuracy}%")
            pass
        if epoch in [int(40 * 0.5), int(40 * 0.75)]:
            optimizer.param_groups[0]['lr'] /= 10.

        if epoch % 10 == 0 and rank_id == 0:
           save_checkpoint(state, './checkpoints/CIFAR10.pt')

        state.epoch = epoch


if __name__ == "__main__":
    parser = argparse.ArgumentParser('DDP training')

    parser.add_argument('--world_size',
                        help='how may nodes to be used',
                        type=int,
                        default=2)
    parser.add_argument('--rank',
                        help='current node',
                        type=int,
                        default=1)
    parser.add_argument('--batch_size',
                        help='total batch size',
                        type=int,
                        default=32)
    parser.add_argument('--max_epochs',
                        help='maximum number of training epoches.',
                        type=int,
                        default=5)

    parser.add_argument('-sc', '--scale',
                        help="scale the size of data_set",
                        default=0,
                        type=int,
                        )
    parser.add_argument('-ma','--master_addr',
                        help="host of the master node",
                        type=str,
                        default="localhost")
    parser.add_argument('-mp', '--master_port',
                        help='port of master node',
                        type=str,
                        default="8082")

    args = parser.parse_args()
    os.environ['MASTER_ADDR'] = args.master_addr
    os.environ['MASTER_PORT'] = args.master_port
    run(args.world_size, args.rank, args.batch_size, args.max_epochs, args.scale)`



cc @pietern @mrshenli @pritamdamania87 @zhaojuanmao @satgera @rohan-varma @gqchen @aazzolini @osalpekar @jiayisuse @SciPioneer @H-Huang