We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[tensor] fix some unittests

[tensor] fix some unittests

[tensor] fix some unittests

[BUG]: colossalai run is stuck during multi-nodes training
### 🐛 Describe the bug

When using colossalai run during multi-nodes training, it's stuck before initializing distributed process group.

This is because potentially wrong launch command.

### Environment

Python 3.8.0
torch 1.12.1+cu113
CUDA 11.4

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `colossalai/cli/launcher/run.py`
Content:
```
1 import os
2 import sys
3 from typing import List
4 
5 import click
6 import torch
7 from packaging import version
8 
9 from colossalai.context import Config
10 
11 from .hostinfo import HostInfo, HostInfoList
12 from .multinode_runner import MultiNodeRunner
13 
14 # Constants that define our syntax
15 NODE_SEP = ','
16 
17 
18 def fetch_hostfile(hostfile_path: str, ssh_port: int) -> HostInfoList:
19     """
20     Parse the hostfile to obtain a list of hosts.
21 
22     A hostfile should look like:
23     worker-0
24     worker-1
25     worker-2
26     ...
27 
28     Args:
29         hostfile_path (str): the path to the hostfile
30         ssh_port (int): the port to connect to the host
31     """
32 
33     if not os.path.isfile(hostfile_path):
34         click.echo(f"Error: Unable to find the hostfile, no such file: {hostfile_path}")
35         exit()
36 
37     with open(hostfile_path, 'r') as fd:
38         device_pool = HostInfoList()
39 
40         for line in fd.readlines():
41             line = line.strip()
42             if line == '':
43                 # skip empty lines
44                 continue
45 
46             # build the HostInfo object
47             hostname = line.strip()
48             hostinfo = HostInfo(hostname=hostname, port=ssh_port)
49 
50             if device_pool.has(hostname):
51                 click.echo(f"Error: found duplicate host {hostname} in the hostfile")
52                 exit()
53 
54             device_pool.append(hostinfo)
55     return device_pool
56 
57 
58 def parse_device_filter(device_pool: HostInfoList, include_str=None, exclude_str=None) -> HostInfoList:
59     '''Parse an inclusion or exclusion string and filter a hostfile dictionary.
60 
61     Examples:
62         include_str="worker-0,worker-1" will execute jobs only on worker-0 and worker-1.
63         exclude_str="worker-1" will use all available devices except worker-1.
64 
65     Args:
66         device_pool (HostInfoList): a list of HostInfo objects
67         include_str (str): --include option passed by user, default None
68         exclude_str (str): --exclude option passed by user, default None
69 
70     Returns:
71         filtered_hosts (HostInfoList): filtered hosts after inclusion/exclusion
72     '''
73 
74     # Ensure include/exclude are mutually exclusive
75     if include_str and exclude_str:
76         click.echo("--include and --exclude are mutually exclusive, only one can be used")
77         exit()
78 
79     # no-op
80     if include_str is None and exclude_str is None:
81         return device_pool
82 
83     # Either build from scratch or remove items
84     if include_str:
85         parse_str = include_str
86         filtered_hosts = HostInfoList()
87     elif exclude_str:
88         parse_str = exclude_str
89         filtered_hosts = device_pool
90 
91     # foreach node in the list
92     for node_config in parse_str.split(NODE_SEP):
93         hostname = node_config
94         hostinfo = device_pool.get_hostinfo(hostname)
95         # sanity check hostname
96         if not device_pool.has(hostname):
97             click.echo(f"Error: Hostname '{hostname}' not found in hostfile")
98             exit()
99 
100         if include_str:
101             filtered_hosts.append(hostinfo)
102         elif exclude_str:
103             filtered_hosts.remove(hostname)
104 
105     return filtered_hosts
106 
107 
108 def get_launch_command(
109     master_addr: str,
110     master_port: int,
111     nproc_per_node: int,
112     user_script: str,
113     user_args: List[str],
114     node_rank: int,
115     num_nodes: int,
116     extra_launch_args: str = None,
117 ) -> str:
118     """
119     Generate a command for distributed training.
120 
121     Args:
122         master_addr (str): the host of the master node
123         master_port (str): the port of the master node
124         nproc_per_node (str): the number of processes to launch on each node
125         user_script (str): the user Python file
126         user_args (str): the arguments for the user script
127         node_rank (int): the unique ID for the node
128         num_nodes (int): the number of nodes to execute jobs
129 
130     Returns:
131         cmd (str): the command the start distributed training
132     """
133 
134     def _arg_dict_to_list(arg_dict):
135         ret = []
136 
137         for k, v in arg_dict.items():
138             if v:
139                 ret.append(f'--{k}={v}')
140             else:
141                 ret.append(f'--{k}')
142         return ret
143 
144     if extra_launch_args:
145         extra_launch_args_dict = dict()
146         for arg in extra_launch_args.split(','):
147             if '=' in arg:
148                 k, v = arg.split('=')
149                 extra_launch_args_dict[k] = v
150             else:
151                 extra_launch_args_dict[arg] = None
152         extra_launch_args = extra_launch_args_dict
153     else:
154         extra_launch_args = dict()
155 
156     torch_version = version.parse(torch.__version__)
157     assert torch_version.major >= 1
158 
159     if torch_version.minor < 9:
160         cmd = [
161             sys.executable, "-m", "torch.distributed.launch", f"--nproc_per_node={nproc_per_node}",
162             f"--master_addr={master_addr}", f"--master_port={master_port}", f"--nnodes={num_nodes}",
163             f"--node_rank={node_rank}"
164         ]
165     else:
166         # extra launch args for torch distributed launcher with torch >= 1.9
167         default_torchrun_rdzv_args = dict(rdzv_backend="c10d",
168                                           rdzv_endpoint=f"{master_addr}:{master_port}",
169                                           rdzv_id="colossalai-default-job")
170 
171         # update rdzv arguments
172         for key in default_torchrun_rdzv_args.keys():
173             if key in extra_launch_args:
174                 value = extra_launch_args.pop(key)
175                 default_torchrun_rdzv_args[key] = value
176 
177         if torch_version.minor < 10:
178             cmd = [
179                 sys.executable, "-m", "torch.distributed.run", f"--nproc_per_node={nproc_per_node}",
180                 f"--nnodes={num_nodes}", f"--node_rank={node_rank}"
181             ]
182         else:
183             cmd = [
184                 "torchrun", f"--nproc_per_node={nproc_per_node}", f"--nnodes={num_nodes}", f"--node_rank={node_rank}"
185             ]
186         cmd += _arg_dict_to_list(default_torchrun_rdzv_args)
187 
188     cmd += _arg_dict_to_list(extra_launch_args) + [user_script] + user_args
189     cmd = ' '.join(cmd)
190     return cmd
191 
192 
193 def launch_multi_processes(args: Config) -> None:
194     """
195     Launch multiple processes on a single node or multiple nodes.
196 
197     The overall logic can be summarized as the pseudo code below:
198 
199         if hostfile given:
200             hostinfo = parse_hostfile(hostfile)
201             hostinfo = include_or_exclude_hosts(hostinfo)
202             launch_on_multi_nodes(hostinfo)
203         elif hosts given:
204             hostinfo = parse_hosts(hosts)
205             launch_on_multi_nodes(hostinfo)
206         else:
207             launch_on_current_node()
208 
209     Args:
210         args (Config): the arguments taken from command line
211 
212     """
213     assert isinstance(args, Config)
214 
215     if args.nproc_per_node is None:
216         click.echo("--nproc_per_node did not receive any value")
217         exit()
218 
219     # cannot accept hosts and hostfile at the same time
220     if args.host and args.hostfile:
221         click.echo("Error: hostfile and hosts are mutually exclusive, only one is required")
222 
223     # check if hostfile is given
224     if args.hostfile:
225         device_pool = fetch_hostfile(args.hostfile, ssh_port=args.ssh_port)
226         active_device_pool = parse_device_filter(device_pool, args.include, args.exclude)
227 
228         if args.num_nodes > 0:
229             # only keep the first num_nodes to execute jobs
230             updated_active_device_pool = HostInfoList()
231             for count, hostinfo in enumerate(active_device_pool):
232                 if args.num_nodes == count:
233                     break
234                 updated_active_device_pool.append(hostinfo)
235             active_device_pool = updated_active_device_pool
236     else:
237         active_device_pool = None
238 
239     env = os.environ.copy()
240 
241     # use hosts if hostfile is not given
242     if args.host and active_device_pool is None:
243         active_device_pool = HostInfoList()
244         host_list = args.host.strip().split(NODE_SEP)
245         for hostname in host_list:
246             hostinfo = HostInfo(hostname=hostname, port=args.ssh_port)
247             active_device_pool.append(hostinfo)
248 
249     if not active_device_pool:
250         # run on local node if not hosts or hostfile is given
251         # add local node to host info list
252         active_device_pool = HostInfoList()
253         localhost_info = HostInfo(hostname='127.0.0.1', port=args.ssh_port)
254         active_device_pool.append(localhost_info)
255 
256     # launch distributed processes
257     runner = MultiNodeRunner()
258     curr_path = os.path.abspath('.')
259 
260     # collect current path env
261     env = dict()
262     for k, v in os.environ.items():
263         # do not support multi-line env var
264         if v and '\n' not in v:
265             env[k] = v
266 
267     # establish remote connection
268     runner.connect(host_info_list=active_device_pool, workdir=curr_path, env=env)
269 
270     # execute distributed launching command
271     for node_id, hostinfo in enumerate(active_device_pool):
272         cmd = get_launch_command(master_addr=args.master_addr,
273                                  master_port=args.master_port,
274                                  nproc_per_node=args.nproc_per_node,
275                                  user_script=args.user_script,
276                                  user_args=args.user_args,
277                                  node_rank=node_id,
278                                  num_nodes=len(active_device_pool),
279                                  extra_launch_args=args.extra_launch_args)
280         runner.send(hostinfo=hostinfo, cmd=cmd)
281 
282     # start training
283     msg_from_node = runner.recv_from_all()
284     has_error = False
285 
286     # print node status
287     click.echo("\n====== Training on All Nodes =====")
288     for hostname, msg in msg_from_node.items():
289         click.echo(f"{hostname}: {msg}")
290 
291         # check if a process failed
292         if msg == "failure":
293             has_error = True
294 
295     # stop all nodes
296     runner.stop_all()
297 
298     # receive the stop status
299     msg_from_node = runner.recv_from_all()
300 
301     # print node status
302     click.echo("\n====== Stopping All Nodes =====")
303     for hostname, msg in msg_from_node.items():
304         click.echo(f"{hostname}: {msg}")
305 
306     # give the process an exit code
307     # so that it behaves like a normal process
308     if has_error:
309         sys.exit(1)
310     else:
311         sys.exit(0)
312
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/colossalai/cli/launcher/run.py b/colossalai/cli/launcher/run.py
--- a/colossalai/cli/launcher/run.py
+++ b/colossalai/cli/launcher/run.py
@@ -164,9 +164,7 @@
         ]
     else:
         # extra launch args for torch distributed launcher with torch >= 1.9
-        default_torchrun_rdzv_args = dict(rdzv_backend="c10d",
-                                          rdzv_endpoint=f"{master_addr}:{master_port}",
-                                          rdzv_id="colossalai-default-job")
+        default_torchrun_rdzv_args = dict(master_addr=master_addr, master_port=master_port)
 
         # update rdzv arguments
         for key in default_torchrun_rdzv_args.keys():

{"golden_diff": "diff --git a/colossalai/cli/launcher/run.py b/colossalai/cli/launcher/run.py\n--- a/colossalai/cli/launcher/run.py\n+++ b/colossalai/cli/launcher/run.py\n@@ -164,9 +164,7 @@\n         ]\n     else:\n         # extra launch args for torch distributed launcher with torch >= 1.9\n-        default_torchrun_rdzv_args = dict(rdzv_backend=\"c10d\",\n-                                          rdzv_endpoint=f\"{master_addr}:{master_port}\",\n-                                          rdzv_id=\"colossalai-default-job\")\n+        default_torchrun_rdzv_args = dict(master_addr=master_addr, master_port=master_port)\n \n         # update rdzv arguments\n         for key in default_torchrun_rdzv_args.keys():\n", "issue": "[tensor] fix some unittests\n\n[tensor] fix some unittests\n\n[tensor] fix some unittests\n\n[BUG]: colossalai run is stuck during multi-nodes training\n### \ud83d\udc1b Describe the bug\n\nWhen using colossalai run during multi-nodes training, it's stuck before initializing distributed process group.\r\n\r\nThis is because potentially wrong launch command.\n\n### Environment\n\nPython 3.8.0\r\ntorch 1.12.1+cu113\r\nCUDA 11.4\n", "before_files": [{"content": "import os\nimport sys\nfrom typing import List\n\nimport click\nimport torch\nfrom packaging import version\n\nfrom colossalai.context import Config\n\nfrom .hostinfo import HostInfo, HostInfoList\nfrom .multinode_runner import MultiNodeRunner\n\n# Constants that define our syntax\nNODE_SEP = ','\n\n\ndef fetch_hostfile(hostfile_path: str, ssh_port: int) -> HostInfoList:\n    \"\"\"\n    Parse the hostfile to obtain a list of hosts.\n\n    A hostfile should look like:\n    worker-0\n    worker-1\n    worker-2\n    ...\n\n    Args:\n        hostfile_path (str): the path to the hostfile\n        ssh_port (int): the port to connect to the host\n    \"\"\"\n\n    if not os.path.isfile(hostfile_path):\n        click.echo(f\"Error: Unable to find the hostfile, no such file: {hostfile_path}\")\n        exit()\n\n    with open(hostfile_path, 'r') as fd:\n        device_pool = HostInfoList()\n\n        for line in fd.readlines():\n            line = line.strip()\n            if line == '':\n                # skip empty lines\n                continue\n\n            # build the HostInfo object\n            hostname = line.strip()\n            hostinfo = HostInfo(hostname=hostname, port=ssh_port)\n\n            if device_pool.has(hostname):\n                click.echo(f\"Error: found duplicate host {hostname} in the hostfile\")\n                exit()\n\n            device_pool.append(hostinfo)\n    return device_pool\n\n\ndef parse_device_filter(device_pool: HostInfoList, include_str=None, exclude_str=None) -> HostInfoList:\n    '''Parse an inclusion or exclusion string and filter a hostfile dictionary.\n\n    Examples:\n        include_str=\"worker-0,worker-1\" will execute jobs only on worker-0 and worker-1.\n        exclude_str=\"worker-1\" will use all available devices except worker-1.\n\n    Args:\n        device_pool (HostInfoList): a list of HostInfo objects\n        include_str (str): --include option passed by user, default None\n        exclude_str (str): --exclude option passed by user, default None\n\n    Returns:\n        filtered_hosts (HostInfoList): filtered hosts after inclusion/exclusion\n    '''\n\n    # Ensure include/exclude are mutually exclusive\n    if include_str and exclude_str:\n        click.echo(\"--include and --exclude are mutually exclusive, only one can be used\")\n        exit()\n\n    # no-op\n    if include_str is None and exclude_str is None:\n        return device_pool\n\n    # Either build from scratch or remove items\n    if include_str:\n        parse_str = include_str\n        filtered_hosts = HostInfoList()\n    elif exclude_str:\n        parse_str = exclude_str\n        filtered_hosts = device_pool\n\n    # foreach node in the list\n    for node_config in parse_str.split(NODE_SEP):\n        hostname = node_config\n        hostinfo = device_pool.get_hostinfo(hostname)\n        # sanity check hostname\n        if not device_pool.has(hostname):\n            click.echo(f\"Error: Hostname '{hostname}' not found in hostfile\")\n            exit()\n\n        if include_str:\n            filtered_hosts.append(hostinfo)\n        elif exclude_str:\n            filtered_hosts.remove(hostname)\n\n    return filtered_hosts\n\n\ndef get_launch_command(\n    master_addr: str,\n    master_port: int,\n    nproc_per_node: int,\n    user_script: str,\n    user_args: List[str],\n    node_rank: int,\n    num_nodes: int,\n    extra_launch_args: str = None,\n) -> str:\n    \"\"\"\n    Generate a command for distributed training.\n\n    Args:\n        master_addr (str): the host of the master node\n        master_port (str): the port of the master node\n        nproc_per_node (str): the number of processes to launch on each node\n        user_script (str): the user Python file\n        user_args (str): the arguments for the user script\n        node_rank (int): the unique ID for the node\n        num_nodes (int): the number of nodes to execute jobs\n\n    Returns:\n        cmd (str): the command the start distributed training\n    \"\"\"\n\n    def _arg_dict_to_list(arg_dict):\n        ret = []\n\n        for k, v in arg_dict.items():\n            if v:\n                ret.append(f'--{k}={v}')\n            else:\n                ret.append(f'--{k}')\n        return ret\n\n    if extra_launch_args:\n        extra_launch_args_dict = dict()\n        for arg in extra_launch_args.split(','):\n            if '=' in arg:\n                k, v = arg.split('=')\n                extra_launch_args_dict[k] = v\n            else:\n                extra_launch_args_dict[arg] = None\n        extra_launch_args = extra_launch_args_dict\n    else:\n        extra_launch_args = dict()\n\n    torch_version = version.parse(torch.__version__)\n    assert torch_version.major >= 1\n\n    if torch_version.minor < 9:\n        cmd = [\n            sys.executable, \"-m\", \"torch.distributed.launch\", f\"--nproc_per_node={nproc_per_node}\",\n            f\"--master_addr={master_addr}\", f\"--master_port={master_port}\", f\"--nnodes={num_nodes}\",\n            f\"--node_rank={node_rank}\"\n        ]\n    else:\n        # extra launch args for torch distributed launcher with torch >= 1.9\n        default_torchrun_rdzv_args = dict(rdzv_backend=\"c10d\",\n                                          rdzv_endpoint=f\"{master_addr}:{master_port}\",\n                                          rdzv_id=\"colossalai-default-job\")\n\n        # update rdzv arguments\n        for key in default_torchrun_rdzv_args.keys():\n            if key in extra_launch_args:\n                value = extra_launch_args.pop(key)\n                default_torchrun_rdzv_args[key] = value\n\n        if torch_version.minor < 10:\n            cmd = [\n                sys.executable, \"-m\", \"torch.distributed.run\", f\"--nproc_per_node={nproc_per_node}\",\n                f\"--nnodes={num_nodes}\", f\"--node_rank={node_rank}\"\n            ]\n        else:\n            cmd = [\n                \"torchrun\", f\"--nproc_per_node={nproc_per_node}\", f\"--nnodes={num_nodes}\", f\"--node_rank={node_rank}\"\n            ]\n        cmd += _arg_dict_to_list(default_torchrun_rdzv_args)\n\n    cmd += _arg_dict_to_list(extra_launch_args) + [user_script] + user_args\n    cmd = ' '.join(cmd)\n    return cmd\n\n\ndef launch_multi_processes(args: Config) -> None:\n    \"\"\"\n    Launch multiple processes on a single node or multiple nodes.\n\n    The overall logic can be summarized as the pseudo code below:\n\n        if hostfile given:\n            hostinfo = parse_hostfile(hostfile)\n            hostinfo = include_or_exclude_hosts(hostinfo)\n            launch_on_multi_nodes(hostinfo)\n        elif hosts given:\n            hostinfo = parse_hosts(hosts)\n            launch_on_multi_nodes(hostinfo)\n        else:\n            launch_on_current_node()\n\n    Args:\n        args (Config): the arguments taken from command line\n\n    \"\"\"\n    assert isinstance(args, Config)\n\n    if args.nproc_per_node is None:\n        click.echo(\"--nproc_per_node did not receive any value\")\n        exit()\n\n    # cannot accept hosts and hostfile at the same time\n    if args.host and args.hostfile:\n        click.echo(\"Error: hostfile and hosts are mutually exclusive, only one is required\")\n\n    # check if hostfile is given\n    if args.hostfile:\n        device_pool = fetch_hostfile(args.hostfile, ssh_port=args.ssh_port)\n        active_device_pool = parse_device_filter(device_pool, args.include, args.exclude)\n\n        if args.num_nodes > 0:\n            # only keep the first num_nodes to execute jobs\n            updated_active_device_pool = HostInfoList()\n            for count, hostinfo in enumerate(active_device_pool):\n                if args.num_nodes == count:\n                    break\n                updated_active_device_pool.append(hostinfo)\n            active_device_pool = updated_active_device_pool\n    else:\n        active_device_pool = None\n\n    env = os.environ.copy()\n\n    # use hosts if hostfile is not given\n    if args.host and active_device_pool is None:\n        active_device_pool = HostInfoList()\n        host_list = args.host.strip().split(NODE_SEP)\n        for hostname in host_list:\n            hostinfo = HostInfo(hostname=hostname, port=args.ssh_port)\n            active_device_pool.append(hostinfo)\n\n    if not active_device_pool:\n        # run on local node if not hosts or hostfile is given\n        # add local node to host info list\n        active_device_pool = HostInfoList()\n        localhost_info = HostInfo(hostname='127.0.0.1', port=args.ssh_port)\n        active_device_pool.append(localhost_info)\n\n    # launch distributed processes\n    runner = MultiNodeRunner()\n    curr_path = os.path.abspath('.')\n\n    # collect current path env\n    env = dict()\n    for k, v in os.environ.items():\n        # do not support multi-line env var\n        if v and '\\n' not in v:\n            env[k] = v\n\n    # establish remote connection\n    runner.connect(host_info_list=active_device_pool, workdir=curr_path, env=env)\n\n    # execute distributed launching command\n    for node_id, hostinfo in enumerate(active_device_pool):\n        cmd = get_launch_command(master_addr=args.master_addr,\n                                 master_port=args.master_port,\n                                 nproc_per_node=args.nproc_per_node,\n                                 user_script=args.user_script,\n                                 user_args=args.user_args,\n                                 node_rank=node_id,\n                                 num_nodes=len(active_device_pool),\n                                 extra_launch_args=args.extra_launch_args)\n        runner.send(hostinfo=hostinfo, cmd=cmd)\n\n    # start training\n    msg_from_node = runner.recv_from_all()\n    has_error = False\n\n    # print node status\n    click.echo(\"\\n====== Training on All Nodes =====\")\n    for hostname, msg in msg_from_node.items():\n        click.echo(f\"{hostname}: {msg}\")\n\n        # check if a process failed\n        if msg == \"failure\":\n            has_error = True\n\n    # stop all nodes\n    runner.stop_all()\n\n    # receive the stop status\n    msg_from_node = runner.recv_from_all()\n\n    # print node status\n    click.echo(\"\\n====== Stopping All Nodes =====\")\n    for hostname, msg in msg_from_node.items():\n        click.echo(f\"{hostname}: {msg}\")\n\n    # give the process an exit code\n    # so that it behaves like a normal process\n    if has_error:\n        sys.exit(1)\n    else:\n        sys.exit(0)\n", "path": "colossalai/cli/launcher/run.py"}], "after_files": [{"content": "import os\nimport sys\nfrom typing import List\n\nimport click\nimport torch\nfrom packaging import version\n\nfrom colossalai.context import Config\n\nfrom .hostinfo import HostInfo, HostInfoList\nfrom .multinode_runner import MultiNodeRunner\n\n# Constants that define our syntax\nNODE_SEP = ','\n\n\ndef fetch_hostfile(hostfile_path: str, ssh_port: int) -> HostInfoList:\n    \"\"\"\n    Parse the hostfile to obtain a list of hosts.\n\n    A hostfile should look like:\n    worker-0\n    worker-1\n    worker-2\n    ...\n\n    Args:\n        hostfile_path (str): the path to the hostfile\n        ssh_port (int): the port to connect to the host\n    \"\"\"\n\n    if not os.path.isfile(hostfile_path):\n        click.echo(f\"Error: Unable to find the hostfile, no such file: {hostfile_path}\")\n        exit()\n\n    with open(hostfile_path, 'r') as fd:\n        device_pool = HostInfoList()\n\n        for line in fd.readlines():\n            line = line.strip()\n            if line == '':\n                # skip empty lines\n                continue\n\n            # build the HostInfo object\n            hostname = line.strip()\n            hostinfo = HostInfo(hostname=hostname, port=ssh_port)\n\n            if device_pool.has(hostname):\n                click.echo(f\"Error: found duplicate host {hostname} in the hostfile\")\n                exit()\n\n            device_pool.append(hostinfo)\n    return device_pool\n\n\ndef parse_device_filter(device_pool: HostInfoList, include_str=None, exclude_str=None) -> HostInfoList:\n    '''Parse an inclusion or exclusion string and filter a hostfile dictionary.\n\n    Examples:\n        include_str=\"worker-0,worker-1\" will execute jobs only on worker-0 and worker-1.\n        exclude_str=\"worker-1\" will use all available devices except worker-1.\n\n    Args:\n        device_pool (HostInfoList): a list of HostInfo objects\n        include_str (str): --include option passed by user, default None\n        exclude_str (str): --exclude option passed by user, default None\n\n    Returns:\n        filtered_hosts (HostInfoList): filtered hosts after inclusion/exclusion\n    '''\n\n    # Ensure include/exclude are mutually exclusive\n    if include_str and exclude_str:\n        click.echo(\"--include and --exclude are mutually exclusive, only one can be used\")\n        exit()\n\n    # no-op\n    if include_str is None and exclude_str is None:\n        return device_pool\n\n    # Either build from scratch or remove items\n    if include_str:\n        parse_str = include_str\n        filtered_hosts = HostInfoList()\n    elif exclude_str:\n        parse_str = exclude_str\n        filtered_hosts = device_pool\n\n    # foreach node in the list\n    for node_config in parse_str.split(NODE_SEP):\n        hostname = node_config\n        hostinfo = device_pool.get_hostinfo(hostname)\n        # sanity check hostname\n        if not device_pool.has(hostname):\n            click.echo(f\"Error: Hostname '{hostname}' not found in hostfile\")\n            exit()\n\n        if include_str:\n            filtered_hosts.append(hostinfo)\n        elif exclude_str:\n            filtered_hosts.remove(hostname)\n\n    return filtered_hosts\n\n\ndef get_launch_command(\n    master_addr: str,\n    master_port: int,\n    nproc_per_node: int,\n    user_script: str,\n    user_args: List[str],\n    node_rank: int,\n    num_nodes: int,\n    extra_launch_args: str = None,\n) -> str:\n    \"\"\"\n    Generate a command for distributed training.\n\n    Args:\n        master_addr (str): the host of the master node\n        master_port (str): the port of the master node\n        nproc_per_node (str): the number of processes to launch on each node\n        user_script (str): the user Python file\n        user_args (str): the arguments for the user script\n        node_rank (int): the unique ID for the node\n        num_nodes (int): the number of nodes to execute jobs\n\n    Returns:\n        cmd (str): the command the start distributed training\n    \"\"\"\n\n    def _arg_dict_to_list(arg_dict):\n        ret = []\n\n        for k, v in arg_dict.items():\n            if v:\n                ret.append(f'--{k}={v}')\n            else:\n                ret.append(f'--{k}')\n        return ret\n\n    if extra_launch_args:\n        extra_launch_args_dict = dict()\n        for arg in extra_launch_args.split(','):\n            if '=' in arg:\n                k, v = arg.split('=')\n                extra_launch_args_dict[k] = v\n            else:\n                extra_launch_args_dict[arg] = None\n        extra_launch_args = extra_launch_args_dict\n    else:\n        extra_launch_args = dict()\n\n    torch_version = version.parse(torch.__version__)\n    assert torch_version.major >= 1\n\n    if torch_version.minor < 9:\n        cmd = [\n            sys.executable, \"-m\", \"torch.distributed.launch\", f\"--nproc_per_node={nproc_per_node}\",\n            f\"--master_addr={master_addr}\", f\"--master_port={master_port}\", f\"--nnodes={num_nodes}\",\n            f\"--node_rank={node_rank}\"\n        ]\n    else:\n        # extra launch args for torch distributed launcher with torch >= 1.9\n        default_torchrun_rdzv_args = dict(master_addr=master_addr, master_port=master_port)\n\n        # update rdzv arguments\n        for key in default_torchrun_rdzv_args.keys():\n            if key in extra_launch_args:\n                value = extra_launch_args.pop(key)\n                default_torchrun_rdzv_args[key] = value\n\n        if torch_version.minor < 10:\n            cmd = [\n                sys.executable, \"-m\", \"torch.distributed.run\", f\"--nproc_per_node={nproc_per_node}\",\n                f\"--nnodes={num_nodes}\", f\"--node_rank={node_rank}\"\n            ]\n        else:\n            cmd = [\n                \"torchrun\", f\"--nproc_per_node={nproc_per_node}\", f\"--nnodes={num_nodes}\", f\"--node_rank={node_rank}\"\n            ]\n        cmd += _arg_dict_to_list(default_torchrun_rdzv_args)\n\n    cmd += _arg_dict_to_list(extra_launch_args) + [user_script] + user_args\n    cmd = ' '.join(cmd)\n    return cmd\n\n\ndef launch_multi_processes(args: Config) -> None:\n    \"\"\"\n    Launch multiple processes on a single node or multiple nodes.\n\n    The overall logic can be summarized as the pseudo code below:\n\n        if hostfile given:\n            hostinfo = parse_hostfile(hostfile)\n            hostinfo = include_or_exclude_hosts(hostinfo)\n            launch_on_multi_nodes(hostinfo)\n        elif hosts given:\n            hostinfo = parse_hosts(hosts)\n            launch_on_multi_nodes(hostinfo)\n        else:\n            launch_on_current_node()\n\n    Args:\n        args (Config): the arguments taken from command line\n\n    \"\"\"\n    assert isinstance(args, Config)\n\n    if args.nproc_per_node is None:\n        click.echo(\"--nproc_per_node did not receive any value\")\n        exit()\n\n    # cannot accept hosts and hostfile at the same time\n    if args.host and args.hostfile:\n        click.echo(\"Error: hostfile and hosts are mutually exclusive, only one is required\")\n\n    # check if hostfile is given\n    if args.hostfile:\n        device_pool = fetch_hostfile(args.hostfile, ssh_port=args.ssh_port)\n        active_device_pool = parse_device_filter(device_pool, args.include, args.exclude)\n\n        if args.num_nodes > 0:\n            # only keep the first num_nodes to execute jobs\n            updated_active_device_pool = HostInfoList()\n            for count, hostinfo in enumerate(active_device_pool):\n                if args.num_nodes == count:\n                    break\n                updated_active_device_pool.append(hostinfo)\n            active_device_pool = updated_active_device_pool\n    else:\n        active_device_pool = None\n\n    env = os.environ.copy()\n\n    # use hosts if hostfile is not given\n    if args.host and active_device_pool is None:\n        active_device_pool = HostInfoList()\n        host_list = args.host.strip().split(NODE_SEP)\n        for hostname in host_list:\n            hostinfo = HostInfo(hostname=hostname, port=args.ssh_port)\n            active_device_pool.append(hostinfo)\n\n    if not active_device_pool:\n        # run on local node if not hosts or hostfile is given\n        # add local node to host info list\n        active_device_pool = HostInfoList()\n        localhost_info = HostInfo(hostname='127.0.0.1', port=args.ssh_port)\n        active_device_pool.append(localhost_info)\n\n    # launch distributed processes\n    runner = MultiNodeRunner()\n    curr_path = os.path.abspath('.')\n\n    # collect current path env\n    env = dict()\n    for k, v in os.environ.items():\n        # do not support multi-line env var\n        if v and '\\n' not in v:\n            env[k] = v\n\n    # establish remote connection\n    runner.connect(host_info_list=active_device_pool, workdir=curr_path, env=env)\n\n    # execute distributed launching command\n    for node_id, hostinfo in enumerate(active_device_pool):\n        cmd = get_launch_command(master_addr=args.master_addr,\n                                 master_port=args.master_port,\n                                 nproc_per_node=args.nproc_per_node,\n                                 user_script=args.user_script,\n                                 user_args=args.user_args,\n                                 node_rank=node_id,\n                                 num_nodes=len(active_device_pool),\n                                 extra_launch_args=args.extra_launch_args)\n        runner.send(hostinfo=hostinfo, cmd=cmd)\n\n    # start training\n    msg_from_node = runner.recv_from_all()\n    has_error = False\n\n    # print node status\n    click.echo(\"\\n====== Training on All Nodes =====\")\n    for hostname, msg in msg_from_node.items():\n        click.echo(f\"{hostname}: {msg}\")\n\n        # check if a process failed\n        if msg == \"failure\":\n            has_error = True\n\n    # stop all nodes\n    runner.stop_all()\n\n    # receive the stop status\n    msg_from_node = runner.recv_from_all()\n\n    # print node status\n    click.echo(\"\\n====== Stopping All Nodes =====\")\n    for hostname, msg in msg_from_node.items():\n        click.echo(f\"{hostname}: {msg}\")\n\n    # give the process an exit code\n    # so that it behaves like a normal process\n    if has_error:\n        sys.exit(1)\n    else:\n        sys.exit(0)\n", "path": "colossalai/cli/launcher/run.py"}]}