mapanything/train/profile_dataloading.py

"""
Debug script to profile dataloading for MapAnything training.

This script measures and analyzes the performance of data loading operations
for MapAnything training workflows. It simulates the training process without
actual model training to isolate and profile the data loading components.
"""

import datetime
import json
import os
import time
from pathlib import Path
from typing import Sized

import numpy as np
import torch
import torch.backends.cudnn as cudnn
from torch.utils.tensorboard import SummaryWriter

import mapanything.utils.train_tools as train_tools
from mapanything.datasets import get_test_data_loader, get_train_data_loader
from mapanything.datasets.base.base_dataset import view_name

# Enable TF32 precision if supported (for GPU >= Ampere and PyTorch >= 1.12)
if hasattr(torch.backends.cuda, "matmul") and hasattr(
    torch.backends.cuda.matmul, "allow_tf32"
):
    torch.backends.cuda.matmul.allow_tf32 = True


def profile_dataloading(args):
    """
    Main profiling function that simulates the training process to measure data loading performance.

    This function initializes the distributed environment, sets up datasets and data loaders,
    and runs through training epochs to profile the data loading operations. It measures
    the time taken for data loading without performing actual model training or optimization.

    In this simulation, an epoch represents a complete pass through a chunk of the dataset.

    Args:
        args: Configuration object containing all parameters including:
            - dataset: Dataset configuration (train_dataset, test_dataset, num_workers)
            - train_params: Training parameters (batch_size, epochs, seed, etc.)
            - distributed: Distributed training configuration
            - output_dir: Directory for saving logs and profiling results
    """
    # Initialize distributed training if required
    train_tools.init_distributed_mode(args.distributed)
    global_rank = train_tools.get_rank()
    world_size = train_tools.get_world_size()  # noqa

    # Init output directory and device
    print("output_dir: " + args.output_dir)
    if args.output_dir:
        Path(args.output_dir).mkdir(parents=True, exist_ok=True)

    print("job dir: {}".format(os.path.dirname(os.path.realpath(__file__))))
    print("{}".format(args).replace(", ", ",\n"))

    device = "cuda" if torch.cuda.is_available() else "cpu"
    device = torch.device(device)

    # Fix the seed
    seed = args.train_params.seed + train_tools.get_rank()
    torch.manual_seed(seed)
    np.random.seed(seed)

    cudnn.benchmark = not args.train_params.disable_cudnn_benchmark

    # Datasets and Dataloaders
    print("Building train dataset {:s}".format(args.dataset.train_dataset))
    data_loader_train = build_dataset(
        dataset=args.dataset.train_dataset,
        num_workers=args.dataset.num_workers,
        test=False,
        max_num_of_imgs_per_gpu=args.train_params.max_num_of_imgs_per_gpu,
    )
    print("Building test dataset {:s}".format(args.dataset.test_dataset))
    test_batch_size = 2 * (
        args.train_params.max_num_of_imgs_per_gpu // args.dataset.num_views
    )  # Since we don't have any backward overhead
    data_loader_test = {
        dataset.split("(")[0]: build_dataset(
            dataset=dataset,
            num_workers=args.dataset.num_workers,
            test=True,
            batch_size=test_batch_size,
        )
        for dataset in args.dataset.test_dataset.split("+")
        if "(" in dataset
    }

    def write_log_stats(epoch, train_stats, test_stats):
        """
        Writes profiling statistics to log files and TensorBoard.

        This function collects metrics from the training and testing phases and writes them
        to log files and TensorBoard for visualization and analysis. It only executes on the
        main process in a distributed setting.

        Args:
            epoch: int, current epoch number
            train_stats: dict, containing training metrics and timing information
            test_stats: dict, containing testing metrics for each test dataset
        """
        if train_tools.is_main_process():
            if log_writer is not None:
                log_writer.flush()

            log_stats = dict(
                epoch=epoch, **{f"train_{k}": v for k, v in train_stats.items()}
            )
            for test_name in data_loader_test:
                if test_name not in test_stats:
                    continue
                log_stats.update(
                    {test_name + "_" + k: v for k, v in test_stats[test_name].items()}
                )

            with open(
                os.path.join(args.output_dir, "log.txt"), mode="a", encoding="utf-8"
            ) as f:
                f.write(json.dumps(log_stats) + "\n")

    if global_rank == 0 and args.output_dir is not None:
        log_writer = SummaryWriter(log_dir=args.output_dir)
    else:
        log_writer = None

    print(f"Start training for {args.train_params.epochs} epochs")
    start_time = time.time()
    train_stats = test_stats = {}
    args.train_params.start_epoch = 0
    for epoch in range(args.train_params.start_epoch, args.train_params.epochs + 1):
        # Save more stuff
        write_log_stats(epoch, train_stats, test_stats)

        if epoch >= args.train_params.epochs:
            break  # exit after writing last test to disk

        # Train
        train_stats = train_one_epoch(
            data_loader_train,
            device,
            epoch,
            log_writer=log_writer,
            args=args,
        )

    total_time = time.time() - start_time
    total_time_str = str(datetime.timedelta(seconds=int(total_time)))
    print("Training time {}".format(total_time_str))


def build_dataset(
    dataset, num_workers, test, batch_size=None, max_num_of_imgs_per_gpu=None
):
    """
    Builds data loaders for training or testing.

    Args:
        dataset: Dataset specification string.
        num_workers: Number of worker processes for data loading.
        test: Boolean flag indicating whether this is a test dataset.
        batch_size: Number of samples per batch. Defaults to None. Used only for testing.
        max_num_of_imgs_per_gpu: Maximum number of images per GPU. Defaults to None. Used only for training.

    Returns:
        DataLoader: PyTorch DataLoader configured for the specified dataset.
    """
    split = ["Train", "Test"][test]
    print(f"Building {split} Data loader for dataset: ", dataset)
    if test:
        assert batch_size is not None, (
            "batch_size must be specified for testing dataloader"
        )
        loader = get_test_data_loader(
            dataset=dataset,
            batch_size=batch_size,
            num_workers=num_workers,
            pin_mem=True,
            shuffle=False,
            drop_last=False,
        )
    else:
        assert max_num_of_imgs_per_gpu is not None, (
            "max_num_of_imgs_per_gpu must be specified for training dataloader"
        )
        loader = get_train_data_loader(
            dataset=dataset,
            max_num_of_imgs_per_gpu=max_num_of_imgs_per_gpu,
            num_workers=num_workers,
            pin_mem=True,
            shuffle=True,
            drop_last=True,
        )

    print(f"{split} dataset length: ", len(loader))
    return loader


def train_one_epoch(
    data_loader: Sized,
    device: torch.device,
    epoch: int,
    args,
    log_writer=None,
):
    """
    Simulates training for one epoch to profile data loading performance.

    This function runs through a single epoch, simulating the data loading and device transfer
    operations that would occur during actual training. It measures and logs the time taken
    for these operations without performing actual model training.

    Args:
        data_loader: Sized, DataLoader providing the training data
        device: torch.device, device to transfer data to (CPU or GPU)
        epoch: int, current epoch number
        args: object, configuration object containing training parameters including:
            - train_params.print_freq: frequency of logging during the epoch
        log_writer: Optional[SummaryWriter], TensorBoard SummaryWriter for logging metrics

    Returns:
        dict: Dictionary containing profiling metrics averaged over the epoch
    """
    metric_logger = train_tools.MetricLogger(delimiter="  ")
    header = "Epoch: [{}]".format(epoch)

    if log_writer is not None:
        print("log_dir: {}".format(log_writer.log_dir))

    if hasattr(data_loader, "dataset") and hasattr(data_loader.dataset, "set_epoch"):
        data_loader.dataset.set_epoch(epoch)
    if hasattr(data_loader, "sampler") and hasattr(data_loader.sampler, "set_epoch"):
        data_loader.sampler.set_epoch(epoch)
    if hasattr(data_loader, "batch_sampler") and hasattr(
        data_loader.batch_sampler, "set_epoch"
    ):
        data_loader.batch_sampler.set_epoch(epoch)

    for data_iter_step, batch in enumerate(
        metric_logger.log_every(data_loader, args.train_params.print_freq, header)
    ):
        epoch_f = epoch + data_iter_step / len(data_loader)

        # Simulate the device loading in loss_of_one_batch_multi_view
        ignore_keys = set(
            [
                "depthmap",
                "dataset",
                "label",
                "instance",
                "idx",
                "true_shape",
                "rng",
                "data_norm_type",
            ]
        )
        for view in batch:
            for name in view.keys():
                if name in ignore_keys:
                    continue
                view[name] = view[name].to(device, non_blocking=True)

        local_rank = train_tools.get_rank()
        n_views = len(batch)
        batch_shape = batch[0]["img"].shape
        first_sample_name = view_name(batch[0], batch_index=0)
        print(
            f"Rank: {local_rank}, Num views: {n_views}, Batch Shape: {batch_shape}, First Sample Name: {first_sample_name}",
            force=True,
        )

        del batch

        metric_logger.update(epoch=epoch_f)
        metric_logger.update(loss=0)

    # # Gather the stats from all processes
    # metric_logger.synchronize_between_processes()
    # print("Averaged stats:", metric_logger)
    return {k: meter.global_avg for k, meter in metric_logger.meters.items()}