code: def _set_data_loaders(self) -> None:
      skip_batches = self.env.steps_completed
  
      # Training and validation data loaders are not built for every slot when model parallelism
      # is used.
      self.training_loader = (
          None
      )  # type: Optional[torch.utils.data.DataLoader]
      self.validation_loader = (
          None
      )  # type: Optional[torch.utils.data.DataLoader]
      self.num_validation_batches = None  # type: Optional[int]
      self.validation_batch_size = None  # type: Optional[int]
  
      # We currently only allow one model parallel strategy per DeepSpeedTrial.
      # We also assume that the data loader is tied to this one parallelization strategy.
      nreplicas = self.context._mpu.data_parallel_world_size
      rank = self.context._mpu.data_parallel_rank
  
      # The data loader is only required on ranks that take the data as input or require
      # the data to compute the loss.  There could be intermediate model parallel ranks
      # that do not need a data loader at all.
      if self.context._mpu.should_build_data_loader:
          train_data = self.trial.build_training_data_loader()
          if isinstance(train_data, pytorch.DataLoader):
              # Repeating the data loader is the default behavior for DeepSpeed data loaders when
              # using pipeline parallel.
              self.training_loader = train_data.get_data_loader(
                  repeat=True,
                  skip=skip_batches,
                  num_replicas=nreplicas,
                  rank=rank,
              )
          else:
              # Non-determined DataLoader; ensure the user meant to do this.
              if not self.context._data_repro_checks_disabled:
                  raise RuntimeError(
                      pytorch._dataset_repro_warning(
                          "build_training_data_loader",
                          train_data,
                          is_deepspeed_trial=True,
                      )
                  )
              self.training_loader = train_data
              logger.warning(
                  "Please make sure custom data loader repeats indefinitely."
              )
  
          validation_data = self.trial.build_validation_data_loader()
          if isinstance(validation_data, pytorch.DataLoader):
              # For pipeline parallel models, we may evaluate on slightly fewer micro batches
              # than there would be in a full pass through the dataset due to automated
              # micro batch interleaving.
              self.validation_loader = validation_data.get_data_loader(
                  repeat=False,
                  skip=0,
                  num_replicas=nreplicas,
                  rank=rank,
              )
  
              if self.context.use_pipeline_parallel:
                  if (
                      len(self.validation_loader)
                      < self.context.num_micro_batches_per_slot
                  ):
                      raise det.errors.InvalidExperimentException(
                          "Number of train micro batches in validation data loader should not be "
                          "less than the number of gradient accumulation steps when using "
                          "pipeline parallelism."
                      )
                  excluded_micro_batches = (
                      len(validation_data)
                      % self.context.num_micro_batches_per_slot
                  )
                  if excluded_micro_batches:
                      logger.warning(
                          "We will compute validation metrics over "
                          f"{excluded_micro_batches} fewer micro batches on rank "
                          f"{self.context.distributed.get_rank()}"
                      )
          else:
              # Non-determined DataLoader; ensure the user meant to do this.
              if not self.context._data_repro_checks_disabled:
                  raise RuntimeError(
                      pytorch._dataset_repro_warning(
                          "build_validation_data_loader",
                          validation_data,
                          is_deepspeed_trial=True,
                      )
                  )
              if self.context.use_pipeline_parallel:
                  logger.warning(
                      "Using custom data loader, please make sure len(validation loader) is "
                      "divisible by gradient accumulation steps."
                  )
              self.validation_loader = validation_data
  
          # We use cast here instead of assert because the user can return an object that behaves
          # like a DataLoader but is not.
          self.validation_loader = cast(
              torch.utils.data.DataLoader, self.validation_loader
          )
          self.num_validation_batches = len(self.validation_loader)
          self.validation_batch_size = pytorch.data_length(
              next(iter(self.validation_loader))
          )
  
          if self.context.use_pipeline_parallel:
              self.num_validation_batches = (
                  self.num_validation_batches
                  // self.context.num_micro_batches_per_slot
              )
              self.validation_batch_size *= (
                  self.context.num_micro_batches_per_slot
              )
  
      # We will do a gather on to get train and val loader lengths and broadcast to all slots.
      self.context._epoch_len = (
          len(self.training_loader)
          if self.training_loader is not None
          else None
      )
      all_epoch_lens = self.context.distributed.gather(
          self.context._epoch_len
      )
      if self.is_chief:
          all_epoch_lens = [
              le for le in all_epoch_lens if le is not None
          ]
          if min(all_epoch_lens) < max(all_epoch_lens):
              logger.warning(
                  "Training data loader length inconsistent across ranks. "
                  "Using the minimum for epoch length."
              )
          self.context._epoch_len = (
              min(all_epoch_lens)
              // self.context.num_micro_batches_per_slot
          )
      self.context._epoch_len = self.context.distributed.broadcast(
          self.context._epoch_len
      )
  
      all_tuples = self.context.distributed.gather(
          (self.num_validation_batches, self.validation_batch_size)
      )
      if self.is_chief:
          all_num_validation_batches, all_validation_batch_size = zip(
              *all_tuples
          )
          all_num_validation_batches = [
              le for le in all_num_validation_batches if le is not None
          ]
          if min(all_num_validation_batches) < max(
              all_num_validation_batches
          ):
              logger.warning(
                  "Validation data loader length inconsistent across ranks. "
                  "Using the minimum for validation length."
              )
          self.num_validation_batches = min(all_num_validation_batches)
          all_validation_batch_size = [
              le for le in all_validation_batch_size if le is not None
          ]
          if min(all_validation_batch_size) < max(
              all_validation_batch_size
          ):
              logger.warning(
                  "Validation batch size inconsistent across ranks. "
                  "Num inputs tracking for validation will be incorrect."
              )
          self.validation_batch_size = min(all_validation_batch_size)
  
      (
          self.num_validation_batches,
          self.validation_batch_size,
      ) = self.context.distributed.broadcast(
          (self.num_validation_batches, self.validation_batch_size)
      )
  
imports: import abc
  import contextlib
  import logging
  import os
  import pathlib
  import pickle
  import random
  import time
  import warnings
  from typing import Any, Callable, Dict, Iterator, List, Optional, Type, Union, cast
  import deepspeed
  import numpy
  import torch
  from deepspeed.runtime import dataloader
  import determined
  from determined import layers, pytorch, util, workload
  from determined.pytorch import deepspeed
  from determined.pytorch import dsat
label_desc: Use image node_id 4 for calling the DeepSpeedTrialController obj's underlying member method code with example usage: obj._set_data_loaders() without return types