runner.py 31.4 KB
Newer Older
C
Chengmo 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
# 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.

from __future__ import print_function

import os
import time
C
Chengmo 已提交
19
import warnings
C
Chengmo 已提交
20
import numpy as np
21
import random
C
Chengmo 已提交
22
import json
F
frankwhzhang 已提交
23
import logging
C
Chengmo 已提交
24
import paddle.fluid as fluid
C
Chengmo 已提交
25

C
Chengmo 已提交
26
from paddlerec.core.utils import envs
27
from paddlerec.core.utils.util import shuffle_files
M
update  
malin10 已提交
28
from paddlerec.core.metric import Metric
C
Chengmo 已提交
29

F
frankwhzhang 已提交
30 31 32
logging.basicConfig(
    format='%(asctime)s - %(levelname)s: %(message)s', level=logging.INFO)

C
Chengmo 已提交
33 34 35 36 37
__all__ = [
    "RunnerBase", "SingleRunner", "PSRunner", "CollectiveRunner", "PslibRunner"
]


C
Chengmo 已提交
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73
def as_numpy(tensor):
    """
    Convert a Tensor to a numpy.ndarray, its only support Tensor without LoD information.
    For higher dimensional sequence data, please use LoDTensor directly.

    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          import numpy

          new_scope = fluid.Scope()
          with fluid.scope_guard(new_scope):
              fluid.global_scope().var("data").get_tensor().set(numpy.ones((2, 2)), fluid.CPUPlace())
          tensor = new_scope.find_var("data").get_tensor()
          fluid.executor.as_numpy(tensor) # or numpy.array(new_scope.find_var("data").get_tensor())

    Args:
       tensor(Variable): a instance of Tensor

    Returns:
        numpy.ndarray
    """
    if isinstance(tensor, fluid.core.LoDTensorArray):
        return [as_numpy(t) for t in tensor]
    if isinstance(tensor, list):
        return [as_numpy(t) for t in tensor]
    assert isinstance(tensor, fluid.core.LoDTensor)
    lod = tensor.lod()
    # (todo) need print lod or return it for user
    if tensor._is_initialized():
        return np.array(tensor)
    else:
        return None


C
Chengmo 已提交
74 75 76 77 78 79 80 81 82 83 84 85 86
class RunnerBase(object):
    """R
    """

    def __init__(self, context):
        pass

    def exuctor(self, context):
        pass

    def _run(self, context, model_dict):
        reader_name = model_dict["dataset_name"]
        name = "dataset." + reader_name + "."
T
tangwei 已提交
87

C
Chengmo 已提交
88
        if envs.get_global_env(name + "type") == "DataLoader":
M
update  
malin10 已提交
89
            return self._executor_dataloader_train(model_dict, context)
C
Chengmo 已提交
90 91
        else:
            self._executor_dataset_train(model_dict, context)
M
update  
malin10 已提交
92
            return None
C
Chengmo 已提交
93 94 95 96 97 98 99 100 101 102

    def _executor_dataset_train(self, model_dict, context):
        reader_name = model_dict["dataset_name"]
        model_name = model_dict["name"]
        model_class = context["model"][model_dict["name"]]["model"]
        fetch_vars = []
        fetch_alias = []
        fetch_period = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".print_interval", 20))
L
liuyuhui 已提交
103

C
Chengmo 已提交
104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
        scope = context["model"][model_name]["scope"]
        program = context["model"][model_name]["main_program"]
        reader = context["dataset"][reader_name]

        with fluid.scope_guard(scope):
            if context["is_infer"]:
                metrics = model_class.get_infer_results()
                if metrics:
                    fetch_vars = metrics.values()
                    fetch_alias = metrics.keys()
                context["exe"].infer_from_dataset(
                    program=program,
                    dataset=reader,
                    fetch_list=fetch_vars,
                    fetch_info=fetch_alias,
X
xjqbest 已提交
119 120
                    print_period=fetch_period,
                    debug=envs.get_global_env("debug", False))
C
Chengmo 已提交
121 122 123 124 125 126 127 128 129 130 131
            else:
                metrics = model_class.get_metrics()
                if metrics:
                    fetch_vars = metrics.values()
                    fetch_alias = metrics.keys()
                with fluid.scope_guard(scope):
                    context["exe"].train_from_dataset(
                        program=program,
                        dataset=reader,
                        fetch_list=fetch_vars,
                        fetch_info=fetch_alias,
X
xjqbest 已提交
132 133
                        print_period=fetch_period,
                        debug=envs.get_global_env("debug", False))
C
Chengmo 已提交
134 135 136 137

    def _executor_dataloader_train(self, model_dict, context):
        model_name = model_dict["name"]
        model_class = context["model"][model_dict["name"]]["model"]
138
        program = self._get_dataloader_program(model_dict, context)
C
Chengmo 已提交
139 140 141 142

        fetch_period = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".print_interval", 20))
L
liuyuhui 已提交
143 144 145
        save_step_interval = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".save_step_interval", -1))
C
Chengmo 已提交
146 147 148 149 150 151 152
        if context["is_infer"]:
            metrics = model_class.get_infer_results()
        else:
            metrics = model_class.get_metrics()

        metrics_varnames = []
        metrics_format = []
F
frankwhzhang 已提交
153 154

        if context["is_infer"]:
C
Chengmo 已提交
155
            metrics_format.append("\t[Infer] {}: {{}}".format("batch"))
F
frankwhzhang 已提交
156
        else:
C
Chengmo 已提交
157 158 159 160 161
            metrics_format.append("\t[Train]")
            if "current_epoch" in context:
                metrics_format.append(" epoch: {}".format(context[
                    "current_epoch"]))
            metrics_format.append(" {}: {{}}".format("batch"))
F
frankwhzhang 已提交
162 163 164

        metrics_format.append("{}: {{:.2f}}s".format("time_each_interval"))

M
update  
malin10 已提交
165
        metrics_names = ["total_batch"]
C
Chengmo 已提交
166
        metrics_indexes = dict()
C
Chengmo 已提交
167
        for name, var in metrics.items():
M
update  
malin10 已提交
168
            metrics_names.append(name)
C
Chengmo 已提交
169
            metrics_varnames.append(var.name)
C
Chengmo 已提交
170
            metrics_indexes[var.name] = len(metrics_varnames) - 1
C
Chengmo 已提交
171 172 173 174 175 176
            metrics_format.append("{}: {{}}".format(name))
        metrics_format = ", ".join(metrics_format)

        reader = context["model"][model_dict["name"]]["model"]._data_loader
        reader.start()
        batch_id = 0
F
frankwhzhang 已提交
177
        begin_time = time.time()
C
Chengmo 已提交
178
        scope = context["model"][model_name]["scope"]
C
Chengmo 已提交
179
        runner_results = []
M
update  
malin10 已提交
180
        result = None
C
Chengmo 已提交
181 182 183
        with fluid.scope_guard(scope):
            try:
                while True:
C
Chengmo 已提交
184 185 186 187
                    metrics_tensors = context["exe"].run(
                        program=program,
                        fetch_list=metrics_varnames,
                        return_numpy=False)
C
Chengmo 已提交
188

F
frankwhzhang 已提交
189 190 191 192 193 194 195
                    metrics = [batch_id]
                    metrics_rets = [
                        as_numpy(metrics_tensor)
                        for metrics_tensor in metrics_tensors
                    ]
                    metrics.extend(metrics_rets)

C
Chengmo 已提交
196 197 198 199 200 201
                    batch_runner_result = {}
                    for k, v in metrics_indexes.items():
                        batch_runner_result[k] = np.array(metrics_rets[
                            v]).tolist()
                    runner_results.append(batch_runner_result)

C
Chengmo 已提交
202
                    if batch_id % fetch_period == 0 and batch_id != 0:
F
frankwhzhang 已提交
203 204
                        end_time = time.time()
                        seconds = end_time - begin_time
F
frankwhzhang 已提交
205
                        metrics_logging = metrics[:]
Z
zhang wenhui 已提交
206
                        metrics_logging.insert(1, seconds)
F
frankwhzhang 已提交
207
                        begin_time = end_time
Z
zhang wenhui 已提交
208
                        logging.info(metrics_format.format(*metrics_logging))
L
liuyuhui 已提交
209 210 211

                    if save_step_interval >= 1 and batch_id % save_step_interval == 0 and context[
                            "is_infer"] == False:
L
fix bug  
liuyuhui 已提交
212 213 214 215 216 217 218
                        if context["is_fleet"]:
                            if context["fleet_mode"].upper() == "PS":
                                train_prog = context["model"][model_dict[
                                    "name"]]["main_program"]
                            else:
                                train_prog = context["model"][model_dict[
                                    "name"]]["default_main_program"]
L
liuyuhui 已提交
219
                        else:
L
liuyuhui 已提交
220 221 222 223 224 225 226 227 228 229
                            train_prog = context["model"][model_dict["name"]][
                                "default_main_program"]
                        startup_prog = context["model"][model_dict["name"]][
                            "startup_program"]
                        with fluid.program_guard(train_prog, startup_prog):
                            self.save(
                                context,
                                is_fleet=context["is_fleet"],
                                epoch_id=None,
                                batch_id=batch_id)
Z
zhang wenhui 已提交
230

C
Chengmo 已提交
231 232 233 234
                    batch_id += 1
            except fluid.core.EOFException:
                reader.reset()

C
Chengmo 已提交
235 236 237 238 239 240 241 242 243 244 245 246
        runner_result_save_path = envs.get_global_env(
            "runner." + context["runner_name"] + ".runner_result_dump_path",
            None)
        if runner_result_save_path:
            if "current_epoch" in context:
                runner_result_save_path = runner_result_save_path + "_epoch_{}".format(
                    context["current_epoch"])
            logging.info("Dump runner result in {}".format(
                runner_result_save_path))
            with open(runner_result_save_path, 'w+') as fout:
                json.dump(runner_results, fout)

M
update  
malin10 已提交
247
        if batch_id > 0:
M
update  
malin10 已提交
248 249
            result = dict(zip(metrics_names, metrics))
        return result
M
update  
malin10 已提交
250

251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
    def _get_dataloader_program(self, model_dict, context):
        model_name = model_dict["name"]
        if context["model"][model_name]["compiled_program"] == None:
            if context["is_infer"]:
                program = context["model"][model_name]["main_program"]
            elif context["is_fleet"]:
                if context["fleet_mode"].upper() == "PS":
                    program = self._get_ps_program(model_dict, context)
                elif context["fleet_mode"].upper() == "COLLECTIVE":
                    program = context["model"][model_name]["main_program"]
            elif not context["is_fleet"]:
                if context["device"].upper() == "CPU":
                    program = self._get_single_cpu_program(model_dict, context)
                elif context["device"].upper() == "GPU":
                    program = self._get_single_gpu_program(model_dict, context)
            context["model"][model_name]["compiled_program"] = program
        return context["model"][model_name]["compiled_program"]

C
Chengmo 已提交
269 270 271 272 273 274 275 276 277 278 279 280 281 282
    def _get_strategy(self, model_dict, context):
        _build_strategy = fluid.BuildStrategy()
        _exe_strategy = fluid.ExecutionStrategy()

        # 0: kCoeffNumDevice; 1: One; 2: Customized
        _gradient_scale_strategy = model_dict.get("gradient_scale_strategy", 0)
        if _gradient_scale_strategy == 0:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.CoeffNumDevice
        elif _gradient_scale_strategy == 1:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.One
        elif _gradient_scale_strategy == 2:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.Customized
        else:
            raise ValueError(
T
tangwei 已提交
283
                "Unsupported config. gradient_scale_strategy must be one of [0, 1, 2]."
C
Chengmo 已提交
284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303
            )
        _build_strategy.gradient_scale_strategy = gradient_scale_strategy

        if "thread_num" in model_dict and model_dict["thread_num"] > 1:
            _build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
            _exe_strategy.num_threads = model_dict["thread_num"]
            os.environ['CPU_NUM'] = str(_exe_strategy.num_threads)

        return _exe_strategy, _build_strategy

    def _get_single_gpu_program(self, model_dict, context):
        model_name = model_dict["name"]
        return context["model"][model_name]["main_program"].clone()

    def _get_single_cpu_program(self, model_dict, context):
        model_name = model_dict["name"]
        model_class = context["model"][model_dict["name"]]["model"]
        program = context["model"][model_name]["main_program"].clone()
        _exe_strategy, _build_strategy = self._get_strategy(model_dict,
                                                            context)
M
update  
malin10 已提交
304

C
Chengmo 已提交
305 306 307 308 309 310 311 312 313 314 315 316 317 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
        program = fluid.compiler.CompiledProgram(program).with_data_parallel(
            loss_name=model_class.get_avg_cost().name,
            build_strategy=_build_strategy,
            exec_strategy=_exe_strategy)
        return program

    def _get_ps_program(self, model_dict, context):
        model_name = model_dict["name"]
        model_class = context["model"][model_dict["name"]]["model"]
        program = context["model"][model_name]["main_program"].clone()

        _build_strategy = context["strategy"].get_build_strategy()
        _exe_strategy = context["strategy"].get_execute_strategy()

        if "thread_num" in model_dict and model_dict["thread_num"] > 1:
            _build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
            _exe_strategy.num_threads = model_dict["thread_num"]
            os.environ['CPU_NUM'] = str(_exe_strategy.num_threads)

        _gradient_scale_strategy = model_dict.get("gradient_scale_strategy", 0)
        if _gradient_scale_strategy == 0:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.CoeffNumDevice
        elif _gradient_scale_strategy == 1:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.One
        elif _gradient_scale_strategy == 2:
            gradient_scale_strategy = fluid.BuildStrategy.GradientScaleStrategy.Customized
        else:
            raise ValueError(
                "Unsurpported config. gradient_scale_strategy must be one of [0, 1, 2]."
            )
        _build_strategy.gradient_scale_strategy = gradient_scale_strategy

        program = fluid.compiler.CompiledProgram(program).with_data_parallel(
            loss_name=model_class.get_avg_cost().name,
            build_strategy=_build_strategy,
            exec_strategy=_exe_strategy)
        return program

L
liuyuhui 已提交
343
    def save(self, context, is_fleet=False, epoch_id=None, batch_id=None):
C
Chengmo 已提交
344
        def need_save(epoch_id, epoch_interval, is_last=False):
345 346 347 348 349
            name = "runner." + context["runner_name"] + "."
            total_epoch = int(envs.get_global_env(name + "epochs", 1))
            if epoch_id + 1 == total_epoch:
                is_last = True

C
Chengmo 已提交
350 351 352 353 354
            if is_last:
                return True
            if epoch_id == -1:
                return False

355
            return (epoch_id + 1) % epoch_interval == 0
C
Chengmo 已提交
356 357

        def save_inference_model():
C
Chengmo 已提交
358
            # get global env
C
Chengmo 已提交
359 360 361 362 363 364 365 366 367 368
            name = "runner." + context["runner_name"] + "."
            save_interval = int(
                envs.get_global_env(name + "save_inference_interval", -1))
            if not need_save(epoch_id, save_interval, False):
                return
            feed_varnames = envs.get_global_env(
                name + "save_inference_feed_varnames", [])
            fetch_varnames = envs.get_global_env(
                name + "save_inference_fetch_varnames", [])
            if feed_varnames is None or fetch_varnames is None or feed_varnames == "" or fetch_varnames == "" or \
C
Chengmo 已提交
369
                    len(feed_varnames) == 0 or len(fetch_varnames) == 0:
C
Chengmo 已提交
370
                return
C
Chengmo 已提交
371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395

            # check feed var exist
            for var_name in feed_varnames:
                if var_name not in fluid.default_main_program().global_block(
                ).vars:
                    raise ValueError(
                        "Feed variable: {} not in default_main_program, global block has follow vars: {}".
                        format(var_name,
                               fluid.default_main_program().global_block()
                               .vars.keys()))

            # check fetch var exist
            fetch_vars = []
            for var_name in fetch_varnames:
                if var_name not in fluid.default_main_program().global_block(
                ).vars:
                    raise ValueError(
                        "Fetch variable: {} not in default_main_program, global block has follow vars: {}".
                        format(var_name,
                               fluid.default_main_program().global_block()
                               .vars.keys()))
                else:
                    fetch_vars.append(fluid.default_main_program()
                                      .global_block().vars[var_name])

C
Chengmo 已提交
396 397 398 399
            dirname = envs.get_global_env(name + "save_inference_path", None)

            assert dirname is not None
            dirname = os.path.join(dirname, str(epoch_id))
L
liuyuhui 已提交
400 401
            logging.info("\tsave epoch_id:%d model into: \"%s\"" %
                         (epoch_id, dirname))
C
Chengmo 已提交
402
            if is_fleet:
C
Chengmo 已提交
403 404 405 406 407 408 409
                warnings.warn(
                    "Save inference model in cluster training is not recommended! Using save checkpoint instead.",
                    category=UserWarning,
                    stacklevel=2)
                if context["fleet"].worker_index() == 0:
                    context["fleet"].save_inference_model(
                        context["exe"], dirname, feed_varnames, fetch_vars)
C
Chengmo 已提交
410 411 412 413 414 415 416 417 418 419 420 421 422 423
            else:
                fluid.io.save_inference_model(dirname, feed_varnames,
                                              fetch_vars, context["exe"])

        def save_persistables():
            name = "runner." + context["runner_name"] + "."
            save_interval = int(
                envs.get_global_env(name + "save_checkpoint_interval", -1))
            if not need_save(epoch_id, save_interval, False):
                return
            dirname = envs.get_global_env(name + "save_checkpoint_path", None)
            if dirname is None or dirname == "":
                return
            dirname = os.path.join(dirname, str(epoch_id))
L
liuyuhui 已提交
424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441
            logging.info("\tsave epoch_id:%d model into: \"%s\"" %
                         (epoch_id, dirname))
            if is_fleet:
                if context["fleet"].worker_index() == 0:
                    context["fleet"].save_persistables(context["exe"], dirname)
            else:
                fluid.io.save_persistables(context["exe"], dirname)

        def save_checkpoint_step():
            name = "runner." + context["runner_name"] + "."
            save_interval = int(
                envs.get_global_env(name + "save_step_interval", -1))
            dirname = envs.get_global_env(name + "save_step_path", None)
            if dirname is None or dirname == "":
                return
            dirname = os.path.join(dirname, str(batch_id))
            logging.info("\tsave batch_id:%d model into: \"%s\"" %
                         (batch_id, dirname))
C
Chengmo 已提交
442
            if is_fleet:
C
Chengmo 已提交
443 444
                if context["fleet"].worker_index() == 0:
                    context["fleet"].save_persistables(context["exe"], dirname)
C
Chengmo 已提交
445 446 447
            else:
                fluid.io.save_persistables(context["exe"], dirname)

L
liuyuhui 已提交
448 449 450 451 452
        if isinstance(epoch_id, int):
            save_persistables()
            save_inference_model()
        if isinstance(batch_id, int):
            save_checkpoint_step()
C
Chengmo 已提交
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467


class SingleRunner(RunnerBase):
    """R
    """

    def __init__(self, context):
        print("Running SingleRunner.")
        pass

    def run(self, context):
        epochs = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".epochs"))
        for epoch in range(epochs):
T
tangwei 已提交
468
            for model_dict in context["phases"]:
M
update  
malin10 已提交
469
                model_class = context["model"][model_dict["name"]]["model"]
M
bug fix  
malin10 已提交
470
                metrics = model_class._metrics
471 472 473 474 475 476
                if "shuffle_filelist" in model_dict:
                    need_shuffle_files = model_dict.get("shuffle_filelist",
                                                        None)
                    filelist = context["file_list"]
                    context["file_list"] = shuffle_files(need_shuffle_files,
                                                         filelist)
C
Chengmo 已提交
477
                context["current_epoch"] = epoch
C
Chengmo 已提交
478
                begin_time = time.time()
M
update  
malin10 已提交
479
                result = self._run(context, model_dict)
C
Chengmo 已提交
480 481
                end_time = time.time()
                seconds = end_time - begin_time
M
update  
malin10 已提交
482
                message = "epoch {} done, use time: {}".format(epoch, seconds)
M
update  
malin10 已提交
483 484 485
                metrics_result = []
                for key in metrics:
                    if isinstance(metrics[key], Metric):
M
bug fix  
malin10 已提交
486
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
487 488
                            None,
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
489
                        metrics_result.append(_str)
M
update  
malin10 已提交
490 491
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
492
                        metrics_result.append(_str)
M
update  
malin10 已提交
493 494
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
495
                print(message)
M
update  
malin10 已提交
496

C
Chengmo 已提交
497 498 499 500 501 502 503
                with fluid.scope_guard(context["model"][model_dict["name"]][
                        "scope"]):
                    train_prog = context["model"][model_dict["name"]][
                        "default_main_program"]
                    startup_prog = context["model"][model_dict["name"]][
                        "startup_program"]
                    with fluid.program_guard(train_prog, startup_prog):
L
liuyuhui 已提交
504
                        self.save(context=context, epoch_id=epoch)
C
Chengmo 已提交
505 506 507 508 509 510 511 512 513 514 515 516
        context["status"] = "terminal_pass"


class PSRunner(RunnerBase):
    def __init__(self, context):
        print("Running PSRunner.")
        pass

    def run(self, context):
        epochs = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".epochs"))
T
tangwei 已提交
517
        model_dict = context["env"]["phase"][0]
M
update  
malin10 已提交
518 519
        model_class = context["model"][model_dict["name"]]["model"]
        metrics = model_class._metrics
C
Chengmo 已提交
520
        for epoch in range(epochs):
521 522 523 524 525
            if "shuffle_filelist" in model_dict:
                need_shuffle_files = model_dict.get("shuffle_filelist", None)
                filelist = context["file_list"]
                context["file_list"] = shuffle_files(need_shuffle_files,
                                                     filelist)
C
Chengmo 已提交
526
            context["current_epoch"] = epoch
C
Chengmo 已提交
527
            begin_time = time.time()
M
update  
malin10 已提交
528
            result = self._run(context, model_dict)
C
Chengmo 已提交
529 530
            end_time = time.time()
            seconds = end_time - begin_time
M
update  
malin10 已提交
531
            message = "epoch {} done, use time: {}".format(epoch, seconds)
M
update  
malin10 已提交
532 533 534 535 536 537 538 539

            # TODO, wait for PaddleCloudRoleMaker supports gloo
            from paddle.fluid.incubate.fleet.base.role_maker import GeneralRoleMaker
            if context["fleet"] is not None and isinstance(context["fleet"],
                                                           GeneralRoleMaker):
                metrics_result = []
                for key in metrics:
                    if isinstance(metrics[key], Metric):
M
bug fix  
malin10 已提交
540
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
541 542
                            context["fleet"],
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
543
                        metrics_result.append(_str)
M
update  
malin10 已提交
544 545
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
546
                        metrics_result.append(_str)
M
update  
malin10 已提交
547 548
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
549
            print(message)
C
Chengmo 已提交
550 551 552 553 554 555 556
            with fluid.scope_guard(context["model"][model_dict["name"]][
                    "scope"]):
                train_prog = context["model"][model_dict["name"]][
                    "main_program"]
                startup_prog = context["model"][model_dict["name"]][
                    "startup_program"]
                with fluid.program_guard(train_prog, startup_prog):
L
liuyuhui 已提交
557
                    self.save(context=context, is_fleet=True, epoch_id=epoch)
C
Chengmo 已提交
558 559 560 561 562 563 564 565 566 567 568 569
        context["status"] = "terminal_pass"


class CollectiveRunner(RunnerBase):
    def __init__(self, context):
        print("Running CollectiveRunner.")
        pass

    def run(self, context):
        epochs = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".epochs"))
T
tangwei 已提交
570
        model_dict = context["env"]["phase"][0]
C
Chengmo 已提交
571
        for epoch in range(epochs):
572 573 574 575 576
            if "shuffle_filelist" in model_dict:
                need_shuffle_files = model_dict.get("shuffle_filelist", None)
                filelist = context["file_list"]
                context["file_list"] = shuffle_files(need_shuffle_files,
                                                     filelist)
C
Chengmo 已提交
577
            context["current_epoch"] = epoch
C
Chengmo 已提交
578 579 580 581 582 583 584 585 586 587 588 589
            begin_time = time.time()
            self._run(context, model_dict)
            end_time = time.time()
            seconds = end_time - begin_time
            print("epoch {} done, use time: {}".format(epoch, seconds))
            with fluid.scope_guard(context["model"][model_dict["name"]][
                    "scope"]):
                train_prog = context["model"][model_dict["name"]][
                    "default_main_program"]
                startup_prog = context["model"][model_dict["name"]][
                    "startup_program"]
                with fluid.program_guard(train_prog, startup_prog):
L
liuyuhui 已提交
590
                    self.save(context=context, is_fleet=True, epoch_id=epoch)
C
Chengmo 已提交
591 592 593 594 595 596 597 598 599 600
        context["status"] = "terminal_pass"


class PslibRunner(RunnerBase):
    def __init__(self, context):
        print("Running PSRunner.")
        pass

    def run(self, context):
        context["fleet"].init_worker()
T
tangwei 已提交
601
        model_dict = context["env"]["phase"][0]
C
Chengmo 已提交
602 603 604 605
        epochs = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".epochs"))
        for epoch in range(epochs):
606 607 608 609 610
            if "shuffle_filelist" in model_dict:
                need_shuffle_files = model_dict.get("shuffle_filelist", None)
                filelist = context["file_list"]
                context["file_list"] = shuffle_files(need_shuffle_files,
                                                     filelist)
C
Chengmo 已提交
611
            context["current_epoch"] = epoch
C
Chengmo 已提交
612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627
            begin_time = time.time()
            self._run(context, model_dict)
            end_time = time.time()
            seconds = end_time - begin_time
            print("epoch {} done, use time: {}".format(epoch, seconds))
        """
        # online Training Can do more, As shown below:

        begin_day = datetime.datetime.strptime("begin_day_d", '%Y%m%d')
        days = int(
            envs.get_global_env("runner." + context["runner_name"] + ".days"))
        for day in range(days):
            for hour in range(24):
                day = begin_day + datetime.timedelta(days=day, hours=hour)
                day_s = day.strftime('%Y%m%d/%H')

T
tangwei 已提交
628
                for dataset in envs.get_global_env("dataset"):
C
Chengmo 已提交
629 630 631 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
                    if dataset["type"] != "DataLoader":
                        name = dataset["name"]
                        train_data_path = envs.get_global_env(name +
                                                              "data_path")
                        train_data_path = os.path.join(train_data_path, day_s)

                        file_list = [
                            os.path.join(train_data_path, x)
                            for x in os.listdir(train_data_path)
                        ]
                        context["dataset"][name].set_filelist(file_list)

                for epoch in range(epochs):
                    begin_time = time.time()
                    self._run(context, model_dict)
                    end_time = time.time()
                    seconds = end_time - begin_time
                    print("epoch {} done, use time: {}".format(epoch, seconds))
                    with fluid.scope_guard(context["model"][model_dict["name"]]
                                           ["scope"]):
                        train_prog = context["model"][model_dict["name"]][
                            "default_main_program"]
                        startup_prog = context["model"][model_dict["name"]][
                            "startup_program"]
                        with fluid.program_guard(train_prog, startup_prog):
                            self.save(epoch, context, True)

        """
        context["status"] = "terminal_pass"
658 659 660 661 662 663 664 665 666 667 668 669


class SingleInferRunner(RunnerBase):
    def __init__(self, context):
        print("Running SingleInferRunner.")
        pass

    def run(self, context):
        self._dir_check(context)

        for index, epoch_name in enumerate(self.epoch_model_name_list):
            for model_dict in context["phases"]:
M
update  
malin10 已提交
670 671
                model_class = context["model"][model_dict["name"]]["model"]
                metrics = model_class._infer_results
672 673
                self._load(context, model_dict,
                           self.epoch_model_path_list[index])
674 675 676 677 678 679
                if "shuffle_filelist" in model_dict:
                    need_shuffle_files = model_dict.get("shuffle_filelist",
                                                        None)
                    filelist = context["file_list"]
                    context["file_list"] = shuffle_files(need_shuffle_files,
                                                         filelist)
680
                begin_time = time.time()
M
update  
malin10 已提交
681
                result = self._run(context, model_dict)
682 683
                end_time = time.time()
                seconds = end_time - begin_time
M
update  
malin10 已提交
684 685
                message = "Infer {} of epoch {} done, use time: {}".format(
                    model_dict["name"], epoch_name, seconds)
M
update  
malin10 已提交
686 687 688
                metrics_result = []
                for key in metrics:
                    if isinstance(metrics[key], Metric):
M
bug fix  
malin10 已提交
689
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
690 691
                            None,
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
692
                        metrics_result.append(_str)
M
update  
malin10 已提交
693 694
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
695
                        metrics_result.append(_str)
M
update  
malin10 已提交
696 697
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
698 699
                print(message)

700 701 702 703 704 705 706 707 708 709 710 711 712 713
        context["status"] = "terminal_pass"

    def _load(self, context, model_dict, model_path):
        if model_path is None or model_path == "":
            return
        print("load persistables from", model_path)

        with fluid.scope_guard(context["model"][model_dict["name"]]["scope"]):
            train_prog = context["model"][model_dict["name"]]["main_program"]
            startup_prog = context["model"][model_dict["name"]][
                "startup_program"]
            with fluid.program_guard(train_prog, startup_prog):
                fluid.io.load_persistables(
                    context["exe"], model_path, main_program=train_prog)
M
update  
malin10 已提交
714 715 716 717
            clear_metrics = context["model"][model_dict["name"]][
                "model"].get_clear_metrics()
            for var in clear_metrics:
                var.clear()
718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733

    def _dir_check(self, context):
        dirname = envs.get_global_env(
            "runner." + context["runner_name"] + ".init_model_path", None)
        self.epoch_model_path_list = []
        self.epoch_model_name_list = []

        for file in os.listdir(dirname):
            file_path = os.path.join(dirname, file)
            if os.path.isdir(file_path):
                self.epoch_model_path_list.append(file_path)
                self.epoch_model_name_list.append(file)

        if len(self.epoch_model_path_list) == 0:
            self.epoch_model_path_list.append(dirname)
            self.epoch_model_name_list.append(dirname)