NCCL error: Invalid rank requested

[loretoparisi]

Bug description

Torch distributed on-prem run error. The env was set on the master was

2024-08-20 18:18:44 WORLD_SIZE:8 NODE_RANK:0 MASTER_ADDR:192.168.1.1 MASTER_PORT:9001

and on the client was seeing the same master IP env with its node rank:

2024-08-20 18:18:WORLD_SIZE:8 NODE_RANK:1 MASTER_ADDR:92.168.1.1 MASTER_PORT:9001

SDK versions:

torch==2.4.0
pytorch-lightning==1.9.5

What version are you seeing the problem on?

v1.x

How to reproduce the bug

hparams = {"learning_rate": float(learning_rate),
               "margin": float(margin),
               "strategy": "batch all",
               "batch_size":  int(int(batch_size)//2),
               "epochs": int(epochs)}

    train_dataset = TensorDataset(torch.tensor(preprocess.pairs))
    trainloader = DataLoader(train_dataset,
                             batch_size = hparams["batch_size"],
                             collate_fn = my_collate,
                             drop_last = False,
                             shuffle=True,
                             num_workers=psutil.cpu_count(),
                             pin_memory=True)

if train_strategy=='ddp':
        train_strategy="ddp_find_unused_parameters_false"

    # Initialize a trainer
    trainer = pl.Trainer(logger=logger,
                        callbacks=[checkpoint_callback],
                         max_epochs=hparams["epochs"],
                         devices=devices,
                         accelerator=accelerator,
                         strategy=train_strategy)

    # Train the model
    log_message(f'training on {accelerator} with {num_of_gpus} gpu')
    trainer.fit(model, trainloader)

Error messages and logs

Initializing distributed: GLOBAL_RANK: 6, MEMBER: 7/4
2024-08-20 17:30:09 training on gpu with 4 gpu
Initializing distributed: GLOBAL_RANK: 7, MEMBER: 8/4
2024-08-20 17:30:09 training on gpu with 4 gpu
Initializing distributed: GLOBAL_RANK: 5, MEMBER: 6/4

and later the dlerror=libnccl-net.so: cannot open shared object file: No such file or directory No plugin found (libnccl-net.so), using internal implementation error:

[1] NCCL INFO cudaDriverVersion 12020
[2] NCCL INFO cudaDriverVersion 12020
 [3] NCCL INFO cudaDriverVersion 12020
 [2] NCCL INFO NET/Plugin : dlerror=libnccl-net.so: cannot open shared object file: No such file or directory No plugin found (libnccl-net.so), using internal implementation
 [1] NCCL INFO NET/Plugin : dlerror=libnccl-net.so: cannot open shared object file: No such file or directory No plugin found (libnccl-net.so), using internal implementation
 [3] NCCL INFO NET/Plugin : dlerror=libnccl-net.so: cannot open shared object file: No such file or directory No plugin found (libnccl-net.so), using internal implementation

 [2] init.cc:1726 NCCL WARN Invalid rank requested : 6/4
 [1] init.cc:1726 NCCL WARN Invalid rank requested : 5/4

 [3] init.cc:1726 NCCL WARN Invalid rank requested : 7/4
[2] NCCL INFO init.cc:1872 -> 4
[1] NCCL INFO init.cc:1872 -> 4
 [3] NCCL INFO init.cc:1872 -> 4
 [1] NCCL INFO init.cc:1876 -> 4
 [3] NCCL INFO init.cc:1876 -> 4
 [2] NCCL INFO init.cc:1876 -> 4
 [0] NCCL INFO cudaDriverVersion 12020
 [0] NCCL INFO Bootstrap : Using eth0:10.0.146.34<0>
 [0] NCCL INFO NET/Plugin : dlerror=libnccl-net.so: cannot open shared object file: No such file or directory No plugin found (libnccl-net.so), using internal implementation

 [0] init.cc:1726 NCCL WARN Invalid rank requested : 4/4
 [0] NCCL INFO init.cc:1872 -> 4
 [0] NCCL INFO init.cc:1876 -> 4
2024-08-20 17:30:09 NCCL error in: ../torch/csrc/distributed/c10d/NCCLUtils.hpp:275, invalid argument (run with NCCL_DEBUG=WARN for details), NCCL version 2.20.5
ncclInvalidArgument: Invalid value for an argument.

Stacktrace:

Last error:
Invalid rank requested : 7/4. traceback: Traceback (most recent call last):
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 608, in fit
    call._call_and_handle_interrupt(
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/call.py", line 38, in _call_and_handle_interrupt
    return trainer_fn(*args, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 650, in _fit_impl
    self._run(model, ckpt_path=self.ckpt_path)
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 1049, in _run
    self.__setup_profiler()
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 1509, in __setup_profiler
    self.profiler.setup(stage=self.state.fn, local_rank=local_rank, log_dir=self.log_dir)
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 1826, in log_dir
    dirpath = self.strategy.broadcast(dirpath)
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/strategies/ddp.py", line 314, in broadcast
    torch.distributed.broadcast_object_list(obj, src, group=_group.WORLD)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 79, in wrapper
    return func(*args, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/distributed_c10d.py", line 2901, in broadcast_object_list
    broadcast(object_sizes_tensor, src=src, group=group)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 79, in wrapper
    return func(*args, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/distributed_c10d.py", line 2205, in broadcast
    work = default_pg.broadcast([tensor], opts)

Environment

Current environment

#- PyTorch Lightning Version (e.g., 2.4.0): 1.9.5
#- PyTorch Version (e.g., 2.4): 2.4.0
#- Python version (e.g., 3.12): 3.8.10
#- OS (e.g., Linux): Ubuntu
#- CUDA/cuDNN version: 12.2
#- GPU models and configuration: 4x NVIDIA A10G
#- How you installed Lightning(`conda`, `pip`, source): pip

More info

No response

[loretoparisi]

[UPDATE]
According to the solution provided in #6454
I have added a ClusterEnvironment and DDPStrategy implementation:

class CustomEnvironment(ClusterEnvironment):
    def __init__(self, num_nodes=2):
        super().__init__()
        self._num_nodes = num_nodes
        self._master_port = None
        self._world_size = None
        self._global_rank = None

    def creates_processes_externally(self):
        # Assuming PyTorch Lightning manages processes internally
        return False

    def detect(self):
        # Implement detection of nodes and processes if necessary
        log.debug("Detect method is called.")

    def global_rank(self):
        if self._global_rank is None:
            self._global_rank = int(os.getenv("RANK", 0))
        log.debug(f"GLOBAL_RANK: {self._global_rank}")
        return self._global_rank

    @property
    def main_address(self):
        return self.master_address()

    @property
    def main_port(self):
        return self.master_port()

    def set_global_rank(self, rank: int):
        self._global_rank = rank
        log.debug(f"Set GLOBAL_RANK: {self._global_rank}")

    def set_world_size(self, world_size: int):
        self._world_size = world_size
        log.debug(f"Set WORLD_SIZE: {self._world_size}")

    def master_address(self):
        MASTER_ADDR = os.getenv("MASTER_ADDR")
        log.debug(f"MASTER_ADDR: {MASTER_ADDR}")
        return MASTER_ADDR

    def master_port(self):
        if self._master_port is None:
            self._master_port = os.getenv("MASTER_PORT")
        log.debug(f"MASTER_PORT: {self._master_port}")
        return int(self._master_port)

    def world_size(self):
        if self._world_size is None:
            log.debug("WORLD_SIZE is not set.")
        return self._world_size

    def node_rank(self):
        MY_RANK = int(os.getenv("NODE_RANK", "0"))
        log.debug(f"NODE_RANK: {MY_RANK}")
        return int(MY_RANK)

    def local_rank(self) -> int:
        LOCAL_RANK = int(os.getenv("LOCAL_RANK", "0"))
        log.debug(f"LOCAL_RANK: {LOCAL_RANK}")
        return LOCAL_RANK

and then used in the trainer as a train strategy:

   strategy = DDPStrategy(
            #accelerator="xpu",
            cluster_environment=CustomEnvironment(),
            #process_group_backend="ccl",
            find_unused_parameters=True
        )


    # Initialize a trainer
    trainer = pl.Trainer(logger=logger,
                        callbacks=[checkpoint_callback],
                         max_epochs=hparams["epochs"],
                         devices=devices,
                         accelerator=accelerator,
                         strategy=strategy)

Now the system apparently starts correctly but I have two issues

• Not sure the values of ranks are correct
• Timeout occurs

This is the output from the CustomEnvironment on the MASTER

DEBUG:train:NODE_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:Set GLOBAL_RANK: 0
DEBUG:train:Set WORLD_SIZE: 4
DEBUG:train:GLOBAL_RANK: 0
GPU available: True (cuda), used: True
TPU available: False, using: 0 TPU cores
IPU available: False, using: 0 IPUs
HPU available: False, using: 0 HPUs
2024-08-21 13:48:45 training on gpu with 4 gpu
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:NODE_RANK: 0
DEBUG:train:NODE_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:Set GLOBAL_RANK: 0
DEBUG:train:Set WORLD_SIZE: 4
DEBUG:train:GLOBAL_RANK: 0
DEBUG:train:GLOBAL_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:GLOBAL_RANK: 0
DEBUG:train:MASTER_ADDR: 192.168.1.1
DEBUG:train:MASTER_PORT: 9001

and this is on the CLIENT

DEBUG:train:NODE_RANK: 1
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:Set GLOBAL_RANK: 4
DEBUG:train:Set WORLD_SIZE: 4
DEBUG:train:GLOBAL_RANK: 4
2024-08-21 13:48:45 training on gpu with 4 gpu
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:NODE_RANK: 1
DEBUG:train:NODE_RANK: 1
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:Set GLOBAL_RANK: 4
DEBUG:train:Set WORLD_SIZE: 4
DEBUG:train:GLOBAL_RANK: 4
DEBUG:train:GLOBAL_RANK: 4
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:GLOBAL_RANK: 4
DEBUG:train:MASTER_ADDR: 10.0.147.2
DEBUG:train:MASTER_PORT: 9001
Initializing distributed: GLOBAL_RANK: 4, MEMBER: 5/4
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:LOCAL_RANK: 0
DEBUG:train:GLOBAL_RANK: 4

and from torch

Initializing distributed: GLOBAL_RANK: 0, MEMBER: 1/4

Is correct that - when on a node - members is 4 assumed that we have WORLD_SIZE=8 ( 4 gpu x 2 nodes)?

2nd issue is the timeout. I'm getting a
torch.distributed.DistStoreError: Timed out after 1801 seconds waiting for clients. 2/4 clients joined.

On the MASTER the error is

  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/trainer/trainer.py", line 1048, in _run
    self.strategy.setup_environment()
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/strategies/ddp.py", line 152, in setup_environment
    self.setup_distributed()
  File "/home/coder/.local/lib/python3.8/site-packages/pytorch_lightning/strategies/ddp.py", line 203, in setup_distributed
    _init_dist_connection(self.cluster_environment, self._process_group_backend, timeout=self._timeout)
  File "/home/coder/.local/lib/python3.8/site-packages/lightning_fabric/utilities/distributed.py", line 245, in _init_dist_connection
    torch.distributed.init_process_group(torch_distributed_backend, rank=global_rank, world_size=world_size, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 79, in wrapper
    return func(*args, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 93, in wrapper
    func_return = func(*args, **kwargs)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/distributed_c10d.py", line 1361, in init_process_group
    store, rank, world_size = next(rendezvous_iterator)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/rendezvous.py", line 258, in _env_rendezvous_handler
    store = _create_c10d_store(master_addr, master_port, rank, world_size, timeout, use_libuv)
  File "/home/coder/.local/lib/python3.8/site-packages/torch/distributed/rendezvous.py", line 185, in _create_c10d_store
    return TCPStore(
torch.distributed.DistStoreError: Timed out after 1801 seconds waiting for clients. 2/4 clients joined.

while on the CLIENT I get

[rank4]:[W821 14:18:47.728693991 socket.cpp:428] [c10d] While waitForInput, poolFD failed with (errno: 0 - Success).
2024-08-21 14:18:47 [4] is setting up NCCL communicator and retrieving ncclUniqueId from [0] via c10d key-value store by key '0', but store->get('0') got error: Connection reset by peer
Exception raised from recvBytes at ../torch/csrc/distributed/c10d/Utils.hpp:669 (most recent call first):
frame #0: c10::Error::Error(c10::SourceLocation, std::string) + 0x96 (0x7f30d2f3ff86 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libc10.so)
frame #1: <unknown function> + 0x5d1008c (0x7f310dfa308c in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #2: c10d::TCPStore::doWait(c10::ArrayRef<std::string>, std::chrono::duration<long, std::ratio<1l, 1000l> >) + 0x257 (0x7f310df9d7e7 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #3: c10d::TCPStore::doGet(std::string const&) + 0x32 (0x7f310df9db82 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #4: c10d::TCPStore::get(std::string const&) + 0xa1 (0x7f310df9ed71 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #5: c10d::PrefixStore::get(std::string const&) + 0x31 (0x7f310df537c1 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #6: c10d::PrefixStore::get(std::string const&) + 0x31 (0x7f310df537c1 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #7: c10d::PrefixStore::get(std::string const&) + 0x31 (0x7f310df537c1 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #8: c10d::ProcessGroupNCCL::broadcastUniqueNCCLID(ncclUniqueId*, bool, std::string const&, int) + 0xaf (0x7f30d423b8ff in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cuda.so)
frame #9: c10d::ProcessGroupNCCL::getNCCLComm(std::string const&, c10::Device&, c10d::OpType, int, bool) + 0x114c (0x7f30d42476dc in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cuda.so)
frame #10: c10d::ProcessGroupNCCL::broadcast(std::vector<at::Tensor, std::allocator<at::Tensor> >&, c10d::BroadcastOptions const&) + 0x643 (0x7f30d42541c3 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cuda.so)
frame #11: <unknown function> + 0x5cb44e6 (0x7f310df474e6 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #12: <unknown function> + 0x5cbf796 (0x7f310df52796 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #13: <unknown function> + 0x52dfa0b (0x7f310d572a0b in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #14: <unknown function> + 0x52dd284 (0x7f310d570284 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #15: <unknown function> + 0x1adf2b8 (0x7f3109d722b8 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #16: <unknown function> + 0x5cc46aa (0x7f310df576aa in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #17: <unknown function> + 0x5cd428c (0x7f310df6728c in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_cpu.so)
frame #18: <unknown function> + 0xdb27c5 (0x7f31212b07c5 in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_python.so)
frame #19: <unknown function> + 0x4afe4f (0x7f31209ade4f in /home/coder/.local/lib/python3.8/site-packages/torch/lib/libtorch_python.so)
<omitting python frames>
frame #22: python3() [0x4e22b3]
frame #41: python3() [0x4d9cef]
frame #47: python3() [0x4e1bd0]
frame #51: python3() [0x4e2079]
. This may indicate a possible application crash on rank 0 or a network set up issue..

torch.distributed.DistBackendError: [4] is setting up NCCL communicator and retrieving ncclUniqueId from [0] via c10d key-value store by key '0', but store->get('0') got error: Connection reset by peer

How to handle this timeout error?

[loretoparisi]

This issue has been solved setting up this NCCL env

export NCCL_VERSION=2.11.4-1 \
export NCCL_SOCKET_IFNAME=eth0 \
export NCCL_DEBUG==INFO \
export NCCL_NSOCKS_PERTHREAD=4 \
export NCCL_SOCKET_NTHREADS=4 \
export NV_LIBNCCL_PACKAGE=libnccl2=2.11.4-1+cuda11.4 \
export NV_LIBNCCL_PACKAGE_NAME=libnccl2 \
export NV_LIBNCCL_PACKAGE_VERSION=2.11.4-1

and using the FSDPStrategy strategy and a auto wrap policy in this way:

from lightning.pytorch.strategies import ModelParallelStrategy,DDPStrategy,FSDPStrategy
from torch.distributed.tensor.parallel import loss_parallel
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
from torch.distributed.fsdp import MixedPrecision

# https://sebastianraschka.com/blog/2023/pytorch-memory-optimization.html
        import torch.nn as nn
        # 1. Define a set of layers that FSDP should manage
        #    Here we are choosing the large encoder and decoder layers
        encoder_decoder_policy = {nn.TransformerEncoderLayer, nn.TransformerDecoderLayer}
        # 2. Pass the policy to the FSDPStrategy object
        auto_wrap_policy = partial(
            transformer_auto_wrap_policy,
            transformer_layer_cls=encoder_decoder_policy
        )

# model and data distributed all gpu: moodel_parallel + ddp
        # https://lightning.ai/docs/pytorch/stable/api/lightning.pytorch.strategies.FSDPStrategy.html#lightning.pytorch.strategies.FSDPStrategy
        strategy = FSDPStrategy(
                timeout=CUSTOM_TIMEOUT,
                cpu_offload=True,
                activation_checkpointing_policy=encoder_decoder_policy,
                auto_wrap_policy=auto_wrap_policy,
                # for numerical stability mixed precision keeps the model weights in full float32 precision 
                # while casting only supported operations to lower bit precision.
                mixed_precision=MixedPrecision(param_dtype=torch.torch.bfloat16, cast_forward_inputs=True),
                process_group_backend="nccl",
                sharding_strategy="FULL_SHARD"
            )