Hi, I'm facing this problem as I want to train a model that depending on the input might not use some parameters/layers. Maybe the following sample code will help. As expected without the changes in the pull request this code fails.

TypeError: _queue_reduction(): incompatible function arguments. The following argument types are supported:
    1. (process_group: torch.distributed.ProcessGroup, grads_batch: List[List[at::Tensor]], devices: List[int]) -> Tuple[torch.distributed.Work, at::Tensor]

Invoked with: <torch.distributed.ProcessGroupNCCL object at 0x7fa6dfd804c8>, [[None, None, tensor([[-6.3432e+11, -2.8616e+12, -1.3041e+12],
        [-3.8208e+12, -1.7236e+13, -7.8549e+12],
        [-3.4752e+12, -1.5678e+13, -7.1446e+12],
        [-2.8048e+12, -1.2653e+13, -5.7663e+12]], device='cuda:0'), tensor([ -387039.5000, -2331274.5000, -2120455.7500, -1711401.0000],
       device='cuda:0')]], [0]

However even with them applied:

terminate called after throwing an instance of 'gloo::EnforceNotMet'
  what():  [enforce fail at /opt/conda/conda-bld/pytorch-nightly_1541790704598/work/third_party/gloo/gloo/transport/tcp/pair.cc:456] op.preamble.length <= op.nbytes. 20 vs 16

Sample code:

import os
import time

import torch
import torch.distributed as dist
import torch.nn.functional as F
from torch.multiprocessing import Process
from torch.utils.data import DataLoader

from torch.nn.parallel import DistributedDataParallel
from torch.utils.data.distributed import DistributedSampler



class CustomDataset(torch.utils.data.Dataset):
    def __init__(self, data_list):
        self.data_list = data_list

    def __len__(self):
        return len(self.data_list)

    def __getitem__(self, item):
        return self.data_list[item]


class Model(torch.nn.Module):
    def __init__(self):
        super(Model, self).__init__()
        self.some_used_lin = torch.nn.Linear(1, 2)
        self.lin = torch.nn.Linear(3, 4)

    def forward(self, input):
        input = input.squeeze()
        if input.size(0) == 3:
            return self.lin(input)
        else:
            return torch.cat((self.lin(input[:3]), self.some_used_lin(input[3].unsqueeze(0))))


def get_dataset():
    data_list = [{"x": torch.FloatTensor(3 + (i % 2)).random_(), "y": torch.FloatTensor(4 + (i % 2)*2).random_()}
                 for i in range(10)]
    return CustomDataset(data_list)


def train(rank, size, device_name):
    dataset = get_dataset()
    distributed_sampler = DistributedSampler(dataset, size, rank)
    train_loader = DataLoader(dataset, pin_memory=True, sampler=distributed_sampler)
    device = torch.device(device_name)
    model = Model()
    model.to(device)
    model = DistributedDataParallel(model, device_ids=[device])
    optimizer = torch.optim.Adam(model.parameters())

    for epoch in range(2):
        model.train()
        for batch_num, batch in enumerate(train_loader):
            print("rank", rank, batch['x'].shape)
            optimizer.zero_grad()
            pred = model(batch['x'].to(device))
            exp = batch['y'].to(device).squeeze()
            loss = F.binary_cross_entropy_with_logits(exp, pred)
            loss.backward()
            del loss

            optimizer.step()


def init_processes(rank, size, fn, ddp_args, device):
    """ Initialize the distributed environment. """
    os.environ['MASTER_ADDR'] = ddp_args['master_addr']
    os.environ['MASTER_PORT'] = ddp_args['master_port']
    os.environ['WORLD_SIZE'] = str(size)
    os.environ['RANK'] = str(rank)
    dist.init_process_group(ddp_args['backend'], rank=rank, world_size=size)
    fn(rank, size, device)
    print("done")


def main():
    ddp_args = {"master_addr": "127.0.0.1", "master_port": "29500", "backend": "gloo"}
    devices = ["cuda:0", "cuda:0"]
    size = len(devices)
    processes = []
    for rank, device in enumerate(devices):
        p = Process(target=init_processes, args=(rank, size, train, ddp_args, device))
        p.start()
        processes.append(p)

    while not any(p.exitcode is not None for p in processes):  # all worker processes have finished
        time.sleep(0.2)

    for p in processes:
        print("join")
        p.join()
        print("join done")
    print("all done")


if __name__ == "__main__":
    main()

Env:

PyTorch version: 1.0.0.dev20181109
Is debug build: No
CUDA used to build PyTorch: 9.2.148

OS: Arch Linux
GCC version: (GCC) 8.2.1 20180831
CMake version: version 3.12.4

Python version: 3.7
Is CUDA available: Yes
CUDA runtime version: 10.0.130
GPU models and configuration: GPU 0: GeForce GTX 1080 Ti
Nvidia driver version: 410.66

Getting DDP support models that has part of the parameters/layers unused #13273

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions