question: Proper way to log things when using DDP
context: 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}'))
answer: 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.