Repository: pytorch
Issue: Distributed Package asynchronous send/receive not working as expected (Gloo)
Labels: oncall: distributed,triaged

Hi :)

## 🐛 Bug

I am currently writing a distributed asynchronous program for which Pytorch provides the [methods](https://pytorch.org/docs/stable/distributed.html) `dist.irecv` and `dist.isend`. They work fine if you couple them with the function `req.wait()` after sending and receiving. However, doing that obviously turns the asynchronous call into synchronous one again because `wait` will block. Instead, I'd like to use `is_completed()` and "occasionally" check whether sending/receiving took place and if not, I would like to use the "free time" for something else. 
Now, it seems like that you *need* to couple each `dist.irecv` and `dist.isend` with a `req.wait()` call in order to make the communication happen in the first place. I'd have expected that you don't need to do that.

I would be interested in whether this is expected behavior or, in fact, a bug. If it isn't a bug, how could I avoid `req.wait()` then? I tried using Threads but that made everything much more complicated and I couldn't derive a solution for me, yet.

## To Reproduce

I created this dummy program to illustrate my problem.

```python
import os
import torch
import torch.distributed as dist
from torch.multiprocessing import Process
import time

def run_worker(rank, size):
    tensor = torch.zeros(1)
    req = None
    tensor += 1
    time.sleep(10)
    req = dist.isend(tensor=tensor, dst=0)
    print(f'Rank {rank} started sending')
    req.wait()

def run_master_wait(rank, size):
    tensor = torch.zeros(1)
    req = None
    req = dist.irecv(tensor=tensor, src=1)
    print('Rank 0 started receiving')
    req.wait() # works
    print('Rank ', rank, ' has data ', tensor[0])

def run_master_loop(rank, size):
    tensor = torch.zeros(1)
    req = None
    req = dist.irecv(tensor=tensor, src=1)
    print('Rank 0 started receiving')
    for i in range(1, 101):
        time.sleep(1)
        print(f"{i}/100")
        if req.is_completed(): # does not work
            break
    print('Rank ', rank, ' has data ', tensor[0])

def init_process(rank, size, backend='gloo'):
    """ Initialize the distributed environment. """
    os.environ['MASTER_ADDR'] = '127.0.0.1'
    os.environ['MASTER_PORT'] = '29500'
    dist.init_process_group(backend, rank=rank, world_size=size)
    if rank == 0:
        # run_master_wait(rank, size) # works
        run_master_loop(rank, size) # does not work
    else:
        run_worker(rank, size)

if __name__ == "__main__":
    size = 2
    processes = []
    for rank in range(size):
        p = Process(target=init_process, args=(rank, size))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()
```

Notice that `run_master_wait` works while `run_master_loop` hangs in a loop, even though my worker started to send its tensor. 

```
→ python test.py # run_master_wait
Rank 0 started receiving
Rank 1 started sending
Rank  0  has data  tensor(1.)

→ python test.py # run_master_loop
Rank 0 started receiving
1/100
2/100
3/100
4/100
5/100
6/100
7/100
8/100
9/100
Rank 1 started sending
10/100
11/100
12/100
13/100
14/100
```

## Expected behavior

```
→ python test.py
Rank 0 started receiving
1/100
2/100
3/100
4/100
5/100
6/100
7/100
8/100
9/100
Rank 1 started sending
10/100
Rank  0  has data  tensor(1.)
```

## Environment

```
Collecting environment information...
PyTorch version: 1.3.1
Is debug build: No
CUDA used to build PyTorch: Could not collect

OS: Mac OSX 10.14.6
GCC version: Could not collect
CMake version: version 3.8.2

Python version: 3.7
Is CUDA available: No
CUDA runtime version: Could not collect
GPU models and configuration: Could not collect
Nvidia driver version: 1.1.0
cuDNN version: Could not collect

Versions of relevant libraries:
[pip3] numpy==1.17.0
[pip3] torch==1.3.1
[pip3] torchvision==0.4.2
[pip3] torchviz==0.0.1
[conda] blas                      1.0                         mkl
[conda] mkl                       2019.4                      233
[conda] mkl-service               2.3.0            py37h01d97ff_0    conda-forge
[conda] mkl_fft                   1.0.12           py37h5e564d8_0
[conda] mkl_random                1.0.2            py37h27c97d8_0
[conda] pytorch                   1.3.1                   py3.7_0    pytorch
[conda] torchvision               0.4.2                  py37_cpu    pytorch
[conda] torchviz                  0.0.1                    pypi_0    pypi
```

## Additional context

I use the Gloo backend. I'd like to avoid the MPI backend because I am not sure if it is available on my cluster. I am aware that I can only use `dist.irecv` and `dist.isend` on CPUs and not GPUs (unless it is supported by MPI) but that's fine for my use case.

Thanks!


cc @pietern @mrshenli @pritamdamania87 @zhaojuanmao @satgera @rohan-varma @gqchen @aazzolini @xush6528