Proper way to log things when using DDP
Hi, I was wondering what is the proper way of logging metrics when using DDP. I noticed that if I want to print something inside validation_epoch_end it will be printed twice when using 2 GPUs. I was expecting validation_epoch_end to be called only on rank 0 and to receive the outputs from all GPUs, but I am not sure this is correct anymore. Therefore I have several questions:

validation_epoch_end(self, outputs) - When using DDP does every subprocess receive the data processed from the current GPU or data processed from all GPUs, i.e. does the input parameter outputs contains the outputs of the entire validation set, from all GPUs?
If outputs is GPU/process specific what is the proper way to calculate any metric on the entire validation set in validation_epoch_end when using DDP?

I understand that I can solve the printing by checking self.global_rank == 0 and printing/logging only in that case, however I am trying to get a deeper understanding of what I am printing/logging in this case.
Here is a code snippet from my use case. I would like to be able to report f1, precision and recall on the entire validation dataset and I am wondering what is the correct way of doing it when using DDP.
    def _process_epoch_outputs(self,
                               outputs: List[Dict[str, Any]]
                               ) -> Tuple[torch.Tensor, torch.Tensor]:
        """Creates and returns tensors containing all labels and predictions

        Goes over the outputs accumulated from every batch, detaches the
        necessary tensors and stacks them together.

        Args:
            outputs (List[Dict])
        """
        all_labels = []
        all_predictions = []

        for output in outputs:
            for labels in output['labels'].detach():
                all_labels.append(labels)

            for predictions in output['predictions'].detach():
                all_predictions.append(predictions)

        all_labels = torch.stack(all_labels).long().cpu()
        all_predictions = torch.stack(all_predictions).cpu()

        return all_predictions, all_labels

    def validation_epoch_end(self, outputs: List[Dict[str, Any]]) -> None:
        """Logs f1, precision and recall on the validation set."""

        if self.global_rank == 0:
            print(f'Validation Epoch: {self.current_epoch}')

        predictions, labels = self._process_epoch_outputs(outputs)
        for i, name in enumerate(self.label_columns):

            f1, prec, recall, t = metrics.get_f1_prec_recall(predictions[:, i],
                                                             labels[:, i],
                                                             threshold=None)
            self.logger.experiment.add_scalar(f'{name}_f1/Val',
                                              f1,
                                              self.current_epoch)
            self.logger.experiment.add_scalar(f'{name}_Precision/Val',
                                              prec,
                                              self.current_epoch)
            self.logger.experiment.add_scalar(f'{name}_Recall/Val',
                                              recall,
                                              self.current_epoch)

            if self.global_rank == 0:
                print((f'F1: {f1}, Precision: {prec}, '
                       f'Recall: {recall}, Threshold {t}'))
Hi all,
Sorry we have not got back to you in time, let me try to answer some of your questions:

Is validation_epoch_end only called on rank 0?

No, it is called by all processes

What does the sync_dist flag do:

Here is the essential code:

  
    
      pytorch-lightning/pytorch_lightning/core/step_result.py
    
    
        Lines 108 to 115
      in
      a72a799
    
  
  
    

        
          
           if sync_dist and isinstance(value, (torch.Tensor, numbers.Number)): 
        

        
          
               is_dist_initialized = torch.distributed.is_available() and torch.distributed.is_initialized() 
        

        
          
               # TODO: Find a way to make the reduction only once, so we don't need to clone. 
        

        
          
               if is_dist_initialized and isinstance(value, torch.Tensor): 
        

        
          
                   value = value.clone() 
        

        
          
               else: 
        

        
          
                   value = torch.tensor(value, device=device, dtype=torch.float) 
        

        
          
               value = sync_fn(value, group=sync_dist_group, reduce_op=sync_dist_op) 
        
    
  


If sync_dist=True then it will as default call the sync_ddp function which will sum the value across all processes using torch.distributed.all_reduce

  
    
      pytorch-lightning/pytorch_lightning/utilities/distributed.py
    
    
         Line 120
      in
      a72a799
    
  
  
    

        
          
           def sync_ddp( 
        
    
  


Use this flag if you want to synchronize the value between different processes.

How to print stuff in distributed lightning:

Recommended is using either the rank_zero_info function. Import as:
from pytorch_lightning.utilities import rank_zero_info
or use the rank_zero_only decorator (imported from the same module) which can be wrapped around any function such that it only gets called on rank=0. Each logger experiment is decorated with rank_zero_experiment which internally calls rank_zero_only

  
    
      pytorch-lightning/pytorch_lightning/loggers/base.py
    
    
        Lines 31 to 43
      in
      a72a799
    
  
  
    

        
          
           def rank_zero_experiment(fn: Callable) -> Callable: 
        

        
          
               """ Returns the real experiment on rank 0 and otherwise the DummyExperiment. """ 
        

        
          
            
        

        
          
               @wraps(fn) 
        

        
          
               def experiment(self): 
        

        
          
            
        

        
          
                   @rank_zero_only 
        

        
          
                   def get_experiment(): 
        

        
          
                       return fn(self) 
        

        
          
            
        

        
          
                   return get_experiment() or DummyExperiment() 
        

        
          
            
        

        
          
               return experiment 
        
    
  



What about pytorch_lightning.metrics (now known as torchmetrics)

Our own metrics have custom synchronization going on. Any metric will automatically synchronize between different processes whenever metric.compute() is called. Metrics calculated this way should therefore not be logged using sync_dist=True.

Recommended way of logging:

Using self.log in your lightning module
Not sure this answers all questions.