runner.py 25.9 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
F
frankwhzhang 已提交
21
import logging
C
Chengmo 已提交
22
import paddle.fluid as fluid
C
Chengmo 已提交
23

C
Chengmo 已提交
24
from paddlerec.core.utils import envs
M
update  
malin10 已提交
25
from paddlerec.core.metric import Metric
C
Chengmo 已提交
26

F
frankwhzhang 已提交
27 28 29
logging.basicConfig(
    format='%(asctime)s - %(levelname)s: %(message)s', level=logging.INFO)

C
Chengmo 已提交
30 31 32 33 34
__all__ = [
    "RunnerBase", "SingleRunner", "PSRunner", "CollectiveRunner", "PslibRunner"
]


C
Chengmo 已提交
35 36 37 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
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 已提交
71 72 73 74 75 76 77 78 79 80 81 82 83
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 已提交
84

C
Chengmo 已提交
85
        if envs.get_global_env(name + "type") == "DataLoader":
M
update  
malin10 已提交
86
            return self._executor_dataloader_train(model_dict, context)
C
Chengmo 已提交
87 88
        else:
            self._executor_dataset_train(model_dict, context)
M
update  
malin10 已提交
89
            return None
C
Chengmo 已提交
90 91 92 93 94

    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"]
95

C
Chengmo 已提交
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115
        fetch_vars = []
        fetch_alias = []
        fetch_period = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".print_interval", 20))
        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 已提交
116 117
                    print_period=fetch_period,
                    debug=envs.get_global_env("debug", False))
C
Chengmo 已提交
118 119 120 121 122 123 124 125 126 127 128
            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 已提交
129 130
                        print_period=fetch_period,
                        debug=envs.get_global_env("debug", False))
C
Chengmo 已提交
131 132 133 134

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

        fetch_period = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".print_interval", 20))
        if context["is_infer"]:
            metrics = model_class.get_infer_results()
        else:
            metrics = model_class.get_metrics()

        metrics_varnames = []
        metrics_format = []
F
frankwhzhang 已提交
147 148

        if context["is_infer"]:
F
frankwhzhang 已提交
149
            metrics_format.append("\t[Infer]\t{}: {{}}".format("batch"))
F
frankwhzhang 已提交
150
        else:
F
frankwhzhang 已提交
151
            metrics_format.append("\t[Train]\t{}: {{}}".format("batch"))
F
frankwhzhang 已提交
152 153 154

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

M
update  
malin10 已提交
155
        metrics_names = ["total_batch"]
F
frankwhzhang 已提交
156

C
Chengmo 已提交
157
        for name, var in metrics.items():
M
update  
malin10 已提交
158
            metrics_names.append(name)
C
Chengmo 已提交
159 160 161 162 163 164 165
            metrics_varnames.append(var.name)
            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 已提交
166
        begin_time = time.time()
C
Chengmo 已提交
167
        scope = context["model"][model_name]["scope"]
M
update  
malin10 已提交
168
        result = None
C
Chengmo 已提交
169 170 171
        with fluid.scope_guard(scope):
            try:
                while True:
C
Chengmo 已提交
172 173 174 175
                    metrics_tensors = context["exe"].run(
                        program=program,
                        fetch_list=metrics_varnames,
                        return_numpy=False)
C
Chengmo 已提交
176

F
frankwhzhang 已提交
177 178 179 180 181 182 183
                    metrics = [batch_id]
                    metrics_rets = [
                        as_numpy(metrics_tensor)
                        for metrics_tensor in metrics_tensors
                    ]
                    metrics.extend(metrics_rets)

C
Chengmo 已提交
184
                    if batch_id % fetch_period == 0 and batch_id != 0:
F
frankwhzhang 已提交
185 186
                        end_time = time.time()
                        seconds = end_time - begin_time
F
frankwhzhang 已提交
187 188
                        metrics_logging = metrics[:]
                        metrics_logging = metrics.insert(1, seconds)
F
frankwhzhang 已提交
189 190
                        begin_time = end_time

F
frankwhzhang 已提交
191
                        logging.info(metrics_format.format(*metrics))
C
Chengmo 已提交
192 193 194 195
                    batch_id += 1
            except fluid.core.EOFException:
                reader.reset()

M
update  
malin10 已提交
196
        if batch_id > 0:
M
update  
malin10 已提交
197 198
            result = dict(zip(metrics_names, metrics))
        return result
M
update  
malin10 已提交
199

200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217
    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 已提交
218 219 220 221 222 223 224 225 226 227 228 229 230 231
    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 已提交
232
                "Unsupported config. gradient_scale_strategy must be one of [0, 1, 2]."
C
Chengmo 已提交
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252
            )
        _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 已提交
253

C
Chengmo 已提交
254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293
        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

    def save(self, epoch_id, context, is_fleet=False):
        def need_save(epoch_id, epoch_interval, is_last=False):
294 295 296 297 298
            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 已提交
299 300 301 302 303
            if is_last:
                return True
            if epoch_id == -1:
                return False

304
            return (epoch_id + 1) % epoch_interval == 0
C
Chengmo 已提交
305 306

        def save_inference_model():
C
Chengmo 已提交
307
            # get global env
C
Chengmo 已提交
308 309 310 311 312 313 314 315 316 317
            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 已提交
318
                    len(feed_varnames) == 0 or len(fetch_varnames) == 0:
C
Chengmo 已提交
319
                return
C
Chengmo 已提交
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

            # 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 已提交
345 346 347 348 349 350
            dirname = envs.get_global_env(name + "save_inference_path", None)

            assert dirname is not None
            dirname = os.path.join(dirname, str(epoch_id))

            if is_fleet:
C
Chengmo 已提交
351 352 353 354 355 356 357
                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 已提交
358 359 360 361 362 363 364 365 366 367 368 369 370 371 372
            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))
            if is_fleet:
C
Chengmo 已提交
373 374
                if context["fleet"].worker_index() == 0:
                    context["fleet"].save_persistables(context["exe"], dirname)
C
Chengmo 已提交
375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394
            else:
                fluid.io.save_persistables(context["exe"], dirname)

        save_persistables()
        save_inference_model()


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 已提交
395
            for model_dict in context["phases"]:
M
update  
malin10 已提交
396
                model_class = context["model"][model_dict["name"]]["model"]
M
bug fix  
malin10 已提交
397
                metrics = model_class._metrics
M
update  
malin10 已提交
398

C
Chengmo 已提交
399
                begin_time = time.time()
M
update  
malin10 已提交
400
                result = self._run(context, model_dict)
C
Chengmo 已提交
401 402
                end_time = time.time()
                seconds = end_time - begin_time
M
update  
malin10 已提交
403
                message = "epoch {} done, use time: {}".format(epoch, seconds)
M
update  
malin10 已提交
404 405 406
                metrics_result = []
                for key in metrics:
                    if isinstance(metrics[key], Metric):
M
bug fix  
malin10 已提交
407
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
408 409
                            None,
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
410
                        metrics_result.append(_str)
M
update  
malin10 已提交
411 412
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
413
                        metrics_result.append(_str)
M
update  
malin10 已提交
414 415
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
416
                print(message)
M
update  
malin10 已提交
417

C
Chengmo 已提交
418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437
                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)
        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 已提交
438
        model_dict = context["env"]["phase"][0]
M
update  
malin10 已提交
439 440
        model_class = context["model"][model_dict["name"]]["model"]
        metrics = model_class._metrics
C
Chengmo 已提交
441 442
        for epoch in range(epochs):
            begin_time = time.time()
M
update  
malin10 已提交
443
            result = self._run(context, model_dict)
C
Chengmo 已提交
444 445
            end_time = time.time()
            seconds = end_time - begin_time
M
update  
malin10 已提交
446
            message = "epoch {} done, use time: {}".format(epoch, seconds)
M
update  
malin10 已提交
447 448 449 450 451 452 453 454

            # 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 已提交
455
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
456 457
                            context["fleet"],
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
458
                        metrics_result.append(_str)
M
update  
malin10 已提交
459 460
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
461
                        metrics_result.append(_str)
M
update  
malin10 已提交
462 463
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
464
            print(message)
C
Chengmo 已提交
465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484
            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):
                    self.save(epoch, context, True)
        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 已提交
485
        model_dict = context["env"]["phase"][0]
C
Chengmo 已提交
486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509
        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"


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

    def run(self, context):
        context["fleet"].init_worker()
T
tangwei 已提交
510
        model_dict = context["env"]["phase"][0]
C
Chengmo 已提交
511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530
        epochs = int(
            envs.get_global_env("runner." + context["runner_name"] +
                                ".epochs"))
        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))
        """
        # 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 已提交
531
                for dataset in envs.get_global_env("dataset"):
C
Chengmo 已提交
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
                    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"
561 562 563 564 565 566 567 568 569 570 571 572


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 已提交
573 574
                model_class = context["model"][model_dict["name"]]["model"]
                metrics = model_class._infer_results
575 576 577
                self._load(context, model_dict,
                           self.epoch_model_path_list[index])
                begin_time = time.time()
M
update  
malin10 已提交
578
                result = self._run(context, model_dict)
579 580
                end_time = time.time()
                seconds = end_time - begin_time
M
update  
malin10 已提交
581 582
                message = "Infer {} of epoch {} done, use time: {}".format(
                    model_dict["name"], epoch_name, seconds)
M
update  
malin10 已提交
583 584 585
                metrics_result = []
                for key in metrics:
                    if isinstance(metrics[key], Metric):
M
bug fix  
malin10 已提交
586
                        _str = metrics[key].calc_global_metrics(
M
update  
malin10 已提交
587 588
                            None,
                            context["model"][model_dict["name"]]["scope"])
M
malin10 已提交
589
                        metrics_result.append(_str)
M
update  
malin10 已提交
590 591
                    elif result is not None:
                        _str = "{}={}".format(key, result[key])
M
malin10 已提交
592
                        metrics_result.append(_str)
M
update  
malin10 已提交
593 594
                if len(metrics_result) > 0:
                    message += ", global metrics: " + ", ".join(metrics_result)
M
update  
malin10 已提交
595 596
                print(message)

597 598 599 600 601 602 603 604 605 606 607 608 609 610
        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 已提交
611 612 613 614
            clear_metrics = context["model"][model_dict["name"]][
                "model"].get_clear_metrics()
            for var in clear_metrics:
                var.clear()
615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630

    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)