Dask-cuda: Usage with distributed pytorch + NCCL

[vfdev-5]

I'm trying to use LocalCUDACluster with pytorch (1.7.0) native distributed framework with NCCL (2.7.8) as communication backend. I have 2 GPUs and I spawn 2 workers and initialize a processing group. However, NCCL complains that

user:568:606 [1] init.cc:573 NCCL WARN Duplicate GPU detected : rank 1 and rank 0 both on CUDA device 3000
user:567:605 [0] init.cc:573 NCCL WARN Duplicate GPU detected : rank 0 and rank 1 both on CUDA device 3000 

and processing group initialization is failing.

Here is a minimal snippet to reproduce:

# test_bug.py

def run():

    import torch
    import dask
    from dask.distributed import Client
    from dask_cuda import LocalCUDACluster

    dask.config.set({"distributed.worker.daemon": False})
    # dask.config.set({"distributed.scheduler.work-stealing": False})

    cluster = LocalCUDACluster()
    client = Client(cluster)

    workers_info = client.scheduler_info().get("workers")
    master_addr = "localhost"
    master_port = 2345
    world_size = len(workers_info)

    print("world_size: ", world_size)

    futures = [
        client.submit(
            dask_spawner,
            i=i,
            master_addr=master_addr,
            master_port=master_port,
            world_size=world_size,
            workers_info=workers_info
        )
        for i in range(world_size)
    ]
    client.gather(futures)


def dask_spawner(i, master_addr, master_port, world_size, workers_info):

    import os
    import torch.distributed as dist

    from dask.distributed import get_worker

    worker = get_worker()
    worker_address = worker.address
    rank = workers_info[worker_address].get("id")
    local_rank = rank

    init_method = f"tcp://{master_addr}:{master_port}"
    dist.init_process_group("nccl", init_method=init_method, rank=rank, world_size=world_size)
    dist.barrier()
    torch.rand(10, 3, 64, 64, device="cuda")
    dist.destroy_process_group()    

if __name__ == "__main__":
    run()

Run it like NCCL_DEBUG=INFO NCCL_DEBUG_SUBSYS=COLL python test_bug.py

Versions:

torch                  1.8.0.dev20201117
torchvision            0.9.0.dev20201125
dask                   2.30.0
dask-cuda              0.16.0

Any hints on this, please ?

EDIT:

The above code can work (does not raise an error) with lower version of pytorch/nccl (1.6.0 and ~2.4.x) or if changing the backend from "nccl" to "gloo". Maybe, gloo is more permissive.

[pentschev]

According to the question in https://github.com/horovod/horovod/issues/1862#issuecomment-612220880 and the answer just below that, NCCL needs exclusive GPU access. When you start a LocalCUDACluster, GPU 0 will be used by the client and one worker (in some situations even the scheduler may access that GPU), which means NCCL won't have exclusive access to the GPU. What you could try is to start LocalCUDACluster(CUDA_VISIBLE_DEVICES=[1,2,3,...,N]) to force that the GPU 0 won't be used by anyone other than the client itself. I know this isn't a good solution, but at least would allow us to verify this is indeed the case so we can think of a better way to solve this in the future.

I'm not familiar with PyTorch, but there's also a chance that it will spawn its own processes too. If that's indeed the case you'll have one process for a Dask worker and a process for PyTorch, and that would lead to the same problem. This is mostly speculation by my part, so take it with a grain of salt and be aware of the possible pitfalls here.

[vfdev-5]

@pentschev thanks for the answer. Yes, that's correct about NCCL and its exclusive need of GPU access per process. And in the default way of performing distributed computations with horovod or native pytorch launcher this is normally respected.

When you start a LocalCUDACluster, GPU 0 will be used by the client and one worker (in some situations even the scheduler may access that GPU), which means NCCL won't have exclusive access to the GPU. What you could try is to start LocalCUDACluster(CUDA_VISIBLE_DEVICES=[1,2,3,...,N]) to force that the GPU 0 won't be used by anyone other than the client itself.

This is interesting, thanks for pointing that out. As for testing your idea, unfortunately I have only 2 GPUs to test on. Probably, I can not verify NCCL that way if I reserve one GPU for the client and another one for the worker.

Is there a way to restrict the client from accessing the GPUs and give the access to GPUs to workers only ? I tried naively the following:

CUDA_VISIBLE_DEVICES="" NCCL_DEBUG=INFO NCCL_DEBUG_SUBSYS=COLL python test_bug.py

and

cluster = LocalCUDACluster(CUDA_VISIBLE_DEVICES=[0, 1])

but this does not work neither (numba.cuda.cudadrv.driver.CudaAPIError: [100] Call to cuInit results in CUDA_ERROR_NO_DEVICE etc).

I'm not familiar with PyTorch, but there's also a chance that it will spawn its own processes too. If that's indeed the case you'll have one process for a Dask worker and a process for PyTorch, and that would lead to the same problem. This is mostly speculation by my part, so take it with a grain of salt and be aware of the possible pitfalls here.

Normally, process spawning is handled by the user. Can be done in 2 ways:

• single parent script which uses multiprocessing function to spawn children processes
• a launcher tool (like horovodrun) which spawns children processes that run users code.

Context: what I'm trying to do is to use dask client as a process spawner instead of for example pytorch native one.

[pentschev]

Is there a way to restrict the client from accessing the GPUs and give the access to GPUs to workers only ?

I _think_ you'll be fine as long as the client itself doesn't need to do any GPU computation. One such case would be when calling dask.compute() or dask_array.compute(), which will bring the data back to the client, that will certainly fail.

... but this does not work neither (numba.cuda.cudadrv.driver.CudaAPIError: [100] Call to cuInit results in CUDA_ERROR_NO_DEVICE etc).

In your case, it's probably safe to ignore that error, it comes from https://github.com/rapidsai/dask-cuda/blob/a27bd93aba8ad2d32003984ffa0c14fbf65e319f/dask_cuda/local_cuda_cluster.py#L227-L234, which initializes the CUDA context and other UCX configurations on the parent process (where the client sits), and that's important for "regular" workflows (where the client may need access to the GPU) and UCX, since you're not using any of them you could even comment that out for your testing now to get rid of errors.

Normally, process spawning is handled by the user. Can be done in 2 ways:

• single parent script which uses multiprocessing function to spawn children processes
• a launcher tool (like horovodrun) which spawns children processes that run users code.

Context: what I'm trying to do is to use dask client as a process spawner instead of for example pytorch native one.

It seems to me that this would work as long as your client really doesn't need any access to the GPUs, if you do then we'll still have the same NCCL issues unless you have a dedicated GPU for it.

[vfdev-5]

Thanks for the details, @pentschev ! I understand better the context of regular usage of dask cuda. Yes, currently, I'm exploring a workflow where the client can be considered as a simple scheduler without gpu or data access (i.e. does nothing except scheduling workers). Data access and sending it to GPUs is done only on workers without sending back any data to the client. Maybe, a question about dask usage, do you think client.submit method is appropriate for this use-case ?

[pentschev]

To be honest, I don't know what would be the appropriate way to deal with that, but using client.submit seems like a good starting point. In case you haven't seem those yet, I think the following links may provide you with some additional inspiration too:

https://docs.dask.org/en/latest/gpu.html
https://ml.dask.org/pytorch.html
https://examples.dask.org/machine-learning/torch-prediction.html

[pentschev]

I think the original question has been answered, I'm tentatively closing this, @vfdev-5 please feel free to reopen if there are follow-up questions or open a new issue should you want to discuss other ways to use Dask-CUDA with PyTorch.