I have narrowed it down! I found that the dataset object is being assigned a class method from a nn.Module class. Thus the solution is to not assign a module’s class method to the dataset. Here is a code snippet that reproduces the problem. NOTE that this problem does not occur if DDP is not used.
I would like to have a high level understanding of this issue. Why would assigning a class method to the dataset cause the worker processes to have CUDA context? Additionally, why is this only occurring when using the DDP module?
import torch.multiprocessing as mp
import torch.distributed as dist
import torch.nn as nn
import torchvision
import os

from torch.nn.parallel import DistributedDataParallel as DDP
from torchvision.datasets import MNIST
from torch.utils.data import DataLoader
import time

class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = nn.Linear(10, 10)

    def forward(self, x):
        return x
    
    def preprocess(self, batch):
        return batch


def main():
    
    mnist_dataset = MNIST(
        'mnist', train=True, download=True,
        transform=torchvision.transforms.ToTensor()
    )
    model = SimpleModel()
    is_parallel = True

    if is_parallel:
        mp.spawn(train_wrapper, nprocs=2, join=True,
                 args=(model, mnist_dataset))
    else:
        train(model, mnist_dataset, False, 'cuda:1')


def train_wrapper(rank, model, train_data):
    os.environ['MASTER_ADDR'] = '127.0.0.1'
    os.environ['MASTER_PORT'] = '12345'

    devices = ['cuda:1', 'cuda:2']

    dist.init_process_group(backend='nccl', rank=rank, world_size=len(devices))
    train(model, train_data, True, devices[rank])
    dist.destroy_process_group()


def train(model, train_data, is_parallel, device):


    # NOTE: THIS IS THE PROBLEM LINE
    # If you comment this line out, the issue no longer persists
    train_data.preprocess = model.preprocess
    
    train_loader = DataLoader(
        train_data,
        num_workers=4,
        batch_size=16
    )

    model = model.to(device)

    if is_parallel:
        model = DDP(model, device_ids=[device])
    
    x = iter(train_loader)
    time.sleep(10)

if __name__ == "__main__":
    main()