launch_utils.py 43.8 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
# 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.

import functools
import logging
import socket
import time
import os
import signal
import copy
import sys
import subprocess
24 25
import tempfile
import shutil
26 27
from contextlib import closing
import socket
28
import warnings
29
import six
30

31 32
import paddle
import paddle.fluid as fluid
33 34 35 36
logger = logging.getLogger("root")
logger.propagate = False


G
gongweibao 已提交
37
class DistributeMode():
38 39 40 41 42 43 44 45
    """
    There are various mode for fleetrun, each of them is designed for different model.
    """
    COLLECTIVE = 0
    PS = 1
    PS_HETER = 2


G
gongweibao 已提交
46
class DeviceMode():
47 48 49
    """
    Training devices type
    """
50
    UNKNOWN = -1
51 52 53
    CPU = 0
    GPU = 1
    KUNLUN = 2
54
    XPU = 2
55 56
    ASCEND_NPU = 3
    UNKNOWN = 3
57 58


59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102
class Cluster(object):
    def __init__(self, hdfs):
        self.job_server = None
        self.pods = []
        self.hdfs = None
        self.job_stage_flag = None

    def __str__(self):
        return "job_server:{} pods:{} job_stage_flag:{} hdfs:{}".format(
            self.job_server, [str(pod) for pod in self.pods],
            self.job_stage_flag, self.hdfs)

    def __eq__(self, cluster):
        if len(self.pods) != len(cluster.pods):
            return False

        for a, b in zip(self.pods, cluster.pods):
            if a != b:
                return False

        if self.job_stage_flag != cluster.job_stage_flag:
            return False

        return True

    def __ne__(self, cluster):
        return not self.__eq__(cluster)

    def update_pods(cluster):
        self.pods = copy.copy(cluster.pods)

    def trainers_nranks(self):
        return len(self.trainers_endpoints())

    def pods_nranks(self):
        return len(self.pods)

    def trainers_endpoints(self):
        r = []
        for pod in self.pods:
            for t in pod.trainers:
                r.append(t.endpoint)
        return r

103 104 105 106 107 108 109 110
    def world_device_ids(self):
        r = []
        for pod in self.pods:
            for t in pod.trainers:
                str_accelerators = [str(acc) for acc in t.accelerators]
                r.append(str_accelerators)
        return r

111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143
    def pods_endpoints(self):
        r = []
        for pod in self.pods:
            ep = "{}:{}".format(pod.addr, pod.port)
            assert pod.port != None and pod.addr != None, "{} not a valid endpoint".format(
                ep)
            r.append(ep)
        return r

    def get_pod_by_id(self, pod_id):
        for pod in self.pods:
            if str(pod_id) == str(pod.id):
                return pod

        return None


class JobServer(object):
    def __init__(self):
        self.endpoint = None

    def __str__(self):
        return "{}".format(self.endpoint)

    def __eq__(self, j):
        return self.endpint == j.endpoint

    def __ne__(self, j):
        return not self == j


class Trainer(object):
    def __init__(self):
144
        self.accelerators = []
145 146 147 148
        self.endpoint = None
        self.rank = None

    def __str__(self):
149 150
        return "accelerator:{} endpoint:{} rank:{}".format(
            self.accelerators, self.endpoint, self.rank)
151 152

    def __eq__(self, t):
153
        if len(self.accelerators) != len(t.accelerators):
154 155 156 157 158 159
            return False

        if self.endpoint != t.endpoint or \
                self.rank != t.rank:
            return False

160
        for a, b in zip(self.accelerators, t.accelerators):
161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
            if a != b:
                return False

        return True

    def __ne__(self, t):
        return not self == t

    def rank(self):
        return self.rank


class Pod(object):
    def __init__(self):
        self.rank = None
        self.id = None
        self.addr = None
        self.port = None
        self.trainers = []
180 181
        self.servers = []
        self.workers = []
182
        self.heter_workers = []
183 184
        self.accelerators = []
        self.device_mode = None
185 186

    def __str__(self):
187
        return "rank:{} id:{} addr:{} port:{} visible_accelerator:{} trainers:{} servers:{} \
188
            workers:{} heter_workers:{}".format(
189
            self.rank, self.id, self.addr, self.port, self.accelerators, [
190 191 192
                str(t) for t in self.trainers
            ], [str(s) for s in self.servers], [str(w) for w in self.workers],
            [str(h) for h in self.heter_workers])
193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212

    def __eq__(self, pod):
        if self.rank != pod.rank or \
                self.id != pod.id or \
                self.addr != pod.addr or \
                self.port != pod.port:
            logger.debug("pod {} != pod".format(self, pod))
            return False

        if len(self.trainers) != len(pod.trainers):
            logger.debug("trainers {} != {}".format(self.trainers,
                                                    pod.trainers))
            return False

        for i in range(len(self.trainers)):
            if self.trainers[i] != pod.trainers[i]:
                logger.debug("trainer {} != {}".format(self.trainers[i],
                                                       pod.trainers[i]))
                return False

213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232
        if len(self.servers) != len(pod.servers):
            logger.debug("servers {} != {}".format(self.servers, pod.servers))
            return False

        for i in range(len(self.servers)):
            if self.servers[i] != pod.servers[i]:
                logger.debug("servers {} != {}".format(self.servers[i],
                                                       pod.servers[i]))
                return False

        if len(self.workers) != len(pod.workers):
            logger.debug("workers {} != {}".format(self.workers, pod.workers))
            return False

        for i in range(len(self.workers)):
            if self.workers[i] != pod.workers[i]:
                logger.debug("workers {} != {}".format(self.workers[i],
                                                       pod.workers[i]))
                return False

233 234 235 236 237 238 239 240 241 242 243
        return True

    def __ne__(self, pod):
        return not self == pod

    def parse_response(self, res_pods):
        pass

    def rank(self):
        return self.rank

244
    def get_visible_accelerators(self):
245
        r = ""
246
        for g in self.accelerators:
247 248
            r += "{},".format(g)

249
        assert r != "", "this pod {} can't see any accelerators".format(self)
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267

        r = r[:-1]
        return r


def get_logger(log_level=20, name="root"):
    logger = logging.getLogger(name)
    logger.setLevel(log_level)

    log_handler = logging.StreamHandler()
    log_format = logging.Formatter(
        '%(levelname)s %(asctime)s %(filename)s:%(lineno)d] %(message)s')
    log_handler.setFormatter(log_format)
    logger.addHandler(log_handler)

    return logger


268 269
def get_cluster(node_ips, node_ip, trainer_endpoints, device_mode,
                devices_per_proc):
270
    assert type(trainer_endpoints) is list, "trainer_endpoints must be list"
271 272 273 274 275 276
    cluster = Cluster(hdfs=None)
    trainer_rank = 0
    for node_rank, ip in enumerate(node_ips):
        pod = Pod()
        pod.rank = node_rank
        pod.addr = ip
277 278
        pod.device_mode = device_mode

279
        cur_node_endpoints = trainer_endpoints[node_rank]
280
        # when use paddlecloud, endpoints may > devices_per_proc(user_defined)
281
        assert len(cur_node_endpoints) >= len(
282
            devices_per_proc
283
        ), "current trainer_endpoints size should be greater equal than acclerators size."
284
        for i in range(len(devices_per_proc)):
285
            trainer = Trainer()
286
            if device_mode == DeviceMode.GPU or device_mode == DeviceMode.ASCEND_NPU:
287
                if isinstance(devices_per_proc[i], (list, tuple)):
288 289
                    trainer.accelerators.extend(devices_per_proc[i])
                    pod.accelerators.extend(devices_per_proc[i])
290
                else:
291 292
                    trainer.accelerators.append(devices_per_proc[i])
                    pod.accelerators.append(devices_per_proc[i])
293 294
            elif device_mode == DeviceMode.XPU:
                if isinstance(devices_per_proc[i], (list, tuple)):
295
                    trainer.accelerators.extend(devices_per_proc[i])
296
                else:
297
                    trainer.accelerators.append(devices_per_proc[i])
298
            trainer.endpoint = "%s" % (cur_node_endpoints[i])
299 300 301 302 303 304 305 306 307 308 309 310 311 312
            trainer.rank = trainer_rank
            trainer_rank += 1

            pod.trainers.append(trainer)
        cluster.pods.append(pod)

    pod_rank = node_ips.index(node_ip)
    return cluster, cluster.pods[pod_rank]


def terminate_local_procs(procs):
    for p in procs:
        if p.proc.poll() is None:
            p.proc.terminate()
313 314
            if p.log_fn:
                p.log_fn.close()
315 316
            logger.debug("terminate process id:{}".format(p.proc.pid))

317
    # wait all process terminiated
318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387
    time.sleep(3)
    for step in range(0, 50):
        alive = False
        for p in procs:
            if p.proc.poll() is None:  # not termniate
                os.kill(p.proc.pid, signal.SIGKILL)
                alive = True

        if not alive:
            logger.info("terminate all the procs")
            return

        time.sleep(3)

    logger.fatal("can't kill all process and exit")
    exit(1)


def get_host_name_ip():
    try:
        host_name = socket.gethostname()
        host_ip = socket.gethostbyname(host_name)
        return host_name, host_ip
    except:
        return None


def add_arguments(argname, type, default, help, argparser, **kwargs):
    """Add argparse's argument.
    Usage:
    .. code-block:: python
        parser = argparse.ArgumentParser()
        add_argument("name", str, "Jonh", "User name.", parser)
        args = parser.parse_args()
    """
    type = distutils.util.strtobool if type == bool else type
    argparser.add_argument(
        "--" + argname,
        default=default,
        type=type,
        help=help + ' Default: %(default)s.',
        **kwargs)


def find_free_ports(num):
    def __free_port():
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
            s.bind(('', 0))
            return s.getsockname()[1]

    port_set = set()
    step = 0
    while True:
        port = __free_port()
        if port not in port_set:
            port_set.add(port)

        if len(port_set) >= num:
            return port_set

        step += 1
        if step > 100:
            print(
                "can't find avilable port and use the specified static port now!"
            )
            return None

    return None


388 389 390 391 392 393 394 395 396 397 398
def get_ports(num, offset):
    if os.environ.get('FLAGS_START_PORT') is None:
        ports = find_free_ports(num)
        if ports is not None:
            ports = list(ports)
    else:
        start_port = os.environ.get('FLAGS_START_PORT')
        ports = range(start_port + offset, start_port + offset + num, 1)
    return ports


399 400 401 402 403 404 405 406
def pretty_print_envs(envs, header=None):
    spacing = 2
    max_k = 40
    max_v = 45

    for k, v in envs.items():
        max_k = max(max_k, len(k))

407 408 409
    h_format = "    " + "|{{:>{}s}}{}{{:^{}s}}|\n".format(max_k, " " * spacing,
                                                          max_v)
    l_format = "    " + "|{{:>{}s}}{{}}{{:^{}s}}|\n".format(max_k, max_v)
410 411
    length = max_k + max_v + spacing

412 413
    border = "    +" + "".join(["="] * length) + "+"
    line = "    +" + "".join(["-"] * length) + "+"
414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438

    draws = ""
    draws += border + "\n"

    if header:
        draws += h_format.format(header[0], header[1])
    else:
        draws += h_format.format("fleetrun Distributed Envs", "Value")

    draws += line + "\n"

    for k, v in envs.items():
        if isinstance(v, str) and len(v) >= max_v:
            str_v = "... " + v[-41:]
        else:
            str_v = v

        draws += l_format.format(k, " " * spacing, str(str_v))

    draws += border

    _str = "\n{}\n".format(draws)
    return _str


439 440 441 442 443 444 445 446 447 448 449 450 451 452
class TrainerProc(object):
    def __init__(self):
        self.proc = None
        self.log_fn = None
        self.log_offset = None
        self.rank = None
        self.local_rank = None
        self.cmd = None


def start_local_trainers(cluster,
                         pod,
                         training_script,
                         training_script_args,
453 454 455 456 457 458 459 460
                         log_dir=None,
                         envs=None):

    if envs is None:
        current_env = copy.copy(os.environ.copy())
    else:
        current_env = copy.copy(envs)

461 462 463 464
    # paddle broadcast ncclUniqueId use socket, and
    # proxy maybe make trainers unreachable, so delete them.
    # if we set them to "", grpc will log error message "bad uri"
    # so just delete them.
465 466 467
    current_env.pop("http_proxy", None)
    current_env.pop("https_proxy", None)

468 469
    ids = cluster.world_device_ids()
    res = [':'.join(ele) for ele in ids]
470 471 472 473 474 475
    procs = []
    for idx, t in enumerate(pod.trainers):
        proc_env = {
            "PADDLE_TRAINER_ID": "%d" % t.rank,
            "PADDLE_CURRENT_ENDPOINT": "%s" % t.endpoint,
            "PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(),
476 477 478 479 480
            "PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints()),
            "PADDLE_RANK_IN_NODE": str(idx),
            "PADDLE_LOCAL_DEVICE_IDS":
            ",".join([str(acc) for acc in t.accelerators]),
            "PADDLE_WORLD_DEVICE_IDS": ",".join(res),
481 482
        }

483
        if len(t.accelerators) > 0 and pod.device_mode == DeviceMode.GPU:
484
            proc_env["FLAGS_selected_gpus"] = "%s" % ",".join(
485 486 487 488 489 490 491
                [str(g) for g in t.accelerators])

        if len(t.accelerators) > 0:
            proc_env["FLAGS_selected_accelerators"] = "%s" % ",".join(
                [str(g) for g in t.accelerators])
        # to do: same code style in future
        if fluid.core.is_compiled_with_xpu() and len(t.accelerators) > 0:
492
            proc_env["FLAGS_selected_xpus"] = "%s" % ",".join(
493
                [str(g) for g in t.accelerators])
494

495 496 497 498
        current_env.update(proc_env)

        cmd = [sys.executable, "-u", training_script] + training_script_args

499 500 501 502 503 504 505 506
        logger.debug("start trainer proc{}  env:{}".format(cmd, current_env))

        if idx == 0:
            logger.info("Local start {} processes. First process distributed "
                        "environment info (Only For Debug): {}".format(
                            len(pod.trainers),
                            pretty_print_envs(proc_env, ("Distributed Envs",
                                                         "Value"))))
507
            logger.info(
G
gongweibao 已提交
508 509
                "details abouts PADDLE_TRAINER_ENDPOINTS can be found in {}/endpoints.log, and detail running logs maybe found in {}/workerlog.0".
                format(log_dir, log_dir))
510 511 512
        fn = None
        if log_dir is not None:
            os.system("mkdir -p {}".format(log_dir))
513 514 515 516 517
            if os.path.exists("%s/endpoints.log" % log_dir):
                os.system("rm -f {}/endpoints.log".format(log_dir))
            with open("%s/endpoints.log" % log_dir, "w") as f:
                f.write("PADDLE_TRAINER_ENDPOINTS: \n")
                f.write("\n".join(cluster.trainers_endpoints()))
518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589
            fn = open("%s/workerlog.%d" % (log_dir, idx), "a")
            proc = subprocess.Popen(cmd, env=current_env, stdout=fn, stderr=fn)
        else:
            proc = subprocess.Popen(cmd, env=current_env)

        tp = TrainerProc()
        tp.proc = proc
        tp.rank = t.rank
        tp.local_rank = idx
        tp.log_fn = fn
        tp.log_offset = fn.tell() if fn else None
        tp.cmd = cmd

        procs.append(tp)

    return procs


def pull_worker_log(tp):
    if tp.log_fn:
        with open(tp.log_fn.name, 'r') as fin:
            fin.seek(tp.log_offset, 0)
            for line in fin:
                try:
                    sys.stdout.write(line)
                except UnicodeEncodeError:
                    sys.stdout.write(
                        'UnicodeEncodeError occurs at this line. '
                        'Please refer to the original log file "%s"\n' %
                        tp.log_fn.name)
            tp.log_offset = fin.tell()


def watch_local_trainers(procs, nranks):
    try:
        error = False
        error_rank = []
        # wait all process finish or one error
        alive = False
        for p in procs:
            if p.log_fn and p.local_rank == 0:
                pull_worker_log(p)

            ret = p.proc.poll()
            if ret is None:
                alive = True
            elif ret != 0:
                error = True
                error_rank.append(p.rank)

        if error:
            terminate_local_procs(procs)
            exit(1)

    except KeyboardInterrupt:
        logger.warning("KeyboardInterrupt, exit")
        terminate_local_procs(procs)
        raise
    except SystemExit:
        logger.error(
            "ABORT!!! Out of all {} trainers, the trainer process with rank={} was aborted. Please check its log.".
            format(nranks, error_rank))
        terminate_local_procs(procs)
        raise
    except:
        logger.error(
            "ABORT!!! Out of all {} trainers, the trainer process with rank={} was aborted. Please check its log.".
            format(nranks, error_rank))
        terminate_local_procs(procs)
        raise

    return alive
590 591


592 593 594 595
def get_ascend_npus(npus):
    if npus is None:
        count = fluid.core.NPUDevice.get_device_count()
        if count <= 0:
596 597
            return None
        ret = [str(x) for x in range(count)]
598 599 600 601 602
    else:
        ret = [x.strip() for x in npus.split(',')]
    return ret


603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631
def get_gpus(gpus):
    if gpus is None:
        gpus_num = fluid.core.get_cuda_device_count()
        res_gpus = [str(x) for x in range(0, gpus_num)]
    else:
        cuda_visible_devices = os.getenv("CUDA_VISIBLE_DEVICES")
        if cuda_visible_devices is None or cuda_visible_devices == "":
            res_gpus = [x.strip() for x in gpus.split(',')]
        else:
            # change gpus into relative values
            # e.g. CUDA_VISIBLE_DEVICES=4,5,6,7; args.gpus=4,5,6,7;
            # therefore gpus=0,1,2,3
            cuda_visible_devices_list = cuda_visible_devices.split(',')
            for x in gpus.split(','):
                assert x in cuda_visible_devices_list, "Can't find "\
                    "your gpus %s in CUDA_VISIBLE_DEVICES[%s]."\
                    % (x, cuda_visible_devices)
            res_gpus = [
                cuda_visible_devices_list.index(x.strip())
                for x in gpus.split(',')
            ]
            logger.info("Change selected_gpus into reletive values. --ips:{} "
                        "will change into relative_ips:{} according to your "
                        "CUDA_VISIBLE_DEVICES:{}".format(
                            gpus, res_gpus, cuda_visible_devices_list))

    return res_gpus


632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660
def get_xpus(xpus):
    if xpus is None:
        xpus_num = fluid.core.get_xpu_device_count()
        res_xpus = [str(x) for x in range(0, xpus_num)]
    else:
        xpu_visible_devices = os.getenv("XPU_VISIBLE_DEVICES")
        if xpu_visible_devices is None or xpu_visible_devices == "":
            res_xpus = [x.strip() for x in xpus.split(',')]
        else:
            # change xpus into relative values
            # e.g. XPU_VISIBLE_DEVICES=4,5,6,7; args.xpus=4,5,6,7;
            # therefore xpus=0,1,2,3
            xpu_visible_devices_list = xpu_visible_devices.split(',')
            for x in xpus.split(','):
                assert x in xpu_visible_devices_list, "Can't find "\
                    "your xpus %s in XPU_VISIBLE_DEVICES[%s]."\
                    % (x, xpu_visible_devices)
            res_xpus = [
                xpu_visible_devices_list.index(x.strip())
                for x in xpus.split(',')
            ]
            logger.info("Change selected_xpus into reletive values. --ips:{} "
                        "will change into relative_ips:{} according to your "
                        "XPU_VISIBLE_DEVICES:{}".format(
                            xpus, res_xpus, xpu_visible_devices_list))

    return res_xpus


661
def get_device_mode():
662 663 664 665 666 667 668 669
    if fluid.core.is_compiled_with_ascend() and \
            fluid.core.NPUDevice.get_device_count() > 0:
        print("launch train in ascend npu mode!")
        return DeviceMode.ASCEND_NPU

    if fluid.core.is_compiled_with_cuda() and \
            fluid.core.get_cuda_device_count() > 0:
        print("launch train in GPU mode!")
670
        return DeviceMode.GPU
671 672

    if fluid.core.is_compiled_with_xpu() and fluid.core.get_xpu_device_count(
673 674 675
    ) > 0:
        print("launch train in XPU mode")
        return DeviceMode.XPU
676

677 678
    print("launch train in CPU mode")
    return DeviceMode.CPU
679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698


def get_device_proc_info(args):
    # device_mode
    device_mode = get_device_mode()

    # devices
    devices_per_proc = []
    if device_mode == DeviceMode.GPU:
        gpus = get_gpus(args.gpus)
        if args.nproc_per_node is not None:
            assert (len(gpus) % int(args.nproc_per_node)) ==0, \
                "gpus' number:{} mod args.nproc_per_node:{} must == 0".format(len(gpus), arg.nproc_per_node)

            n = int(len(gpus) / int(args.nproc_per_node))
            devices_per_proc = [
                gpus[i:i + n] for i in six.moves.range(0, len(gpus), n)
            ]
        else:
            devices_per_proc = gpus
699 700 701 702
    elif device_mode == DeviceMode.ASCEND_NPU:
        npus = get_ascend_npus(args.ascend_npus)
        assert args.nproc_per_node is None, "ascend_npus need't nproc_per_node arguments"
        devices_per_proc = npus
703 704 705 706 707 708 709 710 711 712 713 714
    elif device_mode == DeviceMode.XPU:
        xpus = get_xpus(args.xpus)
        if args.nproc_per_node is not None:
            assert (len(xpus) % int(args.nproc_per_node)) == 0, \
                "xpus' number:{} mod args.nproc_per_node:{} must == 0".format(len(xpus), arg.nproc_per_node)

            n = int(len(xpus) / int(args.nproc_per_node))
            devices_per_proc = [
                xpus[i:i + n] for i in six.moves.range(0, len(xpus), n)
            ]
        else:
            devices_per_proc = xpus
715 716 717 718 719 720
    elif device_mode == DeviceMode.CPU:
        if args.nproc_per_node is None:
            devices_per_proc = [0]
        else:
            devices_per_proc = [x for x in range(0, args.nproc_per_node)]
    else:
721
        assert False, "Can't support device_mode:{}, support only cpu|gpu|xpu now.".format(
722 723 724 725 726
            device_mode)

    return (device_mode, devices_per_proc)


727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764
def direct_start(args):
    # run ps-cpu mode on paddlecloud, using given envs
    cmd = [sys.executable, "-u", args.training_script] + \
        args.training_script_args
    proc = subprocess.Popen(cmd)
    proc.wait()
    return


def get_custom_endpoints(origin_endpoints, offset=0):
    """
    origin_endpoint: ip:port
    user_define_endpoint: ip:(port+offset)
    """
    assert origin_endpoints != None
    paddle_user_define_endpoints_list = []
    for ip_port in origin_endpoints.split(","):
        ip = ip_port.split(":")[0]
        port = ip_port.split(":")[1]
        new_port = int(port) + offset
        paddle_user_define_endpoints_list.append(":".join((ip, str(new_port))))
    paddle_user_define_endpoints = ",".join(paddle_user_define_endpoints_list)
    return paddle_user_define_endpoints


def cloud_ps_heter_env_set(args):
    environs = {}

    paddle_trainer_endpoints = os.getenv("TRAINER_IP_PORT_LIST", "")
    assert paddle_trainer_endpoints != None

    paddle_pserver_endpoints = os.getenv("PSERVER_IP_PORT_LIST", "")
    assert paddle_pserver_endpoints != None

    # hard code for paddlecloud custom-framework
    avilable_ports = os.getenv("TRAINER_PORTS", "").split(",")
    assert len(
        avilable_ports
765
    ) >= 2, "set paddle_ports_num >= 2 in config.ini for paddlecloud job submit"
766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874

    # hard code for paddlecloud custom-framework
    trainers_num = len(paddle_pserver_endpoints.split(","))
    assert trainers_num != 0
    environs["PADDLE_TRAINERS_NUM"] = trainers_num
    environs["TRAINERS_NUM"] = trainers_num

    # hard code for paddlecloud custom-framework
    environs["PADDLE_HETER_TRAINER_IP_PORT_LIST"] = paddle_trainer_endpoints
    environs["PADDLE_PSERVERS_IP_PORT_LIST"] = paddle_pserver_endpoints
    environs["PADDLE_TRAINER_ENDPOINTS"] = get_custom_endpoints(
        paddle_pserver_endpoints, 1)
    heter_worker_num = len(paddle_trainer_endpoints.split(","))
    if (args.heter_worker_num != None) and (
            heter_worker_num != args.heter_worker_num):
        warnings.warn(
            "Your fleetrun setting: heter_worker_num is {}, but we find {} device can be used, this setting has been changed.".
            format(args.heter_worker_num, heter_worker_num))
        args.heter_worker_num = heter_worker_num

    for k, v in environs.items():
        os.environ[k] = str(v)
    logger.info("Set heter parameter server env: {}".format(
        pretty_print_envs(environs)))


class ParameterServerLauncher(object):
    def __init__(self, args, distribute_mode):
        self.args = args
        self.distribute_mode = distribute_mode
        self.server_num = 0
        self.worker_num = 0
        self.heter_worker_num = 0

        self.server_endpoints = ""
        self.server_endpoints_ips = []
        self.server_endpoints_port = []

        self.worker_endpoints = ""
        self.worker_endpoints_ips = []
        self.worker_endpoints_port = []

        self.heter_worker_endpoints = ""
        self.heter_worker_endpoints_ips = []
        self.heter_worker_endpoints_port = []

        self.is_local = True
        self.current_node_ip = ""

        self.get_role_endpoints(args)

    def get_role_endpoints(self, args):
        # get server envs
        if args.server_num:
            self.server_num = args.server_num
            if args.servers:
                assert len(
                    args.servers.split(",")
                ) == self.server_num, "The server_num and servers doesn't match. Expect servers endpoints num epual to server_num, but received servers enpoint num: {} and server_num {}".format(
                    len(args.servers.split(",")), self.server_num)
                self.server_endpoints = args.servers
            else:
                ports = get_ports(self.server_num, 0)
                self.server_endpoints = ",".join(
                    ["127.0.0.1:" + str(x) for x in ports])
        else:
            assert args.servers != "", "The setting of Parameter-Server must has server_num or servers."
            self.server_endpoints = args.servers
            self.server_num = len(self.server_endpoints.split(","))

        # get worker envs
        if args.worker_num:
            self.worker_num = args.worker_num
            if args.workers:
                assert len(
                    args.workers.split(",")
                ) == self.worker_num, "The worker_num and workers doesn't match. Expect workers endpoints num epual to worker_num, but received workers enpoint num: {} and worker_num {}".format(
                    len(args.workers.split(",")), self.worker_num)

                self.worker_endpoints = args.workers
            else:
                ports = get_ports(self.worker_num, self.server_num)
                self.worker_endpoints = ",".join(
                    ["127.0.0.1:" + str(x) for x in ports])
        else:
            assert args.workers != "", "The setting of Parameter-Server must has worker_num or workers."
            worker_endpoints_ips = [
                x.strip().split(":")[0] for x in args.workers.split(",")
            ]
            self.worker_num = len(worker_endpoints_ips)
            worker_endpoints_len = [
                len(x.strip().split(":")) for x in args.workers.split(",")
            ]

            if 1 in worker_endpoints_len:
                # if no port value in worker_endpoints, will set default port values.
                start_port = 6170
                worker_endpoints_port = range(
                    start_port + self.server_num,
                    start_port + self.server_num + self.worker_num, 1)
                # create endpoints str
                worker_endpoints = []
                for i in range(self.worker_num):
                    worker_endpoints.append(":".join((worker_endpoints_ips[
                        i], str(worker_endpoints_port[i]))))
                self.worker_endpoints = ",".join(worker_endpoints)
            else:
                self.worker_endpoints = args.workers

875 876 877 878 879 880 881 882
        # get http_port
        if args.http_port:
            self.http_port = args.http_port
        else:
            http_port = get_ports(1, self.server_num + self.worker_num)
            http_ip = self.server_endpoints.split(",")[0].split(":")[0]
            self.http_port = http_ip + ":" + str(http_port[0])

883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996
        # get heter worker envs
        if self.distribute_mode == DistributeMode.PS_HETER:
            if args.heter_worker_num:
                self.heter_worker_num = args.heter_worker_num
                if args.heter_workers:
                    assert len(
                        args.heter_workers.split(",")
                    ) == self.heter_worker_num, "The heter_worker_num and heter_workers doesn't match. Expect heter_workers endpoints num epual to heter_worker_num, but received heter_workers enpoint num: {} and heter_worker_num {}".format(
                        len(args.heter_workers.split(",")),
                        self.heter_worker_num)
                    self.heter_worker_endpoints = args.heter_workers
                else:
                    ports = get_ports(self.heter_worker_num,
                                      self.server_num + self.worker_num)
                    self.heter_worker_endpoints = ",".join(
                        ["127.0.0.1:" + str(x) for x in ports])
            else:
                assert args.heter_workers != "", "The setting of Parameter-Server heter mode must has heter_worker_num or heter_workers."
                self.heter_worker_endpoints = args.heter_workers
                self.heter_worker_num = len(
                    self.heter_worker_endpoints.split(","))

        # check local or user define
        self.server_endpoints_ips = [
            x.strip().split(":")[0] for x in self.server_endpoints.split(",")
        ]
        self.worker_endpoints_ips = [
            x.strip().split(":")[0] for x in self.worker_endpoints.split(",")
        ]
        self.server_endpoints_port = [
            x.strip().split(":")[1] for x in self.server_endpoints.split(",")
        ]
        self.worker_endpoints_port = [
            x.strip().split(":")[1] for x in self.worker_endpoints.split(",")
        ]
        self.node_ips = list(
            set(self.server_endpoints_ips + self.worker_endpoints_ips))
        if self.distribute_mode == DistributeMode.PS_HETER:
            self.heter_worker_endpoints_ips = [
                x.strip().split(":")[0]
                for x in self.heter_worker_endpoints.split(",")
            ]
            self.heter_worker_endpoints_port = [
                x.strip().split(":")[1]
                for x in self.heter_worker_endpoints.split(",")
            ]
            self.node_ips = list(
                set(self.node_ips + self.heter_worker_endpoints_ips))

        if len(set(self.node_ips)) == 1:
            self.is_local = True
            self.current_node_ip = self.node_ips[0]
        else:
            self.is_local = False
            pod_ip = os.getenv("POD_IP", None)
            if pod_ip == None:
                _, self.current_node_ip = get_host_name_ip()
            else:
                self.current_node_ip = pod_ip
            assert self.current_node_ip in self.node_ips, "Can't find your local ip {%s} in args.servers and args.workers ips: {%s}" \
                % (self.current_node_ip, self.node_ips)
        self.node_rank = self.node_ips.index(self.current_node_ip)

        logger.debug(
            "parsed from args: node_ips:{} current_node_ip:{} node_rank:{}".
            format(self.node_ips, self.current_node_ip, self.node_rank))

    def start_ps(self):
        cluster = Cluster(hdfs=None)
        server_rank = 0
        worker_rank = 0
        heter_worker_rank = 0

        for node_rank, ip in enumerate(self.node_ips):
            pod = Pod()
            pod.rank = node_rank
            pod.addr = ip
            for i in range(len(self.server_endpoints_ips)):
                if ip == self.server_endpoints_ips[i]:
                    server = Trainer()
                    server.endpoint = "%s:%s" % (ip,
                                                 self.server_endpoints_port[i])
                    server.rank = server_rank
                    server_rank += 1
                    pod.servers.append(server)
            for j in range(len(self.worker_endpoints_ips)):
                if ip == self.worker_endpoints_ips[j]:
                    worker = Trainer()
                    worker.endpoint = "%s:%s" % (ip,
                                                 self.worker_endpoints_port[j])
                    worker.rank = worker_rank
                    worker_rank += 1
                    pod.workers.append(worker)
            for k in range(len(self.heter_worker_endpoints_ips)):
                if ip == self.heter_worker_endpoints_ips[k]:
                    heter_worker = Trainer()
                    heter_worker.endpoint = "%s:%s" % (
                        ip, self.heter_worker_endpoints_port[k])
                    heter_worker.rank = heter_worker_rank
                    heter_worker_rank += 1
                    pod.heter_workers.append(heter_worker)

            cluster.pods.append(pod)

        pod = cluster.pods[self.node_rank]
        self.gloo_rendezvous_dir = tempfile.mkdtemp()

        # 3. subproces start
        self.procs = {"worker": [], "server": [], "heter_worker": []}
        self.cmds = {"worker": [], "server": [], "heter_worker": []}
        self.log_fns = {"worker": [], "server": [], "heter_worker": []}

        self.start_pod_server(self.args, pod)
        self.start_pod_worker(self.args, pod)
997 998
        if self.distribute_mode == DistributeMode.PS_HETER:
            self.start_pod_heter_worker(self.args, pod)
999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055

        logger.info(
            "Please check servers, workers and heter_worker logs in {}/workerlog.*, {}/serverlog.* and {}/heterlog.*".
            format(self.args.log_dir, self.args.log_dir, self.args.log_dir))

        # 4. wait for finish training
        if len(self.procs["worker"]) > 0:
            # if node has worker procs
            # only wait worker to finish here
            for i, proc in enumerate(self.procs["worker"]):
                self.procs["worker"][i].proc.wait()
                if len(self.log_fns["worker"]) > 0:
                    self.log_fns["worker"][i].close()
            logger.info(
                "all workers exit, going to finish parameter server and heter_worker."
            )
            if len(self.procs["heter_worker"]) > 0:
                for i, proc in enumerate(self.procs["heter_worker"]):
                    self.log_fns["heter_worker"][i].close()
                    self.procs["heter_worker"][i].proc.terminate()
                logger.info("all heter_worker are killed")

            if len(self.procs["server"]) > 0:
                for i, proc in enumerate(self.procs["server"]):
                    self.log_fns["server"][i].close()
                    self.procs["server"][i].proc.terminate()
                logger.info("all parameter server are killed")

        else:
            # if node has not worker procs
            # blocking training process
            if len(self.procs["server"]) > 0:
                for i, proc in enumerate(self.procs["server"]):
                    self.procs["server"][i].proc.wait()

            if len(self.procs["heter_worker"]) > 0:
                for i, proc in enumerate(self.procs["heter_worker"]):
                    self.procs["heter_worker"][i].proc.wait()

        if os.path.exists(self.gloo_rendezvous_dir):
            shutil.rmtree(self.gloo_rendezvous_dir)

    def start_pod_server(self, args, pod):
        default_env = os.environ.copy()
        current_env = copy.copy(default_env)
        current_env.pop("http_proxy", None)
        current_env.pop("https_proxy", None)
        for idx, cur_server in enumerate(pod.servers):
            proc_env = {
                "PADDLE_PSERVERS_IP_PORT_LIST": self.server_endpoints,
                "PADDLE_TRAINER_ENDPOINTS": self.worker_endpoints,
                "PADDLE_HETER_TRAINER_IP_PORT_LIST":
                self.heter_worker_endpoints,
                "PADDLE_PORT": cur_server.endpoint.split(":")[1],
                "TRAINING_ROLE": "PSERVER",
                "PADDLE_TRAINERS_NUM": str(self.worker_num),
                "POD_IP": cur_server.endpoint.split(":")[0],
L
lilong12 已提交
1056
                "PADDLE_WITH_GLOO": str(os.getenv("PADDLE_WITH_GLOO", "0")),
1057
                "PADDLE_GLOO_RENDEZVOUS": "3",
1058 1059
                "PADDLE_GLOO_FS_PATH": self.gloo_rendezvous_dir,
                "PADDLE_GLOO_HTTP_ENDPOINT": self.http_port
1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109
            }
            current_env.update(proc_env)

            cmd = [sys.executable, "-u", args.training_script
                   ] + args.training_script_args
            self.cmds["server"].append(cmd)

            if idx == 0:
                logger.info(
                    "Local server start {} processes. First process distributed "
                    "environment info (Only For Debug): {}".format(
                        len(pod.servers),
                        pretty_print_envs(proc_env, ("Distributed Envs", "Value"
                                                     ))))

            if args.log_dir is not None:
                os.system("mkdir -p {}".format(args.log_dir))
                fn = open("%s/serverlog.%d" % (args.log_dir, idx), "w")
                self.log_fns["server"].append(fn)
                proc = subprocess.Popen(
                    cmd, env=current_env, stdout=fn, stderr=fn)
            else:
                proc = subprocess.Popen(cmd, env=current_env)

            tp = TrainerProc()
            tp.proc = proc
            tp.rank = cur_server.rank
            tp.local_rank = idx
            tp.log_fn = fn
            tp.log_offset = fn.tell() if fn else None
            tp.cmd = cmd

            self.procs["server"].append(tp)

    def start_pod_worker(self, args, pod):
        default_env = os.environ.copy()
        current_env = copy.copy(default_env)
        current_env.pop("http_proxy", None)
        current_env.pop("https_proxy", None)

        heter_device_num = 0
        device_list = []
        if fluid.core.is_compiled_with_cuda():
            device_list = get_gpus(args.gpus)
            heter_device_num = len(device_list)
        elif fluid.core.is_compiled_with_xpu():
            heter_device_num = fluid.core.get_xpu_device_count()
            device_list = [str(x) for x in range(0, heter_device_num)]

        for idx, cur_worker in enumerate(pod.workers):
1110 1111
            device_id = "0" if heter_device_num == 0 else str(device_list[
                idx % heter_device_num])
1112 1113 1114 1115 1116 1117 1118 1119
            proc_env = {
                "PADDLE_PSERVERS_IP_PORT_LIST": self.server_endpoints,
                "PADDLE_TRAINER_ENDPOINTS": self.worker_endpoints,
                "PADDLE_TRAINERS_NUM": str(self.worker_num),
                "PADDLE_HETER_TRAINER_IP_PORT_LIST":
                self.heter_worker_endpoints,
                "TRAINING_ROLE": "TRAINER",
                "PADDLE_TRAINER_ID": str(cur_worker.rank),
L
lilong12 已提交
1120
                "PADDLE_WITH_GLOO": str(os.getenv("PADDLE_WITH_GLOO", "0")),
1121
                "PADDLE_GLOO_RENDEZVOUS": "3",
1122 1123 1124 1125 1126
                "PADDLE_GLOO_FS_PATH": self.gloo_rendezvous_dir,
                "FLAGS_selected_gpus": "0",
                "FLAGS_selected_xpus": "0",
                "CUDA_VISIBLE_DEVICES": device_id,
                "XPU_VISIBLE_DEVICES": device_id,
1127
                "PADDLE_GLOO_HTTP_ENDPOINT": self.http_port
1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175
            }
            current_env.update(proc_env)

            cmd = [sys.executable, "-u", args.training_script
                   ] + args.training_script_args
            self.cmds["worker"].append(cmd)

            if idx == 0:
                logger.info(
                    "Local worker start {} processes. First process distributed "
                    "environment info (Only For Debug): {}".format(
                        len(pod.workers),
                        pretty_print_envs(proc_env, ("Distributed Envs", "Value"
                                                     ))))

            if args.log_dir is not None:
                os.system("mkdir -p {}".format(args.log_dir))
                fn = open("%s/workerlog.%d" % (args.log_dir, idx), "w")
                self.log_fns["worker"].append(fn)
                proc = subprocess.Popen(
                    cmd, env=current_env, stdout=fn, stderr=fn)
            else:
                proc = subprocess.Popen(cmd, env=current_env)

            tp = TrainerProc()
            tp.proc = proc
            tp.rank = cur_worker.rank
            tp.local_rank = idx
            tp.log_fn = fn
            tp.log_offset = fn.tell() if fn else None
            tp.cmd = cmd

            self.procs["worker"].append(tp)

    def start_pod_heter_worker(self, args, pod):
        default_env = os.environ.copy()
        current_env = copy.copy(default_env)
        current_env.pop("http_proxy", None)
        current_env.pop("https_proxy", None)

        heter_device_num = 0
        device_list = []
        if fluid.core.is_compiled_with_cuda():
            device_list = get_gpus(args.gpus)
            heter_device_num = len(device_list)
        elif fluid.core.is_compiled_with_xpu():
            heter_device_num = fluid.core.get_xpu_device_count()
            device_list = [str(x) for x in range(0, heter_device_num)]
1176 1177
        if heter_device_num == 0:
            return
1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189

        for idx, cur_heter_worker in enumerate(pod.heter_workers):
            device_id = str(device_list[idx % heter_device_num])
            proc_env = {
                "PADDLE_PSERVERS_IP_PORT_LIST": self.server_endpoints,
                "PADDLE_TRAINER_ENDPOINTS": self.worker_endpoints,
                "PADDLE_HETER_TRAINER_IP_PORT_LIST":
                self.heter_worker_endpoints,
                "PADDLE_PORT": cur_heter_worker.endpoint.split(":")[1],
                "TRAINING_ROLE": "HETER_TRAINER",
                "PADDLE_TRAINERS_NUM": str(self.worker_num),
                "POD_IP": cur_heter_worker.endpoint.split(":")[0],
L
lilong12 已提交
1190
                "PADDLE_WITH_GLOO": str(os.getenv("PADDLE_WITH_GLOO", "0")),
1191
                "PADDLE_GLOO_RENDEZVOUS": "3",
1192 1193 1194 1195 1196
                "PADDLE_GLOO_FS_PATH": self.gloo_rendezvous_dir,
                "FLAGS_selected_gpus": "0",
                "FLAGS_selected_xpus": "0",
                "CUDA_VISIBLE_DEVICES": device_id,
                "XPU_VISIBLE_DEVICES": device_id,
1197
                "PADDLE_GLOO_HTTP_ENDPOINT": self.http_port
1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230
            }
            current_env.update(proc_env)

            cmd = [sys.executable, "-u", args.training_script
                   ] + args.training_script_args
            self.cmds["heter_worker"].append(cmd)

            if idx == 0:
                logger.info(
                    "Local heter_worker start {} processes. First process distributed "
                    "environment info (Only For Debug): {}".format(
                        len(pod.heter_workers),
                        pretty_print_envs(proc_env, ("Distributed Envs", "Value"
                                                     ))))

            if args.log_dir is not None:
                os.system("mkdir -p {}".format(args.log_dir))
                fn = open("%s/heterlog.%d" % (args.log_dir, idx), "w")
                self.log_fns["heter_worker"].append(fn)
                proc = subprocess.Popen(
                    cmd, env=current_env, stdout=fn, stderr=fn)
            else:
                proc = subprocess.Popen(cmd, env=current_env)

            tp = TrainerProc()
            tp.proc = proc
            tp.rank = cur_heter_worker.rank
            tp.local_rank = idx
            tp.log_fn = fn
            tp.log_offset = fn.tell() if fn else None
            tp.cmd = cmd

            self.procs["heter_worker"].append(tp)