Trainer: distributed eval
Author: julien-c
Pull request: no
Labels: name: Good First Issue

Tagging this as a Good First issue if anyone's interested.

Comments:
Hi Julian, would you please provide more information on this issue?

In our Trainer, in `torch.distributed` mode, eval/predict are currently running on a single node, even though there's no fundamental reason for this to be the case.

Supporting would probably require writing a new instance of torch's [DistributedSampler](https://github.com/pytorch/pytorch/blob/master/torch/utils/data/distributed.py) maybe named `SequentialDistributedSampler` and then re-aggregating the inference results on a single node to perform the metrics computation.

Hey @julien-c! I'm new here and would like to work on this. Doesn't seem like anyone else is currently working on this.

Go for it @abhavk!

Hey @julien-c ! I've given some thought to what needs to be done after looking at the documentation and I see two ways to implement this - 

1. Use an argument is_distributed in eval and predict functions and set that to true while calling from training loop in distributed setting. This does not mandate changes to the training loop to preserve current functionality (but the loop will need to be changed for actual implementation of distributed evaluation in the training loop).

2. Do the is_distributed check within the eval and predict functions. This will mandate changes to the training loop in order to preserve current functionality. 

I'm personally leaning towards option 2 because that allows no explicit checks for distributed setting in the training loop - which seems cleaner, but there are still questions around how the training loop should change. 

In either case, for the rest of the implementation my plan is to:

1. (as you have suggested) implement a slightly different version of the DistributedSampler (which does not repeat any indices while sampling over the dataset) and use that in the eval/predict functions. 

2. After that call the prediction loop to calculate the outputs and use a function like dist.gather on the rank 0 node to collect outputs. I feel like this aggregating may need to be called within the prediction loop in order to allow for computation of custom metrics (which probably needs to be done on a single node), but correct me if I'm wrong here. 

This is based on my current understanding so would be happy to reconsider the approach or implementation details. Thanks!

@abhavk I've written distributed eval in my own code and AFAIK it is not required to modify the sampler. The point of a distributed sampler in DDP is that each process (sampler + its dataloader) only receives a portion of the data. It is not the case that all processes process the same data!

Also see [the documentation](https://pytorch.org/docs/stable/data.html#torch.utils.data.distributed.DistributedSampler):

> Sampler that restricts data loading to a subset of the dataset.
> 
> It is especially useful in conjunction with torch.nn.parallel.DistributedDataParallel. In such case, each process can pass a DistributedSampler instance as a DataLoader sampler, and load a subset of the original dataset that is exclusive to it.

That should be fine. What should be done, though is that to average well, all results for a given epoch (or step) should be saved (in memory) by each individual process. then, at the end of evaluation, all results should be gathered to a single process which can calculate the metrics over all results. So basically: you do the forward pass distributed, keep all predictions, and to average them you gather them on one GPU to do averaging/other computations. 

So I think that the second point is indeed correct, but the first one is not necessary AFAIK.

If you need help or want to discuss it, let me know and mention me in this topic!

@BramVanroy Thanks for that input. I have gone through the documentation and implementation for DistributedSampler, and from [the code](https://github.com/pytorch/pytorch/blob/master/torch/utils/data/distributed.py) I can see that when the number of datapoints in the dataset is not divisible by the number of processes, it is adding extra samples by repeating the first few indices in order to make it evenly divisible. So for example a dataset with 5 datapoints and being distributed across 2 processes, will actually have 6 samples and that will be done by repeating the first sample (if the data is not shuffled, otherwise a different random sample) in both processes.

While this may be fine for training, I don't think this is desirable/okay during the evaluation/prediction phase, but do let me know if that understanding is not correct (Tagging @julien-c here too). That is primarily why I was suggesting writing a different implementation for the Distributed Sampler, where even though some nodes might have one fewer datum, there is no repetition. 

Actually now that I think about it, even if it is not ideal to repeat eval samples, the difference may not be worth writing a different sampler implementation, since if someone need to use distributed processing, it may be safe to assume that they have a fairly large eval dataset where the number of samples is >> number of processes. 

> Actually now that I think about it, even if it is not ideal to repeat eval samples, the difference may not be worth writing a different sampler implementation, since if someone need to use distributed processing, it may be safe to assume that they have a fairly large eval dataset where the number of samples is >> number of processes.

Exactly. In addition, the DataLoaders's "drop_last" might be of help here, though I am not entirely sure how it would integrate with a distributed sampler.

I don't really agree with this. For evaluation results to be comparable you pretty much need to have the inference results for exactly all the samples.

Also we use the same code for eval and for prediction (the loop is called `_prediction_loop()`), and in the case of prediction, e.g. when predicting on the GLUE hidden test dataset you need to output all the predictions in order.

You probably could do this using torch's DistributedSampler with some hacks (adding dummy samples to be divisible by number of nodes and dropping them afterwards) and knowledge of its internal implementation (how to reorder samples at the end) but it's probably way more explicit to just implement a dedicated one. (torch's DistributedSampler is very few lines of code anyways)

I might take a stab at this today and will ping you here.

**Update**: edited slightly to separate eval and prediction cases

Yes, to be able to compare studies, that is true. For production-oriented systems and large eval datasets that does not matter as much.

Idea, subclass the DistributedSampler and add a property `self.dummy_idxs: List[List[int]]`. The __iter__ method of DistributedSampler can stay the same with the only addition that we keep track of the indices that are larger than the length of self.dataset for each batch. self.dummy_idxs then contains a list of a list for each batch that contains the position in that batch where an item was a dummy item. 

I don't have time to work on this, though, but happy to think along.

Sure @julien-c. I've taken a first stab and this is really quite an easy fix - `total_size` in the `__init__` function just changes to the length of the dataset and the below piece of code within the `iter` function should be removed (with minor changes to the `num_samples` variable) - 
` # add extra samples to make it evenly divisible`
`indices += indices[:(self.total_size - len(indices))]`
`assert len(indices) == self.total_size` 

If this solution looks fine, can you tell me the proper way to add this to the repository when making the update - as in which folder would this piece go in (is it /utils)? 

I took a stab at it in the PR above but would love if you could review and/or test it.

Added my comments @julien-c 

Closed by #4243