CacheDataset crashes with runtime_cache=True, num_workers=0 in DDP

[myron]

CacheDataset crashes with runtime_cache=True, num_workers=0 in DDP

This is because we try to convert a shared list (ListProxy) to a regular list, but must maintain a ListProxy since we're in DDP, and need to access the shared items from different processes.

the exact error is this , that is unable to convert. But even if we can convert, we should not do in DDP, since need a ListProxy

EDIT: expected behavior below:

• it should not crash
• we should not call disable_share_memory_cache in DataLoader, if in DDP and using runtime_cache, even for num_workers==0, because we still need to use ListProxy (it is needed for different processes in DDP to read/write to the same cache indices). if we convert ListProxy -> List in such case, we will get memory copies in each process (potentially going OOM).

Traceback (most recent call last):                                                                                                                              
  File "/usr/local/lib/python3.8/dist-packages/torch/multiprocessing/spawn.py", line 69, in _wrap                                                               
    fn(i, *args)                                                                                                                                                
  File "/scripts/segmenter.py", line 1032, in run_segmenter_worker                                       
    best_metric = segmenter.run()                                                                                                                               
  File "//scripts/segmenter.py", line 696, in run
    self.train()
  File "/scripts/segmenter.py", line 732, in train
    train_loader = DataLoader(train_ds, batch_size=config["batch_size"], shuffle=(train_sampler is None),  num_workers=config["num_workers"], sampler=train_sampler, pin_memory=True)
  File "/mnt/amproj/Code/MONAI/monai/data/dataloader.py", line 87, in __init__
    dataset.disable_share_memory_cache() 
  File "/MONAI/monai/data/dataset.py", line 855, in disable_share_memory_cache
    self._cache = list(self._cache)
  File "<string>", line 2, in __len__
  File "/usr/lib/python3.8/multiprocessing/managers.py", line 831, in _callmethod
    self._connect()
  File "/usr/lib/python3.8/multiprocessing/managers.py", line 818, in _connect
    conn = self._Client(self._token.address, authkey=self._authkey)
  File "/usr/lib/python3.8/multiprocessing/connection.py", line 502, in Client
    c = SocketClient(address)
  File "/usr/lib/python3.8/multiprocessing/connection.py", line 630, in SocketClient
    s.connect(address)
FileNotFoundError: [Errno 2] No such file or directory

[Nic-Ma]

Hi @myron ,

Could you please help share your test program or some simple program to reproduce the error?
Maybe it's not related to the ListProxy -> list logic, but related to missing barrier somewhere? I see the same error in our unit tests if deleted this line:
https://github.com/Project-MONAI/MONAI/blob/dev/tests/test_sampler_dist.py#L67

Thanks in advance.

[myron]

@Nic-Ma
Here is a test program. It crashes due to disable_share_memory_cache()
Replacing ListProxy with List in another way is not a solution here, , since we actually need ListProxy here (always need it in multi-gpu)

import torch
import torch.distributed as dist
from monai.data import CacheDataset, DataLoader, Dataset
from monai.transforms import Transform, Compose
from monai.data.utils import no_collation
import numpy as np

class SimpleTransform(Transform):
    def __call__(self, data):
        print('SimpleTransform called for data', data)
        return data 

def main_worker(rank, nprocs):

    dist.init_process_group(backend='nccl', init_method="tcp://127.0.0.1:12346", world_size=nprocs, rank=rank)
    torch.cuda.set_device(rank)

    data = [rank+10,] * nprocs # each process has data with same length (nprocs) but a different numbers rank0 = [10,10,10,..], rank1=[11,11,11,11..]
    transform = Compose([SimpleTransform()])
    dataset = CacheDataset(data=data, transform=transform, runtime_cache=True, copy_cache=False, num_workers=0)

    dataset.disable_share_memory_cache() #simulate disabling of shared memory  (we actually need shared cache in multi-gpu)

    item = dataset[rank] # each process access 1 item (at the position corresponding to its rank) to cache it
    torch.distributed.barrier()

    # at this point each process cached one item (so all items are in cache)
    # and the cached dataset should be [10,11,12,13...]
    cache = [item for item in dataset] # or cache = dataset._cache
    print('rank', rank, 'sees data', cache)

    expected = list(range(10, 10+nprocs))

   
    assert np.all(np.asarray(cache) == np.asarray(expected))

    torch.distributed.barrier()

    if dist.is_initialized():
        dist.destroy_process_group()

    print('Success')

def main():
    if torch.cuda.is_available() and torch.cuda.device_count() > 1:
        nprocs = torch.cuda.device_count()
        torch.multiprocessing.spawn(main_worker, nprocs=nprocs, args=(nprocs, ))

if __name__ == "__main__":
    main()

[wyli]

thanks for the script, I can reproduce the issue. In my opinion, the root cause is that the ListProxy to list conversion should be done in a synchronized manner. So the fix could be:

    def disable_share_memory_cache(self):
        if self._is_dist:
            torch.distributed.barrier()
        self._cache = list(self._cache)

what do you think? it's better than skipping the ListProxy to list conversion.

[myron]

I think the barrier is not issue. there are 2 issues

• we are unable convert to simple list with list(ListProxy). The objects inside of ListProxy may belong to other processes already , and accessing such converted list will lead to errors (i think that's why it crashes)
• even if we can find a way to convert ListProxy->list (and the code above don't crash), it will fail the assert line, since we still need to use shared cache (ListProxy) to be able to communicate between the processes (when num_workers==0)

[wyli]

• I didn't say that the list(ListProxy) is the issue. the issue is when some processes are doing the conversion, other processes are trying to simultaneously access the ListProxy. a barrier will address the issue.

• why is it needed to share the cache content in this case? a distributed sampler will divide data into subsets and each process will exclusively see a subset anyway.

submitted a PR to address this issue #5595

[myron]

a distributed sampler will divide data into subsets and each process will exclusively see a subset anyway.

@wyli This is true only during the first training epoch (when all data is cached initially). On the second iteration, distributed sampler will have a different random shuffle of all indices, and now each process has to access a different subset (which was previously cached by another process).

Here we do not pre-split the dataset into partitions, we use the full dataset, and all processes can read/write at any index. And the shared cache (ListProxy) allows different processes to read/write to it, this is the idea of using shared cache.

So the PR #5595 will just have incorrect logic (for our use case, where runtime_cache was introduced). Runtime_cache (1 option) enables 2 things (cache on the fly + shared cache). If we convert to a standard list, each process will start creating copies of the dataset.

so the main issue is here, we should not always be converting to a list, if num_workers==0 in DDP.
https://github.com/Project-MONAI/MONAI/blob/dev/monai/data/dataloader.py#L86

so we can either remove this line from Dataloader, or check if it's DDP condition

[wyli]

in your PR (#5565) and the current dev branch, these two features can't coexist with a single option:

1. gpu-cache + runtime-cache + ddp
2. cpu-cache + runtime-cache + ddp when num_workers==0

option2 is probably just for debugging as num_workers==1 is almost always better in terms of performance.

what I'm suggesting in #5595 is to enable both of them, with option 2 potentially has redundant, not shared cache.

anyway I don't feel it's very meaningful to discuss this further, please feel free to decide. @Nic-Ma

[myron]

@Nic-Ma you decide please. I only need the shared cache working with num_workers==0 in DDP.

[Nic-Ma]

Hi @myron , @wyli ,

Sorry for the late response, I was busy with other project.
I personally feel this sleep logic in @wyli 's PR is not very robust, the "0.5s" time is slightly hacking:

if torch.distributed.get_rank() == 0:
    time.sleep(0.5)  # rank 0 should exit after all the other ranks, as the cache was broadcast from rank 0

If we don't have any other better solution, I would vote for @myron 's PR to drop the gpu-cache + runtime-cache + ddp feature. We drop the feature but make the overall logic more robust?
What do you think?

Thanks.