code: def _test_distrib_integration(device):
      rank = idist.get_rank()
  
      n_iters = 10
      batch_size = 10
      n_classes = 10
  
      def _test(metric_device):
          y_true = (
              torch.arange(
                  0, n_iters * batch_size, dtype=torch.int64
              ).to(device)
              % n_classes
          )
          y_pred = 0.2 * torch.rand(n_iters * batch_size, n_classes).to(
              device
          )
          for i in range(n_iters * batch_size):
              if np.random.rand() > 0.4:
                  y_pred[i, y_true[i]] = 1.0
              else:
                  j = np.random.randint(0, n_classes)
                  y_pred[i, j] = 0.7
  
          def update_fn(engine, i):
              y_true_batch = y_true[
                  i * batch_size : (i + 1) * batch_size, ...
              ]
              y_pred_batch = y_pred[
                  i * batch_size : (i + 1) * batch_size, ...
              ]
              return y_pred_batch, y_true_batch
  
          evaluator = Engine(update_fn)
  
          precision = Precision(average=False, device=metric_device)
          recall = Recall(average=False, device=metric_device)
  
          def Fbeta(r, p, beta):
              return torch.mean(
                  (1 + beta**2) * p * r / (beta**2 * p + r)
              ).item()
  
          F1 = MetricsLambda(Fbeta, recall, precision, 1)
          F1.attach(evaluator, "f1")
  
          another_f1 = (
              (
                  1.0
                  + precision
                  * recall
                  * 2
                  / (precision + recall + 1e-20)
              )
              .mean()
              .item()
          )
          another_f1.attach(evaluator, "ff1")
  
          data = list(range(n_iters))
          state = evaluator.run(data, max_epochs=1)
  
          y_pred = idist.all_gather(y_pred)
          y_true = idist.all_gather(y_true)
  
          assert "f1" in state.metrics
          assert "ff1" in state.metrics
          f1_true = f1_score(
              y_true.view(-1).cpu(),
              y_pred.view(-1, n_classes).argmax(dim=-1).cpu(),
              average="macro",
          )
          assert f1_true == approx(state.metrics["f1"])
          assert 1.0 + f1_true == approx(state.metrics["ff1"])
  
      for i in range(3):
          torch.manual_seed(12 + rank + i)
          _test("cpu")
          if device.type != "xla":
              _test(idist.device())
  
imports: import os
  import numpy
  import pytest
  import torch
  from pytest import approx
  from sklearn.metrics import f1_score, precision_score, recall_score
  import ignite.distributed
  from ignite.engine import Engine
  from ignite.metrics import Accuracy, Metric, MetricsLambda, Precision, Recall
label_desc: Use image node_id 10 for calling a global function with example usage: _test_distrib_integration(device) and returns: y_pred_batch,  y_true_batch, torch