question: Why is my gpu-util low?
context: I use one node and 4 gpus for training. And I use dali dataloader, I don't know why my gpu util is low, and training is also slow. About 1:30 per epoch, I train for 200 epoches, which will cost 5 hours. It's slower than the project mmclassification, which only cost 3.5 hours. Compared to mmclassification project which can only support torch.utils.data.dataloader, I think if I use dali_dataloader, it will accelerate my training. But as you can see, it's the opposite. I don't know why. Could anyone give me some advice? I use cifar10 dataset. And I train on slurm.
  
  Here is my code.
  main.py
  import pytorch_lightning as pl
  from pytorch_lightning.callbacks import ModelCheckpoint
  from net import ResNet18
  if __name__ == '__main__':
      model = ResNet18()
      trainer = pl.Trainer( max_epochs=200,log_every_n_steps=1,
          log_gpu_memory='min_max',gpus=4,num_nodes=1,accelerator='ddp',
          fast_dev_run=False,callbacks=[ModelCheckpoint(monitor='val_accuracy',mode='max')],
          progress_bar_refresh_rate=1,replace_sampler_ddp=False)
      trainer.fit(model)
  net.py
  import torch
  import torch.nn as nn
  import torch.nn.functional as F
  import pytorch_lightning as pl
  from dataloader import dali_DataLoader,HybridPipe,dali_CIFAR10
  class BasicBlock(nn.Module):
      expansion = 1
  
      def __init__(self, in_planes, planes, stride=1):
          super(BasicBlock, self).__init__()
          self.conv1 = nn.Conv2d(
              in_planes, planes, kernel_size=3, stride=stride, padding=1, bias=False)
          self.bn1 = nn.BatchNorm2d(planes)
          self.conv2 = nn.Conv2d(planes, planes, kernel_size=3,
                                 stride=1, padding=1, bias=False)
          self.bn2 = nn.BatchNorm2d(planes)
  
          self.shortcut = nn.Sequential()
          if stride != 1 or in_planes != self.expansion*planes:
              self.shortcut = nn.Sequential(
                  nn.Conv2d(in_planes, self.expansion*planes,
                            kernel_size=1, stride=stride, bias=False),
                  nn.BatchNorm2d(self.expansion*planes)
              )
  
      def forward(self, x):
          out = F.relu(self.bn1(self.conv1(x)))
          out = self.bn2(self.conv2(out))
          out += self.shortcut(x)
          out = F.relu(out)
          return out
  
  
  class Bottleneck(nn.Module):
      expansion = 4
  
      def __init__(self, in_planes, planes, stride=1):
          super(Bottleneck, self).__init__()
          self.conv1 = nn.Conv2d(in_planes, planes, kernel_size=1, bias=False)
          self.bn1 = nn.BatchNorm2d(planes)
          self.conv2 = nn.Conv2d(planes, planes, kernel_size=3,
                                 stride=stride, padding=1, bias=False)
          self.bn2 = nn.BatchNorm2d(planes)
          self.conv3 = nn.Conv2d(planes, self.expansion *
                                 planes, kernel_size=1, bias=False)
          self.bn3 = nn.BatchNorm2d(self.expansion*planes)
  
          self.shortcut = nn.Sequential()
          if stride != 1 or in_planes != self.expansion*planes:
              self.shortcut = nn.Sequential(
                  nn.Conv2d(in_planes, self.expansion*planes,
                            kernel_size=1, stride=stride, bias=False),
                  nn.BatchNorm2d(self.expansion*planes)
              )
  
      def forward(self, x):
          out = F.relu(self.bn1(self.conv1(x)))
          out = F.relu(self.bn2(self.conv2(out)))
          out = self.bn3(self.conv3(out))
          out += self.shortcut(x)
          out = F.relu(out)
          return out
  
  
  class ResNet(pl.LightningModule):
      def __init__(self, block, num_blocks, num_classes=10):
          super(ResNet, self).__init__()
          self.in_planes = 64
  
          self.conv1 = nn.Conv2d(3, 64, kernel_size=3,
                                 stride=1, padding=1, bias=False)
          self.bn1 = nn.BatchNorm2d(64)
          self.layer1 = self._make_layer(block, 64, num_blocks[0], stride=1)
          self.layer2 = self._make_layer(block, 128, num_blocks[1], stride=2)
          self.layer3 = self._make_layer(block, 256, num_blocks[2], stride=2)
          self.layer4 = self._make_layer(block, 512, num_blocks[3], stride=2)
          self.linear = nn.Linear(512*block.expansion, num_classes)
          self.correct = 0
          self.total_size = 0
      def _make_layer(self, block, planes, num_blocks, stride):
          strides = [stride] + [1]*(num_blocks-1)
          layers = []
          for stride in strides:
              layers.append(block(self.in_planes, planes, stride))
              self.in_planes = planes * block.expansion
          return nn.Sequential(*layers)
  
      def forward(self, x):
          out = F.relu(self.bn1(self.conv1(x)))
          out = self.layer1(out)
          out = self.layer2(out)
          out = self.layer3(out)
          out = self.layer4(out)
          out = F.avg_pool2d(out, 4)
          out = out.view(out.size(0), -1)
          out = self.linear(out)
          return out
      def training_step(self, batch, batch_idx):
          x, y = batch
          x = self(x)
          loss_fn = nn.CrossEntropyLoss()
          loss = loss_fn(x,y)
          predicted = torch.argmax(x, dim=1, keepdim=False)
          self.correct += (predicted == y).sum().item()
          self.total_size += y.size(0)
          self.log('train_loss', loss,prog_bar=True, logger=True)
          self.log('train_accuracy', self.correct/self.total_size,prog_bar=True, logger=True)
          return loss
      def validation_step(self, batch, batch_idx):
          x, y = batch
          x = self(x)
          loss_fn = nn.CrossEntropyLoss()
          loss = loss_fn(x,y)
          predicted = torch.argmax(x, dim=1, keepdim=False)
          self.correct += (predicted == y).sum().item()
          self.total_size += y.size(0)
          self.log('val_loss', loss,on_step=False, on_epoch=True,prog_bar=True, logger=True)
          self.log('val_accuracy', self.correct/self.total_size,prog_bar=True, logger=True)
          return loss
      def validation_epoch_end(self,out):
          self.log('val_accuracy', self.correct/self.total_size,prog_bar=True, logger=True)
          self.correct=0
          self.total_size=0
      def train_epoch_end(self,out):
          self.log('train_accuracy', self.correct/self.total_size,prog_bar=True, logger=True)
          self.correct=0
          self.total_size=0
      def configure_optimizers(self):
          optimizer = torch.optim.SGD(self.parameters(), lr=0.1)
          scheduler = torch.optim.lr_scheduler.MultiStepLR(optimizer, [100,150], gamma=0.1, last_epoch=-1, verbose=False)
          return [optimizer],[scheduler]
      def train_dataloader(self):
          loader = dali_DataLoader(pipelines=HybridPipe(dali_CIFAR10(root='./data'), batch_size=32, pad_ratio=1.25,num_threads=4,
               is_distribute=True, crop_size=32,ramdom_flip=True,
               normalize=dict(mean=[125.307, 122.961, 113.8575],std=[51.5865, 50.847, 51.255])))
          return loader
  
      def val_dataloader(self):
          loader = dali_DataLoader(pipelines=HybridPipe(dali_CIFAR10(root='./data',test_mode=True), batch_size=100,
               normalize=dict(mean=[125.307, 122.961, 113.8575],std=[51.5865, 50.847, 51.255])))
          return loader
  def ResNet18():
      return ResNet(BasicBlock, [2, 2, 2, 2])
  dataloader.py
  import os,sys,math,random,pickle
  import torch
  import numpy as np
  import torch.distributed as dist
  try:
      from nvidia import dali
      from nvidia.dali.pipeline import Pipeline
      import nvidia.dali.types as types
      import nvidia.dali.fn as fn
      import nvidia.dali.ops as ops
      from nvidia.dali.plugin.pytorch import DALIClassificationIterator
  except:
      print('Could not import DALI')
  class dali_DataLoader():
      def __init__(self, pipelines, **kwargs):
          pipelines.build()
          try:
              self._dali_iterator = DALIClassificationIterator(pipelines=pipelines, size=len(pipelines.iterator.indices))
              self.sampler = pipelines.iterator
          except:
              self._dali_iterator = DALIClassificationIterator(pipelines=pipelines, reader_name='Reader')
              self.sampler = self
      def set_epoch(self,epoch):
          pass
      def __iter__(self):
          return self
  
      def __len__(self):
          return int(math.ceil(self._dali_iterator._size / self._dali_iterator.batch_size))
      def __next__(self):
          
          try:
              data = next(self._dali_iterator)
          except StopIteration:
              self._dali_iterator.reset()
              raise StopIteration
          # Decode the data output
          input = data[0]['data']
          target = data[0]['label'].squeeze().long()
  
          return input,target
  class identity():
      def __call__(self,x,*tmp,**kargs):
          return x
  class HybridPipe(Pipeline):
      def __init__(self,dataset, batch_size, file_root=None,filelist_path=None,num_threads=1, pad_ratio=1,is_distribute=True, resize=None,crop_size=[0,0],ramdom_flip=False,normalize=None,random_rotate_degree=None):
          device_id = torch.cuda.current_device()
          print("device_id",device_id)
          super(HybridPipe, self).__init__(batch_size, num_threads, device_id, seed=12 + device_id)
          
          if is_distribute:
              if filelist_path is not None:
                  if file_root is None:
                      raise Exception("if provide filelist_path, then must provide file_root")
                  else:
                      self.input = ops.readers.File(file_root=file_root,file_list=filelist_path,num_shards=dist.get_world_size(),prefetch_queue_depth=num_threads,read_ahead=True,shard_id=dist.get_rank())
                      self.decode = ops.decoders.Image(device="mixed", output_type=types.RGB)
                      self.use_file=True
              else:
                  self.iterator = iter(Distribute_Input_Iter(dataset, batch_size))
                  #self.input = ops.ExternalSource(source=self.iterator, num_outputs=2)
                  self.input = ops.ExternalSource()
                  self.input_label = ops.ExternalSource()
                  self.use_file=False
          else:
              if filelist_path is not None:
                  if file_root is None:
                      raise Exception("if provide filelist_path, then must provide file_root")
                  else:
                      self.input = ops.readers.File(file_root=file_root,file_list=filelist_path,num_shards=dist.get_world_size(),prefetch_queue_depth=num_threads,read_ahead=True,shard_id=dist.get_rank())
                      self.decode = ops.decoders.Image(device="mixed", output_type=types.RGB)
                      self.use_file=True
              else:
                  self.iterator = iter(Normal_Input_Iter(dataset, batch_size))
                  self.input = ops.ExternalSource()
                  self.input_label = ops.ExternalSource()
                  self.use_file=False
          dali_device = "gpu"
          
          if isinstance(resize,(tuple,list)) and len(resize)==2:
              self.resize = ops.Resize(size=tuple(resize))
          elif isinstance(resize,(int, float)):
              self.resize = ops.Resize(size=tuple(resize,resize))
          else:
              self.resize = identity()
          if normalize is not None and isinstance(normalize,dict):
              self.mean = normalize.get('mean',0)
              self.std = normalize.get('std',1)
          else:
              self.mean = 0
              self.std = 1
          if isinstance(crop_size, (int, float)):
              crop_size = [crop_size,crop_size]
          if (len(crop_size)==2 and (crop_size[0]==0 or crop_size[1]==0)):
              self.crop = identity()
          else:
              self.crop = ops.Crop(device=dali_device, crop_h=crop_size[0], crop_w=crop_size[1])
          if pad_ratio>1:
              self.pad = ops.Paste(device=dali_device, ratio=pad_ratio, fill_value=0)
          else:
              self.pad = identity()
          self.cmnp = ops.CropMirrorNormalize(device="gpu",
                                              dtype=types.FLOAT,
                                              output_layout=types.NCHW,
                                              mean=self.mean,
                                              std=self.std
                                              )
          if ramdom_flip:
              self.coin = ops.random.CoinFlip(probability=0.5)
          else:
              self.coin = lambda :0
          if random_rotate_degree is not None:
              try:
                  tmp = math.abs(int(random_rotate_degree))
                  self.degree = ops.random.Uniform(range=(-tmp, tmp))
                  self.rotate = ops.Rotate()
              except:
                  self.degree = lambda :0
                  self.rotate = identity()
          else:
              self.degree = lambda :0
              self.rotate = identity()
          
      def iter_setup(self):
          if not self.use_file:
              (images, labels) = self.iterator.__next__()
              self.feed_input(self.jpegs, images, layout="HWC")
              self.feed_input(self.labels, labels)
  
      def define_graph(self):
          rng = self.coin()
          print()
          if self.use_file:
              self.jpegs,self.labels = self.input(name="Reader")
              self.jpegs = self.decode(self.jpegs)
          else:
              self.jpegs= self.input()
              self.labels = self.input_label()
          output = self.jpegs
          output = self.resize(output)
          output = self.rotate(output, angle=self.degree())
          output = self.pad(output.gpu())
          output = self.crop(output)
          output = self.cmnp(output, mirror=rng)
          return [output, self.labels]
  class Distribute_Input_Iter():
      def __init__(self,dataset, batch_size, num_replicas=None,rank=None,shuffle=True,seed=0,drop_last=False):
          if num_replicas is None:
              if not dist.is_available():
                  raise RuntimeError("Requires distributed package to be available")
              num_replicas = dist.get_world_size()
              #num_replicas = 1
          if rank is None:
              if not dist.is_available():
                  raise RuntimeError("Requires distributed package to be available")
              rank = dist.get_rank()
              #rank = 0
          if rank >= num_replicas or rank < 0:
              raise ValueError(
                  "Invalid rank {}, rank should be in the interval"
                  " [0, {}]".format(rank, num_replicas - 1))
          self.dataset = dataset
          self.batch_size = batch_size
          self.num_replicas = num_replicas
          self.rank = rank
          self.epoch = 0
          self.drop_last = drop_last
          
          # If the dataset length is evenly divisible by # of replicas, then there
          # is no need to drop any data, since the dataset will be split equally.
          if self.drop_last and len(self.dataset) % self.num_replicas != 0:  # type: ignore
              # Split to nearest available length that is evenly divisible.
              # This is to ensure each rank receives the same amount of data when
              # using this Sampler.
              self.num_samples = math.ceil(
                  # `type:ignore` is required because Dataset cannot provide a default __len__
                  # see NOTE in pytorch/torch/utils/data/sampler.py
                  (len(self.dataset) - self.num_replicas) / self.num_replicas  # type: ignore
              )
          else:
              self.num_samples = math.ceil(len(self.dataset) / self.num_replicas)  # type: ignore
          self.total_size = self.num_samples * self.num_replicas
          self.shuffle = shuffle
          self.seed = seed
          self.epoch=0
          indices = list(range(len(self.dataset)))  # type: ignore
  
          if not self.drop_last:
              # add extra samples to make it evenly divisible
              padding_size = self.total_size - len(indices)
              if padding_size <= len(indices):
                  indices += indices[:padding_size]
              else:
                  indices += (indices * math.ceil(padding_size / len(indices)))[:padding_size]
          else:
              # remove tail of data to make it evenly divisible.
              indices = indices[:self.total_size]
          assert len(indices) == self.total_size,'len(indices) != self.total_size'
  
          # subsample
          indices = indices[self.rank:self.total_size:self.num_replicas]
          assert len(indices) == self.num_samples,'len(indices) != self.num_samples'
          self.indices = indices
      def set_epoch(self,epoch):
          self.epoch = epoch
      def __iter__(self):
          self.i = 0
          self.n = len(self.indices)
          return self
      def __next__(self):
          batch = []
          labels = []
          should_shuffle = False
          
          for _ in range(self.batch_size):
              if self.i % self.n == self.n-1:
                  should_shuffle = True
              img, label = self.dataset.__getitem__(self.indices[self.i])
              batch.append(img)
              labels.append(label)
              self.i = (self.i + 1) % self.n
          if should_shuffle:
              random.shuffle(self.indices)
          return (batch, labels)
  class Normal_Input_Iter():
      def __init__(self,dataset, batch_size):
          self.dataset = dataset
          self.batch_size = batch_size
          self.indices = list(range(len(self.dataset)))
      def __iter__(self):
          self.i = 0
          self.n = len(self.dataset)
          return self
      def __next__(self):
          batch = []
          labels = []
          should_shuffle = False
          
          for _ in range(self.batch_size):
              if self.i % self.n == self.n-1:
                  should_shuffle = True
              img, label = self.dataset.__getitem__(self.indices[self.i])
              batch.append(img)
              labels.append(label)
              self.i = (self.i + 1) % self.n
          if should_shuffle:
              random.shuffle(self.indices)
          return (batch, labels)
answer: When you compare the two implementations, make sure to leave out as many changing variables as possible. For example, since you train with DDP, run it only on 2 GPUs so that you can be sure it's not bottlenecked by CPU. I don't know the Dali data loader very well, but I doubt that they can guarantee a throughput increase for all use cases.