Elastic Launch

Elastic launcher

This module provides similar functionality as torch.distributed.launch, with the following additional functionalities:

1. Worker failures are handled gracefully by restarting all workers.

2. Worker RANK and WORLD_SIZE are assigned automatically.

3. Number of nodes is allowed to change between min and max sizes (elasticity).

Usage:

1. Single-node multi-worker (with sidecar etcd server)

::

>>> python -m torchelastic.distributed.launch
    --standalone
    --nnodes=1
    --nproc_per_node=$NUM_TRAINERS
    YOUR_TRAINING_SCRIPT.py (--arg1 ... train script args...)

2. Fault tolerant (fixed sized number of workers, no elasticity).:

>>> python -m torchelastic.distributed.launch
    --nnodes=$NUM_NODES
    --nproc_per_node=$NUM_TRAINERS
    --rdzv_id=$JOB_ID
    --rdzv_backend=etcd
    --rdzv_endpoint=$ETCD_HOST:$ETCD_PORT
    YOUR_TRAINING_SCRIPT.py (--arg1 ... train script args...)

3. Elastic (min=1, max=4):

>>> python -m torchelastic.distributed.launch
    --nnodes=1:4
    --nproc_per_node=$NUM_TRAINERS
    --rdzv_id=$JOB_ID
    --rdzv_backend=etcd
    --rdzv_endpoint=$ETCD_HOST:$ETCD_PORT
    YOUR_TRAINING_SCRIPT.py (--arg1 ... train script args...)

Note on rendezvous backend:

For multi-node training you need to specify:

1. --rdzv_id: a unique job id (shared by all nodes participating in the job)

2. --rdzv_backend: an implementation of torchelastic.rendevous.RendezvousHandler

3. --rdzv_endpoint: host:port-style endpoint where the rdzv backend is running.

Currently only etcd rdzv backend is supported out of the box. To use etcd, setup an etcd server with the v2 api enabled (e.g. --enable-v2).

Warning

EtcdRendezvous uses etcd api v2. You MUST enable the v2 api on the etcd server. Our tests use etcd v3.4.3.

Definitions:

1. Node - Physical instance or container.

   Maps to the unit that the job manager works with.

2. Worker - A worker in the context of distributed training.

3. Worker Group - Workers that execute the same function (e.g. trainers)

4. Local Worker Group - Subset of the workers in the

   worker group running on the same Node

5. RANK - rank of the worker within a worker group.

6. WORLD_SIZE - total number of workers in a worker group.

7. LOCAL_RANK - rank of the worker within a local worker group

8. LOCAL_WORLD_SIZE - size of the local worker group

9. rdzv_id - user defined id that uniquely identifies the worker group

   for a job. This id is used by each node to join as a member of a particular worker group.

9. rdzv_backend - the backend store of rendezvous (e.g. etcd). This is

   typically a strongly consistent key-value store.

10. rdzv_endpoint - rdzv backend server endpoint in host:port format.

A Node runs LOCAL_WORLD_SIZE workers which comprise a LocalWorkerGroup. The union of all LocalWorkerGroups in the nodes in the job comprise the WorkerGroup.

Environment Variables:

The following environment variables are made available to you in your script:

1. LOCAL_RANK - local rank

2. RANK - global rank

3. GROUP_RANK - rank of the worker group. A number between 0 - max_nnodes.

   When running a single worker group per node, this is the rank of the node.

4. LOCAL_WORLD_SIZE - local world size (e.g. number of workers running locally).

   Equal to --nproc_per_node specified on torchelastic.distributed.launch.

5. WORLD_SIZE - world size (total number of workers in the job).

6. MASTER_ADDR - fqdn of the host that is running worker with rank 0. Used to initialize torch distributed backend.

7. MASTER_PORT - port on the MASTER_ADDR that can be used to host the tcp c10d store.

8. TORCHELASTIC_RESTART_COUNT - number of worker group restarts so far.

9. TORCHELASTIC_MAX_RESTARTS - configured max number of restarts.

Deployment:

1. Start the rdzv backend server and get the endpoint (to be passed as --rdzv_endpoint to the launcher script)

2. Single-node multi-worker - start the launcher on the host to start the agent process which creates and monitors a local worker group.

3. Multi-node multi-worker - Start the launcher with the same arguments on all the nodes participating in training.

When using a job/cluster manager the entry point command to the multi-node job is invoking this launcher.

Failure Modes:

1. Worker failure - For a training job with n workers, if k <= n workers fail all workers are stopped and restarted up to max_restarts.

2. Agent failure - An agent failure results in local worker group failure, it is up to the job manager to fail the entire job (gang semantics) or attempt to replace the node. Both behaviors are supported by the agent.

3. Node failure - Same as agent failure.

Membership Changes:

1. Node departure (scale-down) - agent is notified of the departure, all existing workers are stopped, a new Worker Group is formed and all workers are started with a new RANK and WORLD_SIZE.

2. Node arrival (scale-up) - the new node is admitted to the job, all existing workers are stopped, a new Worker Group is formed and all workers are started with a new RANK and WORLD_SIZE.

Important Notices:

1. All the items in the important notices section of torch.distributed.launch apply to this module as well

2. The environment variables necessary to initialize a torch process group are provided to you by this module, no need for you to pass RANK manually. To initialize a process group in your training script, simply run

>>> import torch.distributed as dist
>>> dist.init_process_group(backend="gloo|nccl")

3. On failures or membership changes ALL surviving workers are killed immediately. Make sure to checkpoint your progress. The frequency of checkpoints should depend on your job’s tolerance for lost work.

4. This module only supports homogeneous LOCAL_WORLD_SIZE. That is, it is assumed that all nodes run the same number of local workers (per role).

5. RANK is NOT stable. Between restarts, the local workers on a node can be assgined a different range of ranks than before. NEVER hard code any assumptions about the stable-ness of ranks or some correlation between RANK and LOCAL_RANK.

6. When using elasticity (min_size != max_size) DO NOT hard code assumptions about WORLD_SIZE as the world size can change as nodes are allowed to leave and join.

7. It is recommended your script have the following structure

def main():
  load_checkpoint(checkpoint_path)
  initialize()
  train()

def train():
  for batch in iter(dataset):
    train_step(batch)

    if should_checkpoint:
      save_checkpoint(checkpoint_path)