wrong global rank when trying multi-nodes

[jungwhank]

Hello,
I'm trying to train my model with multi-nodes (2 nodes, 8 gpus per each, using ddp accelator & trying without using slurm)
But I got problem with GLOBAL_RANK

in node 1,

initializing ddp: GLOBAL_RANK: 1, MEMBER: 2/16
...
initializing ddp: GLOBAL_RANK: 7, MEMBER: 8/16

same as in node 2,

initializing ddp: GLOBAL_RANK: 0, MEMBER: 1/16
...
initializing ddp: GLOBAL_RANK: 7, MEMBER: 8/16

And got stuck with repeated message like below

Waiting in store based barrier to initialize process group for rank: 3, key: store_based_barrier_key:1 (world_size=16, worker_count=13, timeout=0:30:00)
Waiting in store based barrier to initialize process group for rank: 1, key: store_based_barrier_key:1 (world_size=16, worker_count=13, timeout=0:30:00)

I'm trying to setup like this document but also got problem, like below

 os.environ["MASTER_ADDR"] = master_addr
 os.environ["MASTER_PORT"] = master_port
 os.environ["WORLD_SIZE"] = "16"
 os.environ["NODE_RANK"] = rank

  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/trainer/trainer.py", line 474, in fit
    self.accelerator.setup(self, model)
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/accelerators/gpu.py", line 19, in setup
    return super().setup(trainer, model)
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/accelerators/accelerator.py", line 69, in setup
    self.connect_training_type_plugin(self.training_type_plugin, model)
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/accelerators/accelerator.py", line 328, in connect_training_type_plugin
    plugin.connect(model)
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/plugins/training_type/parallel.py", line 68, in connect
    self.setup(model)
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/plugins/training_type/ddp.py", line 95, in setup
    self.task_idx = self.cluster_environment.local_rank()
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/site-packages/pytorch_lightning/plugins/environments/torchelastic_environment.py", line 48, in local_rank
    return int(os.environ['LOCAL_RANK'])
  File "/home/user/.pyenv/versions/3.7.9/lib/python3.7/os.py", line 681, in __getitem__
    raise KeyError(key) from None

I'd appreciate any help. thanks in advance

[carmocca]

I believe the docs might be outdated, have you tried changing NODE_RANK to LOCAL_RANK?

cc: @awaelchli. We should provide better error messages for this

[jungwhank]

Hi, @carmocca
Thanks for answering.

I changed and try it but do not work.
Is there any document or code with lightning multi-nodes without using slurm or shell script?
or is there proper way to setting manually like using pytorch distributed (dist.init_process_group )?

[awaelchli]

Here is a link to our docs that explains how to set it up without slurm:
https://pytorch-lightning.readthedocs.io/en/latest/advanced/cluster.html#computing-cluster
and also see @SeanNaren 's post below for an explicit example :)

[SeanNaren]

What have you set for MASTER_ADDR and MASTER_PORT? These have to reference to one of the two machines you are using. For example if I have two nodes like this:

IP: 512.124.134.4
IP: 512.124.136.8

And I want 512.124.134.4 to be my master node.

For both my machines I'd need to run something like MASTER_ADDR=512.124.134.4 MASTER_PORT=4500 python train.py.

Let me know if this helps! On top of this, we should update the doc if this does work :)

[jungwhank]

Hi, @SeanNaren. Thanks for reply.
I tried as you said (I think it is same as I did before, I used os.environ settings),
but it still don't get right GLOBAL_RANK and get stucked like below.

These are my commands for both machines (2gpus 2nodes)

MASTER_ADDR=NODE1_IP MASTER_PORT=4500 python run_pretrain.py
MASTER_ADDR=NODE1_IP MASTER_PORT=4500 python run_pretrain.py

MASTER_ADDR=NODE1_IP MASTER_PORT=4500 WORLD_SIZE=4 NODE_RANK=0 python run_pretrain.py 
MASTER_ADDR=NODE1_IP MASTER_PORT=4500 WORLD_SIZE=4 NODE_RANK=1 python run_pretrain.py 

and I tried like this document also

MASTER_ADDR=NODE1_IP MASTER_PORT=4500 WORLD_SIZE=4 NODE_RANK=0 LOCAL_RANK=0 python run_pretrain.py
MASTER_ADDR=NODE1_IP MASTER_PORT=4500 WORLD_SIZE=4 NODE_RANK=1 LOCAL_RANK=0 python run_pretrain.py

And it get stuck like below (both node1 & node2)

...
initializing ddp: GLOBAL_RANK: 0, MEMBER: 1/4
[2021-03-15 09:45:37,382][lightning][INFO] - initializing ddp: GLOBAL_RANK: 0, MEMBER: 1/4
LOCAL_RANK: 1 - CUDA_VISIBLE_DEVICES: [0,1,2,3,4,5,6,7]
[2021-03-15 09:45:38,247][lightning][INFO] - LOCAL_RANK: 1 - CUDA_VISIBLE_DEVICES: [0,1,2,3,4,5,6,7]
initializing ddp: GLOBAL_RANK: 1, MEMBER: 2/4
[2021-03-15 09:45:38,293][lightning][INFO] - initializing ddp: GLOBAL_RANK: 1, MEMBER: 2/4

Also I tried

MASTER_ADDR=NODE1_IP MASTER_PORT=4500 WORLD_SIZE=4 python run_pretrain.py

But it get raise KeyError(key) from None KeyError: 'LOCAL_RANK'

Is there any other way for multi-nodes training?

[jungwhank]

@SeanNaren @awaelchli
Hi, I'm trying to use custom ClusterEnvironment setup with ddp accelerators,
but I think docs1 and docs2 is outdated.

from pytorch_lightning.plugins.environments.cluster_environment import ClusterEnvironment
class MyCluster(ClusterEnvironment):
    def master_address(self):
        return MY_ADDR
    def master_port(self):
        return MY_PORT
    def world_size(self):
        return MY_WORLD_SIZE

in main,

...
# case1
cluster = [MyCluster()]
trainer = Trainer(args,  cluster_environment ='ddp')

# case2
cluster = [MyCluster()]
trainer = Trainer(args,  accelators='ddp', plugins=[cluster])

but got error like this

# case1
No cluster_environment in trainer init

# case2
Found invalid type for plugin [<ppl.nsml_cluster.MyCluster object at 0x14e0fd8ed650>]. Expected a precision or training type plugin.

How can I setup my own cluster and using with ddp? Thanks in advance

[jungwhank]

It seems like I have succeeded in multi-nodes training in my cluster environment,
but I'm not sure it is right way to make custom plugins.
Is it right way to implement my custom cluster with ddp?

from pytorch_lightning.plugins.environments.cluster_environment import ClusterEnvironment
from pytorch_lightning.plugins.training_type import DDPPlugin

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

    def master_address(self):
        if self._num_nodes > 1:
            MASTER_ADDR = MY_ADDR
        else:
            MASTER_ADDR = os.environ.get("MASTER_ADDR", "127.0.0.1")
        log.debug(f"MASTER_ADDR: {MASTER_ADDR}")
        return MASTER_ADDR

    def master_port(self):
        if self._master_port is None:
            if self._num_nodes > 1:
                self._master_port = MY_PORT
            else:
                self._master_port = os.environ.get("MASTER_PORT", find_free_network_port())
        log.debug(f"MASTER_ADDR: {self._master_port}")
        return int(self._master_port)

    def world_size(self):
        return None

    def node_rank(self):
        log.debug(f"NODE_RANK: {MY_RANK}")
        return int(MY_RANK)

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


class CustomPlugin(DDPPlugin):
    def __init__(self, **kwargs: Union[Any, Dict[str, Any]]) -> None:
        super().__init__(
            num_nodes=2,
            cluster_environment=CustomEnvironment(),
            **kwargs,
        )


# In main
...
trainer = Trainer(args.trainer, accelerator='ddp', plugins=CustomPlugin(), callbacks=callbacks)

# Test - passed
assert isinstance(trainer.accelerator, GPUAccelerator)
assert isinstance(trainer.accelerator.training_type_plugin, CustomPlugin)
assert isinstance(trainer.accelerator.training_type_plugin.cluster_environment, CustomEnvironment)

[awaelchli]

Looks ok, yes. Have you been successful with this?

[jungwhank]

Thanks! It works well with custom cluster environment

[loretoparisi]

Thanks @jungwhank but
When using CustomPlugin as defined above in the trainer:

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

I'm getting an init error

    cluster_environment=CustomEnvironment(),
TypeError: Can't instantiate abstract class CustomEnvironment with abstract methods creates_processes_externally, detect, global_rank, main_address, main_port, set_global_rank, set_world_size

completed on node 

[loretoparisi]

I've tried to implement missing abstract method as specified above.
More details in #20219

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