Thanks for your reply!
I’m currently testing only using cpu tensors. It seems that the data transfer between process dominates.
I use the following script to test bandwidth(I use slurm to run it):
from __future__ import division
from __future__ import print_function
import os
import torch
import torch.distributed as dist
import argparse
import numpy as np
import time

parser = argparse.ArgumentParser(description='PyTorch ImageNet Training')
parser.add_argument('--dist-backend', default='gloo', type=str,
                    help='distributed backend')

def run_allreduce(rank, size):
    data = torch.from_numpy(np.ones(size,dtype=np.float32))
    t0 = time.time()
    n = 10
    for i in range(n):
        dist.all_reduce(data, op=dist.reduce_op.SUM)
    t1 = time.time()
    print('average time:', (t1-t0)/n)

def run_sendrecv(rank, size):
    assert(dist.get_world_size() == 2)
    data = torch.from_numpy(np.zeros(size,dtype=np.float32))
    t0 = time.time()
    n = 10
    for i in range(n):
        if rank == 0:
            dist.send(data, 1)
        else:
            dist.recv(data, 0)
    t1 = time.time()
    print('average time:', (t1-t0)/n, 'BW:', size*4/(t1-t0)*n/1024/1024/1024,'GB/s')

args = parser.parse_args()

proc_id = int(os.environ['SLURM_PROCID'])
ntasks = int(os.environ['SLURM_NTASKS'])
node_list = os.environ['SLURM_NODELIST']

if '[' in node_list:
    node_list = node_list.split(',')[0].replace('[', '')
addr = node_list[8:].replace('-', '.')
addr = 'tcp://'+addr+':23456'
print(addr)

dist.init_process_group(backend=args.dist_backend, init_method=addr, world_size=ntasks, rank=proc_id)

run_allreduce(proc_id, 1024*1024*25)
#run_sendrecv(proc_id, 1024*1024*25)

The training scrip is modified from official imagenet example, since its a bit longer, I just show the concept. The DataParallel or DistributedDataParallel code are removed, instead i use following function after loss.backward to average gradients:
def average_gradients(model):
    """ Gradient averaging. """
    size = float(dist.get_world_size())
    for param in model.parameters():
        new_grad = param.grad.data.cpu()
        dist.all_reduce(new_grad, op=dist.reduce_op.SUM)
        param.grad.data = new_grad.cuda()
        param.grad.data /= size

As you pointed out, there are so many memory copies here.
It seems that there are much more works did in DistributedDataParallel, I will take a closer look.