Why is my gpu-util low?

[yllgl]

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)

[awaelchli]

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.

[awaelchli]

You could compare the two implementations using our profilers:
https://pytorch-lightning.readthedocs.io/en/latest/advanced/profiler.html#performance-and-bottleneck-profiler
Maybe this can help you find where the bottleneck is.

[yllgl]

How to read the profile report? I'm not sure how the actions relate to each other, which actions add up to the total time?

Profiler Report

Action                             	|  Mean duration (s)	|Num calls      	|  Total time (s) 	|  Percentage %   	|
------------------------------------------------------------------------------------------------------------------------------------
Total                              	|  -              	|_              	|  193.0          	|  100 %          	|
------------------------------------------------------------------------------------------------------------------------------------
run_training_epoch                 	|  89.129         	|2              	|  178.26         	|  92.362         	|
run_training_batch                 	|  0.12825        	|782            	|  100.29         	|  51.963         	|
optimizer_step_and_closure_0       	|  0.038406       	|782            	|  30.034         	|  15.562         	|
training_step_and_backward         	|  0.037109       	|782            	|  29.019         	|  15.036         	|
model_forward                      	|  0.023912       	|782            	|  18.699         	|  9.6886         	|
training_step                      	|  0.023513       	|782            	|  18.387         	|  9.5271         	|
model_backward                     	|  0.011458       	|782            	|  8.9599         	|  4.6424         	|
evaluation_step_and_end            	|  0.056162       	|49             	|  2.7519         	|  1.4259         	|
validation_step                    	|  0.055992       	|49             	|  2.7436         	|  1.4216         	|
get_train_batch                    	|  0.0022135      	|782            	|  1.7309         	|  0.89685        	|
on_train_batch_end                 	|  0.00065149     	|782            	|  0.50947        	|  0.26397        	|
on_validation_end                  	|  0.13549        	|3              	|  0.40648        	|  0.21061        	|
cache_result                       	|  2.4877e-05     	|4089           	|  0.10172        	|  0.052707       	|
on_train_end                       	|  0.096949       	|1              	|  0.096949       	|  0.050233       	|
on_batch_start                     	|  5.2834e-05     	|782            	|  0.041316       	|  0.021407       	|
on_before_zero_grad                	|  3.0646e-05     	|782            	|  0.023965       	|  0.012417       	|
on_train_start                     	|  0.023235       	|1              	|  0.023235       	|  0.012039       	|
on_after_backward                  	|  2.6336e-05     	|782            	|  0.020595       	|  0.010671       	|
on_batch_end                       	|  2.0109e-05     	|782            	|  0.015725       	|  0.0081477      	|
on_train_batch_start               	|  1.9767e-05     	|782            	|  0.015458       	|  0.0080094      	|
training_step_end                  	|  1.4332e-05     	|782            	|  0.011207       	|  0.0058069      	|
on_validation_batch_end            	|  0.00018954     	|49             	|  0.0092876      	|  0.0048122      	|
on_validation_start                	|  0.00036286     	|3              	|  0.0010886      	|  0.00056404     	|
on_validation_batch_start          	|  1.9223e-05     	|49             	|  0.00094193     	|  0.00048805     	|
on_train_epoch_start               	|  0.00031408     	|2              	|  0.00062815     	|  0.00032547     	|
validation_step_end                	|  1.1229e-05     	|49             	|  0.00055024     	|  0.0002851      	|
on_epoch_start                     	|  2.5055e-05     	|5              	|  0.00012527     	|  6.4909e-05     	|
on_validation_epoch_end            	|  3.433e-05      	|3              	|  0.00010299     	|  5.3363e-05     	|
on_epoch_end                       	|  1.9623e-05     	|5              	|  9.8117e-05     	|  5.0838e-05     	|
on_train_epoch_end                 	|  4.749e-05      	|2              	|  9.498e-05      	|  4.9213e-05     	|
on_validation_epoch_start          	|  1.96e-05       	|3              	|  5.88e-05       	|  3.0466e-05     	|
on_fit_start                       	|  1.9558e-05     	|1              	|  1.9558e-05     	|  1.0134e-05     	|
on_before_accelerator_backend_setup	|  1.5616e-05     	|1              	|  1.5616e-05     	|  8.0914e-06     	|

[ndalton12]

You probably don't need to write your own distributed sampling procedure since dali and pytorch lightning can do it for you: https://docs.nvidia.com/deeplearning/dali/user-guide/docs/examples/frameworks/pytorch/pytorch-lightning.html. Also, try increasing the number of threads provided to the pipeline from 1 to maybe >12? You also should re-run the profiler with the regular dataloader version so we can compare to this profile.

[yllgl]

Could you tell me how pytorch-lightning processes iterable data_loader when using multiple gpus on slurm? Does it run dataloader in each process? In my case, I originally think, there are four processes, and each process get one GPU and run their dataloader. And in my dataloader, I address distributed sampler and print dataloader's device_id. Here is my code:

device_id = torch.cuda.current_device()
print("device_id",device_id)#print dataloader's device_id
super(HybridPipe, self).__init__(batch_size, num_threads, device_id, seed=12 + device_id)
self.iterator = iter(Distribute_Input_Iter(dataset, batch_size))# process sampler and get data_iterator

however, my output is like this:

Multi-processing is handled by Slurm.
LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0,1,2,3]
LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0,1,2,3]                                                                               LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0,1,2,3]                                                                               initializing ddp: GLOBAL_RANK: 1, MEMBER: 2/4
initializing ddp: GLOBAL_RANK: 0, MEMBER: 1/4
initializing ddp: GLOBAL_RANK: 2, MEMBER: 3/4
initializing ddp: GLOBAL_RANK: 3, MEMBER: 4/4

  | Name   | Type        | Params
---------------------------------------                                                                                       0 | conv1  | Conv2d      | 1.7 K                                                                                              1 | bn1    | BatchNorm2d | 128
2 | layer1 | Sequential  | 147 K
3 | layer2 | Sequential  | 525 K
4 | layer3 | Sequential  | 2.1 M
5 | layer4 | Sequential  | 8.4 M
6 | linear | Linear      | 5.1 K
---------------------------------------                                                                                       11.2 M    Trainable params                                                                                                    0         Non-trainable params
11.2 M    Total params
44.696    Total estimated model params size (MB)
device_id 0                                                                                                                                                                                                                                                 device_id 0

Epoch 1: 100%|██████████| 123/123 [00:58<00:00,  2.10it/s, loss=1.16, v_num=8, train_loss=1.010, train_accuracy=0.524]
device_id 2
device_id 2

device_id 1
device_id 1

device_id 3
device_id 3

As you can see, only if training ends it outputs the device_id 1-3. It seems that there is no parallel in the dataloader. I don't know why. Could you give me some advice?

[awaelchli]

Could you tell me how pytorch-lightning processes iterable data_loader when using multiple gpus on slurm?

A DistributedSampler is used. If you want to avoid that Lightning automatically adds that, you can set Trainer(replace_sampler_ddp=False) (true is the default). Then you can do your own sampling if you want.