@mrshenli so i debugged the code and found one bug that i was not returning the wrapped model to my parent function.
so after these changes DDP and DP take similar time ~ 1.6 seconds per iteration but still DDP is not faster than DP in my case.
.....
if config.distributed_training.enable:
     logging.info(f"spawning multiprocesses with {config.distributed_training.num_gpus} gpus")
     multiprocessing.spawn(  # type: ignore
           _train_model,
           nprocs=config.distributed_training.num_gpus,
           args=(pretrained_model, config, train_dataset, output_dir),
      )

def _train_model(
    gpu_index: int, pretrained_model: str, config: CfgNode, train_dataset: Dataset, output_dir: Path
) -> nn.Module:  **# this was returning None before**
    train_sampler = None
    world_size = _get_world_size(config)
    local_rank = gpu_index
    device = None
    if config.distributed_training.enable:
        local_rank = _setup_distributed_process(gpu_index, world_size, config)

        train_sampler = torch.utils.data.DistributedSampler(
            train_dataset, num_replicas=world_size, rank=local_rank
        )
        device = torch.device(local_rank)
    else:
        device = torch.device("cuda")

    model = MyModel(config) 
    model = _transfer_model_to_device(model, local_rank, gpu_index, config) # **now i assing to the model** 
    dataloader = _initialize_data_loader(train_dataset, config, train_sampler)
    _execute_training(config, device, model, train_sampler, dataloader, output_dir)

def _execute_training(
    config: CfgNode,
    device: torch.device,
    model: nn.Module,
    train_sampler: Optional[torch.utils.data.DistributedSampler],
    dataloader: DataLoader,
    output_dir: Path,
) -> None:
    network_config = config.network_config
    loss_combiner = MultiTaskLoss(num_tasks=2).to(device)
    trainable_params_model = list(filter(lambda p: p.requires_grad, model.parameters()))
    trainable_params = trainable_params_model + list(loss_combiner.parameters())
    optimizer = optim.Adam(
        trainable_params, lr=network_config.learning_rate, betas=(0.9, 0.99), weight_decay=0.001
    )
    logging.info(f"logging to tensorboard at {DEFAULT_TENSORBOARD_LOG_LOCATION}")
    with SummaryWriter(DEFAULT_TENSORBOARD_LOG_LOCATION) as summary_writer:  # type: ignore
        for epoch_idx in range(config.network_config.epochs):
            if config.distributed_training.enable and train_sampler is not None:
                train_sampler.set_epoch(epoch_idx)
            logging.info(f"starting epoch {epoch_idx}")
            _adjust_learning_rate(optimizer, epoch_idx, network_config.learning_rate, network_config.lrepochs)
             #**This is the entry point to the rest of the agnostic code that is same for both DP and DDP**
            _train_batches(
                epoch_idx, dataloader, model, optimizer, device, config, loss_combiner, summary_writer
            )


def _initialize_data_loader(
    train_dataset: Dataset, config: CfgNode, train_sampler: Optional[torch.utils.data.DistributedSampler]
) -> DataLoader:
    network_config = config.network_config
    dataloader = DataLoader(
        train_dataset,
        network_config.batch_size,
        collate_fn=custom_collate,
        shuffle=network_config.training_data_shuffle,
        num_workers=network_config.data_loader_num_workers,
        drop_last=network_config.drop_last_training_sample,
        pin_memory=network_config.data_loader_pin_memory,
        sampler=train_sampler,
    )
    return dataloader