I found a feasible solution. Because pymongodb is not fork-safe, we must be careful about fork() when using MongoClient instances.  In particular, MongoClient instances must never be copied from the parent process to the child process. Instead, the parent process and each child process must create their own MongoClient instance.
here is my code. This does work, when num_workers > 0.
import torch
from torch.utils.data import Dataset, DataLoader
import pymongo
from pymongo import MongoClient

client = MongoClient('localhost', 27017, connect=False)
db = client.dataset
collection = db.train

class MyDataset(Dataset):
    def __init__(self, trian=True):
        self.trian = trian
        
    def __getitem__(self, idx):
        if self.train:
            data_i = collection.find_one({'index': idx}) 
        else:
            data_i = collection.find_one({'index': idx}) 
        
        return data_i 
        
    def __len__(self):
        return collection.estimated_document_count()  

def collate_fn(recs): 
    forward = list(map(lambda x: x['forward'], recs)) 
    backward = list(map(lambda x: x['backward'], recs)) 

    def to_tensor_dict(recs):
        values = torch.FloatTensor(list(map(lambda r: r['values'], recs)))
        masks = torch.FloatTensor(list(map(lambda r: r['masks'], recs)))
        deltas = torch.FloatTensor(list(map(lambda r: r['deltas'], recs)))

        evals = torch.FloatTensor(list(map(lambda r: r['evals'], recs)))
        eval_masks = torch.FloatTensor(list(map(lambda r: r['eval_masks'], recs)))
        forwards = torch.FloatTensor(list(map(lambda r: r['forwards'], recs)))

        return {'values': values, 'forwards': forwards, 'masks': masks, 'deltas': deltas,
                'evals': evals, 'eval_masks': eval_masks}

    # transform recs to tensor dict
    ret_dict = {'forward': to_tensor_dict(forward), 'backward': to_tensor_dict(backward)}
    ret_dict['labels'] = torch.FloatTensor(list(map(lambda x: x['label'], recs))) 
    ret_dict['is_train'] = torch.FloatTensor(list(map(lambda x: x['is_train'], recs)))

    return ret_dict

def get_loader(batch_size = 64, shuffle = True):
    data_set = MyDataset()
    data_iter = DataLoader(dataset = data_set,
                           batch_size = batch_size,
                           num_workers = 4,
                           shuffle = shuffle,
                           pin_memory = True,
                           collate_fn = collate_fn )

    return data_iter

if __name__ == '__main__':
    data_loader = get_loader(batch_size=32, shuffle = True)
    for idx, data in enumerate(data_loader):
        print(idx)