code: def _run_pytorch_trial(
      trial_class: "Type[det.pytorch.PyTorchTrial]",
      info: det.ClusterInfo,
  ) -> int:
      from determined import pytorch
  
      det.common.set_logger(info.trial._debug)
  
      logger.debug("Starting harness.")
  
      with maybe_periodic_stacktraces(info.trial._debug):
          with pytorch.init(
              hparams=info.trial.hparams,
              exp_conf=info.trial._config,
              aggregation_frequency=int(
                  info.trial._config["optimizations"][
                      "aggregation_frequency"
                  ]
              ),
          ) as train_context:
              fp16_compression = bool(
                  info.trial._config["optimizations"][
                      "gradient_compression"
                  ]
              )
              average_aggregated_gradients = bool(
                  info.trial._config["optimizations"][
                      "average_aggregated_gradients"
                  ]
              )
  
              train_context._set_default_gradient_compression(
                  fp16_compression
              )
              train_context._set_default_average_aggregated_gradients(
                  average_aggregated_gradients
              )
              train_context._set_is_pre_trainer()
  
              trial_inst = trial_class(train_context)
  
              if (
                  train_context.distributed.size > 1
                  and not train_context.distributed.rank == 0
              ):
                  log_level = (
                      logging.DEBUG
                      if info.trial._debug
                      else logging.WARNING
                  )
                  logging.getLogger().setLevel(log_level)
  
              logger.info(
                  f"Creating {pytorch._PyTorchTrialController.__name__} with {trial_class.__name__}."
              )
  
              trainer = pytorch.Trainer(trial_inst, train_context)
  
              trainer.configure_profiler(
                  sync_timings=bool(
                      info.trial._config["profiling"]["sync_timings"]
                  ),
                  enabled=bool(
                      info.trial._config["profiling"]["enabled"]
                  ),
                  begin_on_batch=info.trial._config["profiling"][
                      "begin_on_batch"
                  ],
                  end_after_batch=info.trial._config["profiling"][
                      "end_after_batch"
                  ],
              )
  
              if "global_batch_size" in info.trial.hparams:
                  global_batch_size = int(
                      info.trial.hparams["global_batch_size"]
                  )  # type: Optional[int]
              else:
                  global_batch_size = None
  
              trainer.fit(
                  checkpoint_period=pytorch.TrainUnit._from_values(
                      **info.trial._config["min_checkpoint_period"],
                      global_batch_size=global_batch_size,
                  ),
                  validation_period=pytorch.TrainUnit._from_values(
                      **info.trial._config["min_validation_period"],
                      global_batch_size=global_batch_size,
                  ),
                  reporting_period=pytorch.Batch(
                      info.trial._config["scheduling_unit"]
                  ),
                  checkpoint_policy=info.trial._config[
                      "checkpoint_policy"
                  ],
                  latest_checkpoint=info.latest_checkpoint,
                  step_zero_validation=info.trial._config[
                      "perform_initial_validation"
                  ],
                  test_mode=False,
              )
  
      return 0
  
file_path: determined/harness/determined/exec/harness.py
imports: import argparse
  import contextlib
  import faulthandler
  import logging
  import sys
  from typing import Iterator, Optional, Type
  import determined
  from determined import core, horovod, load
  from determined.common.api import analytics, certs
label_desc: Use image node_id 3 for calling a global function with example usage: _run_pytorch_trial(trial_class, info) and returns: int