launch.py 12.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13
# Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
14
r"""
15
fleetrun is a module that spawns multiple distributed
16 17
process on each training node for gpu training and cpu training.
Usage:
18
    In both of single node training or multiple node training, this module
19 20 21 22 23 24 25 26
launch a process on each of the given gpu card or cpu machine.
    GPU training:
    1. for single node training with all visible gpu cards:
       fleetrun your_training_py (arg1 arg2 and all others)
    2. for single node training with [0,4) cards
       fleetrun --gpus="0,1,2,3" your_training_py (arg1 arg2 and all others)
    3. for multiple node training such as two node:192.168.0.16, 192.168.0.17
        on 192.168.0.16:
27
            fleetrun --ips="192.168.0.16,192.168.0.17" \
28 29 30 31 32 33
                your_training_py (arg1 arg2 and all others)
        on 192.168.0.17:
            fleetrun --ips="192.168.0.16,192.168.0.17" \
                your_training_py (arg1 arg2 and all others)
    CPU training:
    1. for single node training with multi servers and workers:
34
        fleetrun --server_num=2 --worker_num=2 your_training_py (arg1 arg2 and all others)
35
    2. for multiple node training such as two node:192.168.0.16, 192.168.0.17 \
36
        with 2 servers and 4 workers.
37
        on 192.168.0.16:
38 39
            fleetrun --servers="192.168.0.16:6170,192.168.0.17:6170" \
                --workers="192.168.0.16,192.168.0.17,192.168.0.16,192.168.0.17" \
40 41 42
                your_training_py (arg1 arg2 and all others)
        on 192.168.0.17:
            fleetrun --servers="192.168.0.16:6170,192.168.0.17:6171" \
43 44 45 46 47 48 49 50 51 52 53
                --workers="192.168.0.16,192.168.0.17,192.168.0.16,192.168.0.17" \
                your_training_py (arg1 arg2 and all others)
    3. use gloo backend for multiple node training such as two node:192.168.0.16, 192.168.0.17 \
        with 2 servers and 4 workers. (workers should set port)
        on 192.168.0.16:
            fleetrun --servers="192.168.0.16:6170,192.168.0.17:6170" \
                --workers="192.168.0.16:6171,192.168.0.17:6171,192.168.0.16:6172,192.168.0.17:6172" \
                your_training_py (arg1 arg2 and all others)
        on 192.168.0.17:
            fleetrun --servers="192.168.0.16:6170,192.168.0.17:6170" \
                --workers="192.168.0.16:6171,192.168.0.17:6171,192.168.0.16:6172,192.168.0.17:6172" \
54 55 56 57
                your_training_py (arg1 arg2 and all others)
"""

from __future__ import print_function
58 59

import shutil
60
import sys
61
import tempfile
62 63 64 65 66 67 68 69 70
from sys import version
import subprocess
import os
import time
import six
import copy
from argparse import ArgumentParser, REMAINDER
import paddle
import paddle.fluid as fluid
71
from paddle.distributed.fleet import launch_utils
72

73
# TODO(danleifeng): Don't import * from a module
74 75
from paddle.distributed.fleet.launch_utils import *
import paddle.distributed.fleet.cloud_utils as cloud_utils
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93


def _print_arguments(args):
    print("-----------  Configuration Arguments -----------")
    for arg, value in sorted(six.iteritems(vars(args))):
        print("%s: %s" % (arg, value))
    print("------------------------------------------------")


def _parse_args():
    """
    Helper function parsing the command line options
    @retval ArgumentParser
    """
    parser = ArgumentParser(
        description='''start paddle training using multi-process mode.
see: http://www.paddlepaddle.org/documentation/docs/zh/1.6/user_guides/howto/training/cluster_howto.html#permalink-8--nccl2-
''')
94
    base_group = parser.add_argument_group("Base Parameters")
95

96 97
    base_group.add_argument(
        "--log_dir",
98
        type=str,
99 100 101 102
        default="log",
        help="The path for each process's log.If it's not set, the log will printed to default pipe."
    )

103 104 105 106 107 108 109 110
    base_group.add_argument(
        "--nproc_per_node",
        type=int,
        default=None,
        help="The number of processes to launch on a node."
        "In gpu training, it should be less or equal to the gpus number of you system(or you set by --gpus). And so each process can"
        " bound to one or average number of gpus.")

111
    base_group.add_argument(
112 113 114
        "--gpus",
        type=str,
        default=None,
115 116 117
        help="It's for gpu training."
        "For example:"
        "--gpus=\"0,1,2,3\" will launch four training processes each bound to one gpu."
118 119
    )

120
    base_group.add_argument(
121 122 123 124 125 126 127
        "training_script",
        type=str,
        help="The full path to the single GPU training "
        "program/script to be launched in parallel, "
        "followed by all the arguments for the "
        "training script")

128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
    base_group.add_argument('training_script_args', nargs=REMAINDER)

    # Optional arguments for the launch helper
    # for collective
    collective_group = parser.add_argument_group("Collective Parameters")
    collective_group.add_argument(
        "--ips",
        type=str,
        default="127.0.0.1",
        help="Paddle cluster nodes ips, such as 192.168.0.16,192.168.0.17..")

    ps_group = parser.add_argument_group("Parameter-Server Parameters")
    # for parameter server
    ps_group.add_argument(
        "--servers", type=str, default="", help="User defined servers ip:port")
    ps_group.add_argument(
        "--workers", type=str, default="", help="User defined workers ip:port")
    ps_group.add_argument(
        "--heter_workers",
        type=str,
        default="",
        help="User defined heter workers ip:port")

    ps_group.add_argument("--worker_num", type=int, help="number of workers")
    ps_group.add_argument("--server_num", type=int, help="number of servers")
    ps_group.add_argument(
        "--heter_worker_num", type=int, help="number of heter_workers")
155
    ps_group.add_argument("--http_port", type=int, help="Gloo http Port")
156

157 158 159
    return parser.parse_args()


160
def get_cluster_from_args(args, device_mode, devices_per_proc):
161 162 163 164 165 166
    node_ips = [x.strip() for x in args.ips.split(',')]
    if len(node_ips) == 1:
        node_ip = node_ips[0]
    else:
        _, node_ip = get_host_name_ip()

167
    assert node_ip in node_ips, "Can't find your local ip {%s} in node_ips: {%s}" \
168
        % (node_ip, node_ips)
169 170
    node_rank = node_ips.index(node_ip)

171
    logger.debug("parsed from args: node_ips:{} node_ip:{} node_rank:{}".format(
172 173 174 175 176
        node_ips, node_ip, node_rank))

    free_ports = None
    if not cloud_utils.use_paddlecloud() and len(
            node_ips) <= 1 and os.environ.get('FLAGS_START_PORT') is None:
177
        free_ports = find_free_ports(len(devices_per_proc))
178 179 180 181 182
        if free_ports is not None:
            free_ports = list(free_ports)
    else:
        start_port = 6070
        if os.environ.get('FLAGS_START_PORT') is not None:
183
            start_port = int(os.environ.get('FLAGS_START_PORT'))
184

185 186 187
        free_ports = [
            x for x in range(start_port, start_port + len(devices_per_proc))
        ]
188

189 190 191
    trainer_endpoints = []
    for ip in node_ips:
        trainer_endpoints.append(["%s:%d" % (ip, port) for port in free_ports])
192 193
    return get_cluster(node_ips, node_ip, trainer_endpoints, device_mode,
                       devices_per_proc)
194 195 196 197


def launch_collective(args):
    # parse arguments, used for cloud-single-machine and local
198
    (device_mode, devices_per_proc) = launch_utils.get_device_proc_info(args)
199
    trainers_num = cloud_utils.get_trainers_num()
200 201
    logger.debug("parsed from args trainerss_num:{} mode:{} devices:{}".format(
        trainers_num, device_mode, devices_per_proc))
202 203 204 205

    cluster = None
    pod = None

206 207 208
    start_port = 6170
    if os.environ.get('FLAGS_START_PORT') is not None:
        start_port = os.environ.get('FLAGS_START_PORT')
209
    if cloud_utils.use_paddlecloud() and trainers_num != 1:
210 211
        cluster, pod = cloud_utils.get_cloud_cluster(
            args.ips, device_mode, devices_per_proc, start_port)
212
        logger.debug("get cluster from cloud:{}".format(cluster))
213 214
    else:
        # trainers_num = 1 or not use paddlecloud ips="a,b"
215 216
        cluster, pod = get_cluster_from_args(args, device_mode,
                                             devices_per_proc)
217
        logger.debug("get cluster from args:{}".format(cluster))
218

219 220 221
    global_envs = copy.copy(os.environ.copy())
    gloo_rendezvous_dir = tempfile.mkdtemp()
    # add gloo env
222
    global_envs["PADDLE_WITH_GLOO"] = str(os.getenv("PADDLE_WITH_GLOO", "1"))
223
    global_envs["PADDLE_GLOO_RENDEZVOUS"] = "3"
224 225
    global_envs["PADDLE_GLOO_FS_PATH"] = gloo_rendezvous_dir

226 227 228 229 230
    procs = start_local_trainers(
        cluster,
        pod,
        training_script=args.training_script,
        training_script_args=args.training_script_args,
231 232
        log_dir=args.log_dir,
        envs=global_envs)
233 234 235 236 237

    while True:
        alive = watch_local_trainers(procs, cluster.trainers_nranks())

        if not alive:
238 239
            logger.info("Local processes completed.")
            logger.debug("POD info:{}".format(pod))
240 241 242 243
            break

        time.sleep(3)

244 245 246
    if os.path.exists(gloo_rendezvous_dir):
        shutil.rmtree(gloo_rendezvous_dir)

247

248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267
def launch_ps(args, distribute_mode):
    cloud_flag = cloud_utils.use_paddlecloud()

    # for ps-cpu on paddlecloud
    if cloud_flag and distribute_mode == DistributeMode.PS:
        direct_start(args)
        return
    elif cloud_flag and distribute_mode == DistributeMode.PS_HETER:
        cloud_ps_heter_env_set(args)
        args.workers = os.getenv("PADDLE_TRAINER_ENDPOINTS")
        args.servers = os.getenv("PADDLE_PSERVERS_IP_PORT_LIST")
        args.heter_workers = os.getenv("PADDLE_HETER_TRAINER_IP_PORT_LIST")

    ps_launcher = ParameterServerLauncher(args, distribute_mode)
    ps_launcher.start_ps()
    return


def which_distributed_mode(args):
    ps_args = [
268 269
        '--worker_num', '--server_num', '--heter_worker_num', '--servers',
        '--workers', '--heter_workers', '--http_port'
270
    ]
271
    collective_args = ['--ips']
272

273
    ps_heter_args = ["--heter_worker_num", "--heter_workers"]
274 275 276 277 278 279 280 281

    has_ps_args = [
        ps_arg for ps_arg in ps_args if ps_arg in " ".join(sys.argv[1:-1])
    ]
    has_collective_args = [
        co_arg for co_arg in collective_args
        if co_arg in " ".join(sys.argv[1:-1])
    ]
282 283 284 285 286 287

    if len(has_ps_args) > 1 and len(has_collective_args) > 1:
        raise ValueError(
            "Only one mode(Collective or Parameter-Server) can be selected at the same time, but more than one configuration was received."
        )

288 289 290 291 292
    if fluid.core.is_compiled_with_cuda():
        cuda_device_num = fluid.core.get_cuda_device_count()
    else:
        cuda_device_num = 0

293 294 295 296 297 298 299 300 301
    if len(has_ps_args) > 0:
        logger.info(
            "Run parameter-sever mode. pserver arguments:{}, cuda count:{}".
            format(has_ps_args, cuda_device_num))
        has_ps_heter_args = list(set(has_ps_args) & set(ps_heter_args))
        if len(has_ps_heter_args) > 0:
            return DistributeMode.PS_HETER
        else:
            return DistributeMode.PS
302
    elif len(has_collective_args) > 0:
303
        logger.info("Run collective gpu mode. gpu arguments:{}, cuda count:{}".
304
                    format(has_collective_args, cuda_device_num))
305
        return DistributeMode.COLLECTIVE
306
    else:
307 308 309 310 311 312 313 314 315 316
        if not fluid.core.is_compiled_with_cuda():
            logger.warning(
                "Not found distinct arguments and not compiled with cuda. Default use ps mode"
            )
            return DistributeMode.PS
        else:
            logger.warning(
                "Not found distinct arguments and compiled with cuda. Default use collective mode"
            )
            return DistributeMode.COLLECTIVE
317 318 319 320 321 322 323 324 325


def launch():
    args = _parse_args()
    logger = get_logger()
    _print_arguments(args)

    distribute_mode = which_distributed_mode(args)
    if distribute_mode == DistributeMode.COLLECTIVE:
326
        launch_collective(args)
327 328
    else:
        launch_ps(args, distribute_mode)
329 330 331 332


if __name__ == "__main__":
    launch()