Bringing together results on ddp on a single machine

[brucemuller]

I'd like to understand how to use ddp properly with multiple GPUs on a single machine as I'm unsure of how to bring results together using this method.

I'm using TensorBoard for logging

The problem seems to be that my code (below) runs on each of the three GPUs (with a third of the data each), but the variables like "overall_correct" only exist for each of the three processes so only a third of the data gets logged. For example, my overall performance on a single GPU is 82% but with the above process on 3 GPUs it is a third of that. I know this is a kindof silly thing but can someone explain how I should bring together the required validation/training statistics from the sub-processes using pytorch lightning?

My process is roughly:

model = MyModel(hparams)
tt_logger = TestTubeLogger(save_dir="path",name=expname)
trainer = Trainer(logger = tt_logger , gpus=3, distributed_backend='ddp' )
trainer.fit(model)

class MyModel(LightningModule):

     def __init__(self, hparams):
          super(MyModel, self).__init__() 
          self.hparams = hparams
          self.resnet = ResNetEncoder(self.hparams)
          self.loss_meter_training = averageMeter()
          self.overall_correct = 0.0

    def training_step(self, batch, batch_i):   
        ...
        self.loss_meter_training.update(float(total_loss))
        return {'loss': total_loss}

    def validation_step(self, batch, batch_nb):
        ...
        if something:
            self.overall_correct += 0.0
        return {'val_loss': total_loss}

    def validation_end(self, outputs):

        self.logger.experiment.add_scalar('epoch losses/training/total', self.loss_meter_training.avg, self.epoch_nb)
        self.logger.experiment.add_scalar('metrics/validation performance', self.overall_correct/20000, self.epoch_nb)
        self.loss_meter_validation.reset() 
        self.overall_correct = 0.0

    def configure_optimizers(self):
        optimizer = optim.Adam(self.parameters(), lr=self.hparams.lr)
        return optimizer 

    @pl.data_loader
    def tng_dataloader(self):

        t_loader = PairLoader('dummy', self.hparams , split='training')
        dist_sampler = torch.utils.data.distributed.DistributedSampler(t_loader)
        trainloader = data.DataLoader(t_loader,batch_size=self.hparams.batch_size, sampler=dist_sampler, num_workers = 12)  
    
        return trainloader
    
    @pl.data_loader
    def val_dataloader(self):
        
        v_loader = PairLoader('dummy', self.hparams , split='validation')
        dist_sampler = torch.utils.data.distributed.DistributedSampler(v_loader)
        trainloader = data.DataLoader(v_loader,batch_size=self.hparams.batch_size, sampler=dist_sampler, num_workers = 12) 
        
        return trainloader

[matthew-z]

I don't know if pytorch-lightning has done anything special about it.

With pure PyTorch, you may use dist.all_gather to sync the validation score among workers.

For example, if you have 2 workers and each of them evaluated 2 examples, then you can use dist.all_gather to get the 4 scores and then compute the mean validation score.

[brucemuller]

@matthew-z thanks for your reply. That seems promising. I have also found this issue requesting similar behaviour: #243
I'm not sure how all_gather works yet. Should it be like:

    def validation_end(self, outputs):

        self.logger.experiment.add_scalar('epoch losses/training/total', self.loss_meter_training.avg, self.epoch_nb)
        self.logger.experiment.add_scalar('metrics/validation performance', self.overall_correct/20000, self.epoch_nb)
        self.loss_meter_validation.reset() 
        
        dist.all_gather(output_list, self.overall_correct)
        self.overall_correct = 0.0

If so, where would we put output_list when using pytorch lightning? Thanks for your help! I think it is something many new users will be confused about when it comes to ddp use

[matthew-z]

I didn't try it with PL, but in general you may try to sync data as follow with DDP:

def gather_list_and_concat(list_of_nums):
    tensor = torch.Tensor(list_of_nums).cuda()
    gather_t = [torch.ones_like(tensor) for _ in range(dist.get_world_size())]
    dist.all_gather(gather_t, tensor)
    return torch.cat(gather_t)

>>> res = gather_list_and_concat([1, 2])   # in node 1
>>> res = gather_list_and_concat([3, 4])  # in node 2
>>> res
torch.Tensor([1,2,3,4])

[sneiman]

Hi -
I have been working with the same issue - how to get the benefit of ddp with distributed sampling and still have accurate statistics, expecting to use reduce, or gather from pytorch distributed to pull things together. I did some investigation and found that this was not necessary. The key fact I found is that pl is not duplicating all the tensors in the model to each sub-process. They are in fact shared by all the sub-processes. I did not look at the source, but I expect only the parameters and buffers are being duped.

So no need to keep it in sync - they are all sharing the same tensor. You can create these in the model itself. I put them in a class that tracks a few other things, and keep a reference to the instance in the model. Works the same.

There are still some challenges. You need to index the data by the batch number - but each sub process's batch numbers start with 0. And they do not all have exactly the same number of batches, as the sampler has to adapt total batches when things don't divide evenly. What I did was to add a dimension to capture the process rank - always 0 based. I also sized the storage on the batch count / number of process, rounding down - and not worrying about the few missed elements of data.

Of course, DistributedSampler will hand one or more sub processes 1 or 2 more batches than you have allowed room for - obviously don't try to index a position in the tensor beyond the max index-1.
Computing means, std devation etc is no different, as torch doesn't care how many dimensions a tensor has when computing these.

The result statistics for me have been essentially identical to running on a single gpu - except for the time which has gone from 15 minutes/epoch to 85 secs/epoch.

Hope this helps,

seth

[matthew-z]

I just read the source code, and actually PL didn't do much modification to the original DDP except redirecting forward to train/validation/test_step.

In DDP, I don't think different sub-processes share the same tensor. Try to put a breakpoint in train_step and run PyCharm debugger, and you will find that the loss values are different in sub-processes, so do the validation outputs. The logger only records the rank-0 process.

[sneiman]

Sorry - my explanation was obviously insufficent. This definitely works, and - I just have to explain it better. I don't believe the logger() behavior is indicative of what happens with the tensor.

Try this - in the init() of your model class, create test tensor with 1 value per gpu you are using. Also create a flag so that this is only called once per process:

    self.firstEpochFlag   = True
    self.testData         = torch.Tensor([0, 0, 0])  # 3 elements - 1 for each gpu used in this test

In on_epoch_start(), on the first epoch, write into self.test.data, and print it:

if firstEpochFlag:
    self.testData[self.trainer.proc_rank] = self.trainer.proc_rank+1
    print(f"testData {self.testData} {self.trainer.proc_rank}")
    self.firstEpochFlag = False

With 3 gpus you will see something like this:

testData tensor([0., 0., 3.]) 2 cpu
testData tensor([0., 2., 3.]) 1 cpu
testData tensor([1., 2., 3.]) 0 cpu

As you can see, the the self.testData tensor is collecting ALL the values and ALL the values are seen by EACH process. And that tensor is on the cpu. If there was one in each sub process we would see:

testData tensor([0., 0., 3.]) 2 cuda:2
testData tensor([0., 2., 0.]) 1 cuda:1
testData tensor([1., 0., 0.]) 0 cuda:0

I can only conclude that there is only one copy of it and all the sub processes can see it, or it is coalesced by ddp for us.

The key issue here is you can use this to produce a process rank indexed tensor of result values that you store yourself - at the end of training_step() - to use to calculate statistics.

This works. Before I did this, I had the same problem you had - loss values only recorded for the 0 gpu, and therefore of questionable value. With this, the statistics make sense, and are essentially identical to what I get if I use on a single gpu.

I hope this is useful to you,

seth

FWIW - there are some behaviors here that bear thinking hard about, As already noted, the printed testData contents only make sense if there is only one copy - or ddp merges this tensor for us. Interestingly though, there must be 3 copies of firstEpochFlag - or the printing would only have been done once as it is set to False the first time it is seen. The choices made in DDP about which elements of the model class are copied to sub processes must be pretty well thought through.

Again, hope this helps. I can show you my entire model class if you like - there is a lot in it ...

seth

[sneiman]

I have verified that there is one testData tensor per proc - so ddp must be coalescing them

[williamFalcon]

Lightning only routes forward to training_step, validation_step. BUT, we should be calling dist.all_gather with the outputs out of training_step and validation_step if we want full batch metrics.

so, maybe someone wants to submit a PR for this?
@sneiman @brucemuller @matthew-z

[sneiman]

My solution works for me - and has the benefit of allowing me compute any statistics I like on the results. Of course, my model code has to know a little about what only doing certain things on proc 0 ...

[jmarsil]

@sneiman I'm facing a similar issue. Could you show the entire model class?

Aside: When I try to pass a dictionary of tensors into progress_bar or even log, I've been getting this consistent error. (Training using 'ddp' on 4 gpus). Have a follow up questions, why can't we pass dictionaries of tensors like 'dp'?

Here is the error:

Let me know your thoughts!

[sneiman]

If I am not mistaken, your error as above is because in v.item(), v has more than one value - so item() cannot convert it to a python scalar. I do not use most of the metrics and progress bar features of pytorch-lightning, but I feel pretty certain that is your problem here.

I will post more of my code for you to take a look at to see how I collect information, in the next post. The overall idea is pretty simple ... declare a multi-dimensional tensor to hold the information you want to collect, with one dimension per gpu. then write into it during running each gpu's data in its own portion of the tensor. pl will reduce the to coalescel. you still have to mindful of multiprocessing. example to come.

[jmarsil]

Ahh thanks for doing that. Right, that would make sense. Since each GPU has its own set of processes, it probably passes a tensor with len(n_gpus) to the callbacks step correct? In the mean time I will try to implement something & update you with how that goes.

[sneiman]

Let's see if this will help:

Within my model I have an instance of a class (cleverly named 'data') which holds the data loaders, and does the work of breaking the data in train, validation and test sets. Because statistics and history change along with data format, I keep all my history and statistics calculations in the same class. It gets tweaked whenever I change the data, or use the model in a way that affects statistics or data collected.

In this class I create my history tensors, and related statistic variables:

class data():
    def __init__(self, trn_ratio=.6, tot_batch_size=16, num_gpus=1, back_end='dp', pct_to_use=1.0):

        self.tot_batch_size     = tot_batch_size
        self.num_gpus           = num_gpus
        self.gpu_batch_size     = self.tot_batch_size//self.num_gpus if back_end=='dp' else self.tot_batch_size

        # req'd transforms
        self.basictransform     = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

        # load basic data set and generate subsets
        # get len and calc end index for each subset
        ds_name                 = f'GUAC_faces_square_w125_dataset_classes_{num_classes}'
        guac_dataset            = GUAC('./data/GUAC', ds_name, transform=self.basictransform, target_transform=None, labelmask=None)

        # create end points for each subset
        dst_end                 = int(len(guac_dataset)*pct_to_use)
        trn_end                 = int(trn_ratio*dst_end)
        val_end                 = int(((1.0-trn_ratio)/2.0)*dst_end)+trn_end

        # create subsets, get batch cnts for each susbset, max idx for each subset

        self.trn_batch_cnt      = len(self.trn_dataset)//self.gpu_batch_size
        self.val_batch_cnt      = len(self.val_dataset)//self.gpu_batch_size
        self.tst_batch_cnt      = len(self.tst_dataset)//self.gpu_batch_size

        self.max_trn_idx        = self.trn_batch_cnt//self.num_gpus
        self.max_val_idx        = self.val_batch_cnt//self.num_gpus
        self.max_tst_idx        = self.tst_batch_cnt//self.num_gpus

        ..... skipping stuff you don't need to see
        # HERE is what you are looking for ... I included the rest because the code refers to some of the calculated values above

        # just showing some of it and for training only to keep things simple ...
        # per batch loss, accuracy history - cleared every epoch
        # epoch history by batch
        self.trn_batch_loss     = torch.Tensor(self.num_gpus, self.max_trn_idx, self.gpu_batch_size)   .fill_(0.0)
        self.trn_batch_acc      = torch.Tensor(self.num_gpus, self.max_trn_idx, self.gpu_batch_size, 2).fill_(0.0)

        self.trn_cw_cor         = torch.Tensor(num_classes).fill_(0.0)
        self.trn_ch_cor         = torch.Tensor(num_classes).fill_(0.0)

        self.trn_avg_loss       = 0.0
        self.trn_avg_acc        = 0.0
        self.trn_avg_acc_w      = 0.0
        self.trn_avg_acc_h      = 0.0
        self.trn_acc_std_w      = 0.0
        self.trn_acc_std_h      = 0.0

    # this function is called at the end of every training step to collect the info
    # loss tensor, accuracy in width, accuracy in height, the gpu as a index into history, AND idx - one entry per batch
    # notice that I copy the information, so that I am not keeping tensors around across iterations
    # without this, and a few other things, you create a memory leak
    # whypr
    def cap_trn_batch(self, loss, acc_w, acc_h, gpu, idx):
        if idx<self.max_trn_idx:
            for i in range(len(acc_w)):
                self.trn_batch_loss[gpu][idx][i]   = loss.item()
                self.trn_batch_acc[gpu][idx][i,0]  = acc_w[i].item()
                self.trn_batch_acc[gpu][idx][i,1]  = acc_h[i].item()

    # function to clear the tensors - just fill them so we don't reallocate
    def clear_trn_results(self):
        self.trn_batch_loss.fill_(0.0)
        self.trn_batch_acc. fill_(0.0)

    # called once an epoch to generate statistics
    def calc_epoch_stats(self):

        self.trn_avg_loss   = self.trn_batch_loss.        abs().mean()
        self.trn_avg_acc    = self.trn_batch_acc.         abs().mean()
        self.trn_avg_acc_w  = self.trn_batch_acc[:,:,:,0].mean()
        self.trn_avg_acc_h  = self.trn_batch_acc[:,:,:,1].mean()
        self.trn_acc_std_w  = self.trn_batch_acc[:,:,:,0].std()
        self.trn_acc_std_h  = self.trn_batch_acc[:,:,:,1].std()

Ok - thats it ... lets look at training_step:

   # called once for each training batch
    def training_step(self, batch, batch_nb):

        # fwd, loss, acc: classes and ypr
        imgs, lab_w, lab_h  = batch[0], batch[1][:,0].long(), batch[1][:,1].long()
        lab_y               = batch[1][:,2]
        lab_p               = batch[1][:,3]
        lab_r               = batch[1][:,4]
        out_w, out_h, p     = self.forward(imgs)

        trn_loss_w          = self.crit_w(out_w,  lab_w)
        trn_loss_h          = self.crit_h(out_h,  lab_h)*1.5
        trn_loss_y          = self.crit_p(p[:,0], lab_y)*.75
        trn_loss_p          = self.crit_p(p[:,1], lab_p)*1.5
        trn_loss_r          = self.crit_p(p[:,2], lab_r)
        trn_loss            = trn_loss_w+trn_loss_h+trn_loss_y+trn_loss_p+trn_loss_r

        acc_w               = lab_w==out_w.argmax(1)
        acc_h               = lab_h==out_h.argmax(1)

        # track acc by class
        for i in range(len(acc_w)):
            self.data.trn_cw_cor[lab_w[i]] += acc_w[i].item()
            self.data.trn_ch_cor[lab_h[i]] += acc_h[i].item()

        # capture data - adjust gpu arg to be zero based to store in data arrays
        dev                 = batch[0].device.index
        self.data.cap_trn_batch(trn_loss.detach(), acc_w.detach(), acc_h.detach(), dev-self.base_gpu_num, batch_nb)



        del imgs, lab_w, lab_h, lab_y, lab_p, lab_r, out_w, out_h, p, acc_w, acc_h, trn_loss_w, trn_loss_h, trn_loss_y, trn_loss_p, trn_loss_r
        return {'loss': trn_loss}

I think you can see the idea here. Create history TENSORS as part of your model - being in an instance that is part of your model is just as good. Write functions to put data in them, clear them and to calculate statistics. Put the data in them after every step. Calculate and use the results as you need to - generally once an epoch. BE CAREFUL to detach those tensors before you make the cap call - or else you will have a memory leak. I also found that I needed to delete them to avoid memory leaks.

Because you are using distributed processing you need to force the processes to all finish their calculations before you clear the data. I do it at the end of on_epoch_start

        .......

        # barrier() causes all the processes to wait for each other
        # only needed for ddp because other approaches have a single process - 
        # I think this would also be needed for ddp2, but I dont use it so have not allowed for it
        if self.back_end == 'ddp': torch.distributed.barrier()
        self.data.clear_all_results()

I think you can get it from here ....