Repository: pytorch
Issue: Can't re-init rpc with a different rank after rpc shutdown
Labels: oncall: distributed,has workaround

## 🐛 Bug

It seems like some information about the rpc agent is retained after `rpc.shutdown`, making it impossible to later `rpc_init` with conflicting arguments. Specifically, `rpc_init` with a different `rank` argument fails.

## To Reproduce

Steps to reproduce the behavior:

1. init an rpc agent with a certain rank
1. shutdown that rpc agent
1. try to init rpc again with a different rank

<!-- If you have a code sample, error messages, stack traces, please provide it here as well -->

#### Reproduction Code
```python
import os
from torch.distributed import rpc

os.environ['MASTER_ADDR'] = 'localhost'
os.environ['MASTER_PORT'] = '29500'

rpc.init_rpc('agent', rank=0, world_size=1)
rpc.shutdown()
rpc.init_rpc('agent', rank=1, world_size=2)
```
#### Output
```
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "path/to/env/lib/python3.8/site-packages/torch/distributed/rpc/__init__.py", line 190, in init_rpc
    dist_autograd._init(rank)
RuntimeError: Container is already initialized with worker_id: 0, cannot initialize with different worker_id: 1
```

## Expected behavior

Should be able re-init rpc with different arguments after `rpc.shutdown`

## Environment

```
PyTorch version: 1.9.0.dev20210525
Is debug build: False
CUDA used to build PyTorch: 11.1
ROCM used to build PyTorch: N/A

OS: Ubuntu 18.04.5 LTS (x86_64)
GCC version: (Ubuntu 7.5.0-3ubuntu1~18.04) 7.5.0
Clang version: Could not collect
CMake version: version 3.10.2

Python version: 3.8 (64-bit runtime)
Is CUDA available: True
CUDA runtime version: 9.1.85
GPU models and configuration:
GPU 0: GeForce GTX 1080 Ti
GPU 1: GeForce GTX 1080 Ti

Nvidia driver version: 450.119.03
cuDNN version: Could not collect
HIP runtime version: N/A
MIOpen runtime version: N/A

Versions of relevant libraries:
[pip3] numpy==1.20.2
[pip3] torch==1.9.0.dev20210525
[pip3] torchaudio==0.9.0a0+bfd1d33
[pip3] torchvision==0.10.0.dev20210525
[conda] blas                      1.0                         mkl
[conda] cudatoolkit               11.1.74              h6bb024c_0    nvidia
[conda] mkl                       2021.2.0           h06a4308_296
[conda] mkl-service               2.3.0            py38h27cfd23_1
[conda] mkl_fft                   1.3.0            py38h42c9631_2
[conda] mkl_random                1.2.1            py38ha9443f7_2
[conda] numpy                     1.20.2           py38h2d18471_0
[conda] numpy-base                1.20.2           py38hfae3a4d_0
[conda] pytorch                   1.9.0.dev20210525 py3.8_cuda11.1_cudnn8.0.5_0    pytorch-nightly
[conda] torchaudio                0.9.0.dev20210525            py38    pytorch-nightly
[conda] torchvision               0.10.0.dev20210525      py38_cu111    pytorch-nightly
```

Can also reproduce on stable pytorch version 1.8.1

## Additional context

<!-- Add any other context about the problem here. -->

I encountered this error when trying to write unit tests for my application. I tried to set it up such that before every test case a new process is forked and an rpc connection is established. Basically something like the following:

```python
import os
import unittest

from torch import multiprocessing as mp
from torch.distributed import rpc


def run_other():
    rpc.init_rpc('other', rank=1, world_size=2)
    rpc.shutdown()


class TestApplication(unittest.TestCase):
    def setUp(self):
        os.environ['MASTER_ADDR'] = 'localhost'
        os.environ['MASTER_PORT'] = '29500'
        ctx = mp.get_context('fork')
        self.other = ctx.Process(target=run_other)
        self.other.start()
        rpc.init_rpc('this', rank=0, world_size=2)

    def tearDown(self):
        rpc.shutdown()
        self.other.join()

    def test1(self):
        # this test runs as expected
        return

    def test2(self):
        # this test doesn't run, because the `setUp` method, which runs before it, fails with
        # `RuntimeError: Container is already initialized with worker_id: 0, cannot initialize with different worker_id: 1`
        # likely because the `other` process was forked after `rpc.shutdown` in `this` process,
        # and then tried to `rpc_init` with a different rank
        return
```

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