run.py 15.8 KB
Newer Older
T
tangwei 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
# Copyright (c) 2020 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.

T
tangwei 已提交
15
import os
T
tangwei 已提交
16
import subprocess
X
test  
xjqbest 已提交
17
import sys
T
tangwei 已提交
18 19
import argparse
import tempfile
C
Chengmo 已提交
20

T
tangwei 已提交
21
import yaml
X
fix  
xjqbest 已提交
22
import copy
23 24
from paddlerec.core.factory import TrainerFactory
from paddlerec.core.utils import envs
C
Chengmo 已提交
25
from paddlerec.core.utils import validation
26
from paddlerec.core.utils import util
X
test  
xjqbest 已提交
27
from paddlerec.core.utils import validation
T
tangwei 已提交
28

T
tangwei 已提交
29 30
engines = {}
device = ["CPU", "GPU"]
T
tangwei 已提交
31
engine_choices = [
C
Chengmo 已提交
32 33
    "TRAIN", "SINGLE_TRAIN", "INFER", "SINGLE_INFER", "LOCAL_CLUSTER",
    "LOCAL_CLUSTER_TRAIN", "CLUSTER_TRAIN"
T
tangwei 已提交
34
]
T
tangwei 已提交
35 36


T
tangwei 已提交
37
def engine_registry():
T
tangwei 已提交
38 39 40
    engines["TRANSPILER"] = {}
    engines["PSLIB"] = {}

C
Chengmo 已提交
41
    engines["TRANSPILER"]["TRAIN"] = single_train_engine
X
fix  
xjqbest 已提交
42
    engines["TRANSPILER"]["SINGLE_TRAIN"] = single_train_engine
C
Chengmo 已提交
43
    engines["TRANSPILER"]["INFER"] = single_infer_engine
X
fix  
xjqbest 已提交
44
    engines["TRANSPILER"]["SINGLE_INFER"] = single_infer_engine
T
tangwei 已提交
45
    engines["TRANSPILER"]["LOCAL_CLUSTER"] = local_cluster_engine
C
Chengmo 已提交
46
    engines["TRANSPILER"]["LOCAL_CLUSTER_TRAIN"] = local_cluster_engine
T
tangwei 已提交
47
    engines["TRANSPILER"]["CLUSTER"] = cluster_engine
C
Chengmo 已提交
48 49 50
    engines["PSLIB"]["SINGLE_TRAIN"] = local_mpi_engine
    engines["PSLIB"]["TRAIN"] = local_mpi_engine
    engines["PSLIB"]["LOCAL_CLUSTER_TRAIN"] = local_mpi_engine
T
tangwei 已提交
51
    engines["PSLIB"]["LOCAL_CLUSTER"] = local_mpi_engine
C
Chengmo 已提交
52
    engines["PSLIB"]["CLUSTER_TRAIN"] = cluster_mpi_engine
T
tangwei 已提交
53
    engines["PSLIB"]["CLUSTER"] = cluster_mpi_engine
T
tangwei 已提交
54

T
tangwei 已提交
55

X
fix  
xjqbest 已提交
56
def get_inters_from_yaml(file, filters):
X
test  
xjqbest 已提交
57
    _envs = envs.load_yaml(file)
T
tangwei 已提交
58 59 60
    flattens = envs.flatten_environs(_envs)
    inters = {}
    for k, v in flattens.items():
X
fix  
xjqbest 已提交
61 62 63
        for f in filters:
            if k.startswith(f):
                inters[k] = v
T
tangwei 已提交
64
    return inters
T
tangwei 已提交
65 66


X
fix  
xjqbest 已提交
67
def get_all_inters_from_yaml(file, filters):
C
Chengmo 已提交
68
    _envs = envs.load_yaml(file)
X
fix  
xjqbest 已提交
69 70 71 72 73 74 75 76 77 78 79 80
    all_flattens = {}

    def fatten_env_namespace(namespace_nests, local_envs):
        for k, v in local_envs.items():
            if isinstance(v, dict):
                nests = copy.deepcopy(namespace_nests)
                nests.append(k)
                fatten_env_namespace(nests, v)
            elif (k == "dataset" or k == "phase" or
                  k == "runner") and isinstance(v, list):
                for i in v:
                    if i.get("name") is None:
C
Chengmo 已提交
81
                        raise ValueError("name must be in dataset list. ", v)
X
fix  
xjqbest 已提交
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
                    nests = copy.deepcopy(namespace_nests)
                    nests.append(k)
                    nests.append(i["name"])
                    fatten_env_namespace(nests, i)
            else:
                global_k = ".".join(namespace_nests + [k])
                all_flattens[global_k] = v

    fatten_env_namespace([], _envs)
    ret = {}
    for k, v in all_flattens.items():
        for f in filters:
            if k.startswith(f):
                ret[k] = v
    return ret


C
chengmo 已提交
99
def get_engine(args):
T
tangwei 已提交
100
    transpiler = get_transpiler()
X
fix  
xjqbest 已提交
101
    with open(args.model, 'r') as rb:
C
Chengmo 已提交
102
        _envs = yaml.load(rb.read(), Loader=yaml.FullLoader)
X
fix  
xjqbest 已提交
103
    run_extras = get_all_inters_from_yaml(args.model, ["train.", "runner."])
X
fix  
xjqbest 已提交
104 105 106

    engine = run_extras.get("train.engine", None)
    if engine is None:
C
Chengmo 已提交
107
        engine = run_extras.get("runner." + _envs["mode"] + ".class", None)
X
fix  
xjqbest 已提交
108
    if engine is None:
C
Chengmo 已提交
109 110 111 112 113 114 115 116 117 118
        engine = "train"

    device = run_extras.get("runner." + _envs["mode"] + ".device", "CPU")
    if device.upper() == "GPU":
        selected_gpus = run_extras.get(
            "runner." + _envs["mode"] + ".selected_gpus", "0")
        selected_gpus_num = len(selected_gpus.split(","))
        if selected_gpus_num > 1:
            engine = "LOCAL_CLUSTER"

T
tangwei 已提交
119 120
    engine = engine.upper()
    if engine not in engine_choices:
C
Chengmo 已提交
121
        raise ValueError("runner.class can not be chosen in {}".format(
T
tangwei 已提交
122
            engine_choices))
T
tangwei 已提交
123

T
tangwei 已提交
124
    print("engines: \n{}".format(engines))
T
tangwei 已提交
125

T
tangwei 已提交
126
    run_engine = engines[transpiler].get(engine, None)
T
tangwei 已提交
127

T
tangwei 已提交
128 129 130 131
    return run_engine


def get_transpiler():
T
tangwei 已提交
132
    FNULL = open(os.devnull, 'w')
T
tangwei 已提交
133 134 135 136
    cmd = [
        "python", "-c",
        "import paddle.fluid as fluid; fleet_ptr = fluid.core.Fleet(); [fleet_ptr.copy_table_by_feasign(10, 10, [2020, 1010])];"
    ]
T
tangwei 已提交
137 138 139
    proc = subprocess.Popen(cmd, stdout=FNULL, stderr=FNULL, cwd=os.getcwd())
    ret = proc.wait()
    if ret == -11:
T
tangwei 已提交
140
        return "PSLIB"
T
tangwei 已提交
141
    else:
T
tangwei 已提交
142
        return "TRANSPILER"
T
tangwei 已提交
143 144


T
tangwei 已提交
145 146 147
def set_runtime_envs(cluster_envs, engine_yaml):
    if cluster_envs is None:
        cluster_envs = {}
T
tangwei 已提交
148

T
tangwei 已提交
149
    engine_extras = get_inters_from_yaml(engine_yaml, "train.trainer.")
150 151
    if "train.trainer.threads" in engine_extras and "CPU_NUM" in cluster_envs:
        cluster_envs["CPU_NUM"] = engine_extras["train.trainer.threads"]
T
tangwei 已提交
152

T
tangwei 已提交
153
    envs.set_runtime_environs(cluster_envs)
154
    envs.set_runtime_environs(engine_extras)
T
fix bug  
tangwei 已提交
155 156 157

    need_print = {}
    for k, v in os.environ.items():
T
tangwei 已提交
158
        if k.startswith("train.trainer."):
T
fix bug  
tangwei 已提交
159 160 161
            need_print[k] = v

    print(envs.pretty_print_envs(need_print, ("Runtime Envs", "Value")))
T
tangwei 已提交
162 163


C
Chengmo 已提交
164 165 166 167 168
def single_train_engine(args):
    _envs = envs.load_yaml(args.model)
    run_extras = get_all_inters_from_yaml(args.model, ["train.", "runner."])
    trainer_class = run_extras.get(
        "runner." + _envs["mode"] + ".trainer_class", None)
T
tangwei 已提交
169

C
Chengmo 已提交
170 171 172 173 174 175 176 177 178 179 180 181 182 183
    if trainer_class:
        trainer = trainer_class
    else:
        trainer = "GeneralTrainer"

    executor_mode = "train"
    fleet_mode = run_extras.get("runner." + _envs["mode"] + ".fleet_mode",
                                "ps")
    device = run_extras.get("runner." + _envs["mode"] + ".device", "cpu")
    selected_gpus = run_extras.get(
        "runner." + _envs["mode"] + ".selected_gpus", "0")
    selected_gpus_num = len(selected_gpus.split(","))
    if device.upper() == "GPU":
        assert selected_gpus_num == 1, "Single Mode Only Support One GPU, Set Local Cluster Mode to use Multi-GPUS"
C
chengmo 已提交
184 185

    single_envs = {}
C
Chengmo 已提交
186 187
    single_envs["selsected_gpus"] = selected_gpus
    single_envs["FLAGS_selected_gpus"] = selected_gpus
C
chengmo 已提交
188
    single_envs["train.trainer.trainer"] = trainer
C
Chengmo 已提交
189 190
    single_envs["fleet_mode"] = fleet_mode
    single_envs["train.trainer.executor_mode"] = executor_mode
C
chengmo 已提交
191 192
    single_envs["train.trainer.threads"] = "2"
    single_envs["train.trainer.platform"] = envs.get_platform()
C
Chengmo 已提交
193 194
    single_envs["train.trainer.engine"] = "single"

X
fix  
xjqbest 已提交
195 196 197
    set_runtime_envs(single_envs, args.model)
    trainer = TrainerFactory.create(args.model)
    return trainer
X
fix  
xjqbest 已提交
198

X
fix  
xjqbest 已提交
199 200

def single_infer_engine(args):
C
Chengmo 已提交
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221
    _envs = envs.load_yaml(args.model)
    run_extras = get_all_inters_from_yaml(args.model, ["train.", "runner."])
    trainer_class = run_extras.get(
        "runner." + _envs["mode"] + ".trainer_class", None)

    if trainer_class:
        trainer = trainer_class
    else:
        trainer = "GeneralTrainer"

    executor_mode = "infer"
    fleet_mode = run_extras.get("runner." + _envs["mode"] + ".fleet_mode",
                                "ps")

    device = run_extras.get("runner." + _envs["mode"] + ".device", "cpu")
    selected_gpus = run_extras.get(
        "runner." + _envs["mode"] + ".selected_gpus", "0")
    selected_gpus_num = len(selected_gpus.split(","))
    if device.upper() == "GPU":
        assert selected_gpus_num == 1, "Single Mode Only Support One GPU, Set Local Cluster Mode to use Multi-GPUS"

X
fix  
xjqbest 已提交
222
    single_envs = {}
C
Chengmo 已提交
223 224
    single_envs["selected_gpus"] = selected_gpus
    single_envs["FLAGS_selected_gpus"] = selected_gpus
X
fix  
xjqbest 已提交
225
    single_envs["train.trainer.trainer"] = trainer
C
Chengmo 已提交
226 227
    single_envs["train.trainer.executor_mode"] = executor_mode
    single_envs["fleet_mode"] = fleet_mode
X
fix  
xjqbest 已提交
228 229
    single_envs["train.trainer.threads"] = "2"
    single_envs["train.trainer.platform"] = envs.get_platform()
C
Chengmo 已提交
230 231
    single_envs["train.trainer.engine"] = "single"

X
fix  
xjqbest 已提交
232 233 234
    set_runtime_envs(single_envs, args.model)
    trainer = TrainerFactory.create(args.model)
    return trainer
C
chengmo 已提交
235

X
fix  
xjqbest 已提交
236

T
tangwei 已提交
237
def cluster_engine(args):
T
tangwei 已提交
238 239
    def update_workspace(cluster_envs):
        workspace = cluster_envs.get("engine_workspace", None)
T
tangwei 已提交
240

T
tangwei 已提交
241 242
        if not workspace:
            return
T
tangwei 已提交
243
        path = envs.path_adapter(workspace)
T
tangwei 已提交
244 245 246
        for name, value in cluster_envs.items():
            if isinstance(value, str):
                value = value.replace("{workspace}", path)
T
tangwei 已提交
247
                value = envs.windows_path_converter(value)
T
tangwei 已提交
248 249 250
                cluster_envs[name] = value

    def master():
T
tangwei 已提交
251
        role = "MASTER"
252
        from paddlerec.core.engine.cluster.cluster import ClusterEngine
X
test  
xjqbest 已提交
253
        _envs = envs.load_yaml(args.backend)
T
tangwei 已提交
254
        flattens = envs.flatten_environs(_envs, "_")
T
tangwei 已提交
255
        flattens["engine_role"] = role
T
tangwei 已提交
256
        flattens["engine_run_config"] = args.model
T
tangwei 已提交
257 258 259 260
        flattens["engine_temp_path"] = tempfile.mkdtemp()
        update_workspace(flattens)

        envs.set_runtime_environs(flattens)
C
Chengmo 已提交
261
        print(envs.pretty_print_envs(flattens, ("Submit Envs", "Value")))
T
tangwei 已提交
262 263 264 265 266

        launch = ClusterEngine(None, args.model)
        return launch

    def worker():
T
tangwei 已提交
267
        role = "WORKER"
C
Chengmo 已提交
268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288

        _envs = envs.load_yaml(args.model)
        run_extras = get_all_inters_from_yaml(args.model,
                                              ["train.", "runner."])
        trainer_class = run_extras.get(
            "runner." + _envs["mode"] + ".trainer_class", None)

        if trainer_class:
            trainer = trainer_class
        else:
            trainer = "GeneralTrainer"

        executor_mode = "train"

        distributed_strategy = run_extras.get(
            "runner." + _envs["mode"] + ".distribute_strategy", "async")
        selected_gpus = run_extras.get(
            "runner." + _envs["mode"] + ".selected_gpus", "0")
        fleet_mode = run_extras.get("runner." + _envs["mode"] + ".fleet_mode",
                                    "ps")

T
tangwei 已提交
289
        cluster_envs = {}
C
Chengmo 已提交
290 291
        cluster_envs["selected_gpus"] = selected_gpus
        cluster_envs["fleet_mode"] = fleet_mode
T
tangwei 已提交
292
        cluster_envs["train.trainer.trainer"] = trainer
C
Chengmo 已提交
293
        cluster_envs["train.trainer.executor_mode"] = executor_mode
T
tangwei 已提交
294
        cluster_envs["train.trainer.engine"] = "cluster"
C
Chengmo 已提交
295
        cluster_envs["train.trainer.strategy"] = distributed_strategy
T
tangwei 已提交
296 297
        cluster_envs["train.trainer.threads"] = envs.get_runtime_environ(
            "CPU_NUM")
T
tangwei 已提交
298
        cluster_envs["train.trainer.platform"] = envs.get_platform()
C
chengmo 已提交
299 300
        print("launch {} engine with cluster to with model: {}".format(
            trainer, args.model))
T
tangwei 已提交
301
        set_runtime_envs(cluster_envs, args.model)
T
tangwei 已提交
302

T
bug fix  
tangwei12 已提交
303 304
        trainer = TrainerFactory.create(args.model)
        return trainer
T
tangwei 已提交
305

T
tangwei 已提交
306 307 308
    role = os.getenv("PADDLE_PADDLEREC_ROLE", "MASTER")

    if role == "WORKER":
T
tangwei 已提交
309 310 311
        return worker()
    else:
        return master()
C
chengmo 已提交
312 313


T
tangwei 已提交
314
def cluster_mpi_engine(args):
T
tangwei 已提交
315 316
    print("launch cluster engine with cluster to run model: {}".format(
        args.model))
T
tangwei 已提交
317

T
fix bug  
tangwei 已提交
318
    cluster_envs = {}
T
tangwei 已提交
319
    cluster_envs["train.trainer.trainer"] = "CtrCodingTrainer"
T
tangwei 已提交
320
    cluster_envs["train.trainer.platform"] = envs.get_platform()
T
tangwei 已提交
321

T
tangwei 已提交
322
    set_runtime_envs(cluster_envs, args.model)
T
tangwei 已提交
323

T
tangwei 已提交
324 325 326 327 328
    trainer = TrainerFactory.create(args.model)
    return trainer


def local_cluster_engine(args):
329
    from paddlerec.core.engine.local_cluster import LocalClusterEngine
C
chengmo 已提交
330

C
Chengmo 已提交
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
    _envs = envs.load_yaml(args.model)
    run_extras = get_all_inters_from_yaml(args.model, ["train.", "runner."])
    trainer_class = run_extras.get("runner." + _envs["mode"] + ".runner_class",
                                   None)

    if trainer_class:
        trainer = trainer_class
    else:
        trainer = "GeneralTrainer"

    executor_mode = "train"
    distributed_strategy = run_extras.get(
        "runner." + _envs["mode"] + ".distribute_strategy", "async")

    worker_num = run_extras.get("runner." + _envs["mode"] + ".worker_num", 1)
    server_num = run_extras.get("runner." + _envs["mode"] + ".server_num", 1)
    selected_gpus = run_extras.get(
        "runner." + _envs["mode"] + ".selected_gpus", "0")

    fleet_mode = run_extras.get("runner." + _envs["mode"] + ".fleet_mode", "")
    if fleet_mode == "":
        device = run_extras.get("runner." + _envs["mode"] + ".device", "cpu")
        if len(selected_gpus.split(",")) > 1 and device.upper() == "GPU":
            fleet_mode = "COLLECTIVE"
        else:
            fleet_mode = "PS"

C
chengmo 已提交
358
    cluster_envs = {}
C
Chengmo 已提交
359 360 361
    cluster_envs["server_num"] = server_num
    cluster_envs["worker_num"] = worker_num
    cluster_envs["selected_gpus"] = selected_gpus
C
chengmo 已提交
362
    cluster_envs["start_port"] = envs.find_free_port()
C
Chengmo 已提交
363
    cluster_envs["fleet_mode"] = fleet_mode
C
chengmo 已提交
364
    cluster_envs["log_dir"] = "logs"
C
chengmo 已提交
365
    cluster_envs["train.trainer.trainer"] = trainer
C
Chengmo 已提交
366 367
    cluster_envs["train.trainer.executor_mode"] = executor_mode
    cluster_envs["train.trainer.strategy"] = distributed_strategy
C
chengmo 已提交
368 369 370 371 372
    cluster_envs["train.trainer.threads"] = "2"
    cluster_envs["train.trainer.engine"] = "local_cluster"
    cluster_envs["train.trainer.platform"] = envs.get_platform()

    cluster_envs["CPU_NUM"] = "2"
T
tangwei 已提交
373 374
    print("launch {} engine with cluster to run model: {}".format(trainer,
                                                                  args.model))
C
chengmo 已提交
375 376 377 378 379 380

    set_runtime_envs(cluster_envs, args.model)
    launch = LocalClusterEngine(cluster_envs, args.model)
    return launch


T
tangwei 已提交
381
def local_mpi_engine(args):
T
tangwei 已提交
382 383
    print("launch cluster engine with cluster to run model: {}".format(
        args.model))
384
    from paddlerec.core.engine.local_mpi import LocalMPIEngine
T
tangwei 已提交
385

T
tangwei 已提交
386 387
    print("use 1X1 MPI ClusterTraining at localhost to run model: {}".format(
        args.model))
T
tangwei 已提交
388

T
tangwei 已提交
389 390 391
    mpi = util.run_which("mpirun")
    if not mpi:
        raise RuntimeError("can not find mpirun, please check environment")
C
Chengmo 已提交
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407

    _envs = envs.load_yaml(args.model)
    run_extras = get_all_inters_from_yaml(args.model, ["train.", "runner."])
    trainer_class = run_extras.get("runner." + _envs["mode"] + ".runner_class",
                                   None)
    executor_mode = "train"
    distributed_strategy = run_extras.get(
        "runner." + _envs["mode"] + ".distribute_strategy", "async")
    fleet_mode = run_extras.get("runner." + _envs["mode"] + ".fleet_mode",
                                "ps")

    if trainer_class:
        trainer = trainer_class
    else:
        trainer = "GeneralTrainer"

T
fix bug  
tangwei 已提交
408 409
    cluster_envs = {}
    cluster_envs["mpirun"] = mpi
C
Chengmo 已提交
410
    cluster_envs["train.trainer.trainer"] = trainer
T
fix bug  
tangwei 已提交
411
    cluster_envs["log_dir"] = "logs"
T
tangwei 已提交
412
    cluster_envs["train.trainer.engine"] = "local_cluster"
C
Chengmo 已提交
413 414 415 416 417
    cluster_envs["train.trainer.executor_mode"] = executor_mode
    cluster_envs["fleet_mode"] = fleet_mode
    cluster_envs["train.trainer.strategy"] = distributed_strategy
    cluster_envs["train.trainer.threads"] = "2"
    cluster_envs["train.trainer.engine"] = "local_cluster"
T
tangwei 已提交
418
    cluster_envs["train.trainer.platform"] = envs.get_platform()
T
tangwei 已提交
419

T
tangwei 已提交
420
    set_runtime_envs(cluster_envs, args.model)
T
tangwei 已提交
421 422 423 424
    launch = LocalMPIEngine(cluster_envs, args.model)
    return launch


T
tangwei 已提交
425
def get_abs_model(model):
426
    if model.startswith("paddlerec."):
T
tangwei 已提交
427
        dir = envs.path_adapter(model)
T
tangwei 已提交
428
        path = os.path.join(dir, "config.yaml")
T
tangwei 已提交
429 430 431 432 433 434 435
    else:
        if not os.path.isfile(model):
            raise IOError("model config: {} invalid".format(model))
        path = model
    return path


T
tangwei 已提交
436
if __name__ == "__main__":
437
    parser = argparse.ArgumentParser(description='paddle-rec run')
T
tangwei 已提交
438
    parser.add_argument("-m", "--model", type=str)
T
tangwei 已提交
439
    parser.add_argument("-b", "--backend", type=str, default=None)
T
tangwei 已提交
440

T
tangwei 已提交
441 442 443
    abs_dir = os.path.dirname(os.path.abspath(__file__))
    envs.set_runtime_environs({"PACKAGE_BASE": abs_dir})

T
tangwei 已提交
444
    args = parser.parse_args()
C
fix  
chengmo 已提交
445
    model_name = args.model.split('.')[-1]
T
tangwei 已提交
446
    args.model = get_abs_model(args.model)
X
test  
xjqbest 已提交
447 448
    if not validation.yaml_validation(args.model):
        sys.exit(-1)
T
tangwei 已提交
449
    engine_registry()
C
chengmo 已提交
450
    which_engine = get_engine(args)
T
tangwei 已提交
451 452
    engine = which_engine(args)
    engine.run()