executor.py 32.0 KB
Newer Older
1
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
D
dzhwinter 已提交
2
#
D
dzhwinter 已提交
3 4 5
# 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
D
dzhwinter 已提交
6
#
D
dzhwinter 已提交
7
#     http://www.apache.org/licenses/LICENSE-2.0
D
dzhwinter 已提交
8
#
D
dzhwinter 已提交
9 10 11 12 13 14
# 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.

15 16
from __future__ import print_function

17 18
import os
import multiprocessing
D
dzhwinter 已提交
19
import numpy as np
S
rename  
sneaxiy 已提交
20
from .wrapped_decorator import signature_safe_contextmanager
21
import six
22
from .framework import Program, default_main_program, Variable
23
from . import core
24 25
from . import compiler
from .. import compat as cpt
26
from .trainer_factory import TrainerFactory
27

T
Tink_Y 已提交
28
__all__ = ['Executor', 'global_scope', 'scope_guard']
Y
Yu Yang 已提交
29

Y
Yu Yang 已提交
30
g_scope = core.Scope()
F
flame 已提交
31 32
InferNativeConfig = core.NativeConfig
InferAnalysisConfig = core.AnalysisConfig
Y
Yu Yang 已提交
33

Y
Yu Yang 已提交
34

Y
Yang Yu 已提交
35
def global_scope():
Y
yuyang18 已提交
36 37 38 39 40 41 42
    """
    Get the global/default scope instance. There are a lot of APIs use
    :code:`global_scope` as its default value, e.g., :code:`Executor.run`

    Returns:
        Scope: The global/default scope instance.
    """
Y
Yang Yu 已提交
43 44 45
    return g_scope


46
def _switch_scope(scope):
Y
Yang Yu 已提交
47 48 49 50 51 52
    global g_scope
    ex = g_scope
    g_scope = scope
    return ex


S
rename  
sneaxiy 已提交
53
@signature_safe_contextmanager
Y
Yang Yu 已提交
54
def scope_guard(scope):
Y
yuyang18 已提交
55 56 57 58 59 60 61 62 63 64 65 66 67
    """
    Change the global/default scope instance by Python `with` statement. All
    variable in runtime will assigned to the new scope.

    Examples:
        >>> import paddle.fluid as fluid
        >>> new_scope = fluid.Scope()
        >>> with fluid.scope_guard(new_scope):
        >>>     ...

    Args:
        scope: The new global/default scope.
    """
68
    ex = _switch_scope(scope)
Y
Yang Yu 已提交
69
    yield
70
    _switch_scope(ex)
Y
Yang Yu 已提交
71 72


D
dzhwinter 已提交
73
def as_numpy(tensor):
74 75 76 77 78 79 80 81 82 83 84 85 86 87 88
    """
    Convert a Tensor to a numpy.ndarray, its only support Tensor without LoD information.
    For higher dimensional sequence data, please use LoDTensor directly.
    Examples:
        >>> import paddle.fluid as fluid
        >>> outs = executor.run(...)
        >>> np_outs = map(lambda x: as_numpy(x), outs)
        >>>     ...

    Args:
       tensor(Variable): a instance of Tensor

    Returns:
        numpy.ndarray
    """
C
chengduo 已提交
89 90
    if isinstance(tensor, core.LoDTensorArray):
        return [as_numpy(t) for t in tensor]
D
dzhwinter 已提交
91 92 93 94
    if isinstance(tensor, list):
        return [as_numpy(t) for t in tensor]
    assert isinstance(tensor, core.LoDTensor)
    lod = tensor.lod()
95
    if len(lod) > 0:
D
dzhwinter 已提交
96
        raise RuntimeError("Some of your fetched tensors hold LoD information. \
97 98 99 100
            They can not be completely cast to Python ndarray. \
            Please set the parameter 'return_numpy' as 'False' to \
            return LoDTensor itself directly.")
    return np.array(tensor)
D
dzhwinter 已提交
101 102


103 104 105 106 107 108 109 110 111 112 113 114
def has_feed_operators(block, feed_targets, feed_holder_name):
    """ Check whether the block already has feed operators.

    Return false if the block does not have any feed operators.
    If some feed operators have been prepended to the block, check that
    the info contained in these feed operators matches the feed_targets
    and feed_holder_name. Raise exception when any mismatch is found.
    Return true when the block has feed operators with matching info.

    Args:
        block: a block instance (typically global block of a program)
        feed_targets: a dictionary of {feed_target_name: feed_target_data}
X
xuwei06 已提交
115 116
        feed_holder_name: the name of the variable that holds the data of
            all feed targets. The type of this feed_holder variable is
117 118 119
            FEED_MINIBATCH, which is essentially vector<LoDTensor>.

    Returns:
X
xuwei06 已提交
120
        A boolean value that indicates whether a block has feed operators
121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142
        that match the info contained in feed_targets and feed_holder_name.
    """

    feed_count = 0
    for op in block.ops:
        if op.desc.type() == 'feed':
            feed_count += 1
            assert op.desc.input('X')[0] == feed_holder_name
            feed_target_name = op.desc.output('Out')[0]
            if feed_target_name not in feed_targets:
                raise Exception("'feed_targets' does not have {} variable".
                                format(feed_target_name))
        else:
            break
    if feed_count > 0 and feed_count != len(feed_targets):
        raise Exception(
            "Feed operators in program desc do not match 'feed_targets'")
    return feed_count > 0


def has_fetch_operators(block, fetch_targets, fetch_holder_name):
    """ Check whether the block already has fetch operators.
X
xuwei06 已提交
143

144 145 146 147 148 149 150 151 152
    Return false if the block does not have any fetch operators.
    If some fetch operators have been appended to the block, check that
    the info contained in these fetch operators matches the fetch_targets
    and fetch_holder_name. Raise exception when any mismatch is found.
    Return true when the block has fetch operators with matching info.

    Args:
        block: a block instance (typically global block of a program)
        fetch_targets: a dictionary of {fetch_target_name: fetch_target_data}
X
xuwei06 已提交
153 154 155
        fetch_holder_name: the name of the variable that holds the data of
            all fetch targets. The type of this fetch_holder variable is
            FETCH_LIST, which is essentially vector<LoDTensor>.
156

X
xuwei06 已提交
157 158 159
    Return:
        A boolean value that indicates whether a block has fetch operators
        that match the info contained in fetch_targets and fetch_holder_name.
160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180
    """

    fetch_count = 0
    for op in block.ops:
        if op.desc.type() == 'fetch':
            fetch_count += 1
            assert op.desc.output('Out')[0] == fetch_holder_name
            fetch_target_name = op.desc.input('X')[0]
            if fetch_target_name not in [
                    var.desc.name() for var in fetch_targets
            ]:
                raise Exception("'fetch_targets' does not have {} variable".
                                format(fetch_target_name))
            idx = op.desc.attr('col')
            assert fetch_target_name == fetch_targets[idx].desc.name()
    if fetch_count > 0 and fetch_count != len(fetch_targets):
        raise Exception(
            "Fetch operators in program desc do not match 'fetch_targets'")
    return fetch_count > 0


W
Wu Yi 已提交
181
def _fetch_var(name, scope=None, return_numpy=True):
X
xuwei06 已提交
182
    """
C
chengduoZH 已提交
183 184 185
    Fetch the value of the variable with the given name from the
    given scope.

X
xuwei06 已提交
186
    Args:
187 188 189 190
        name(str): name of the variable. Typically, only persistable variables
            can be found in the scope used for running the program.
        scope(core.Scope|None): scope object. It should be the scope where
            you pass to Executor.run() when running your program.
C
chengduoZH 已提交
191 192 193 194
            If None, global_scope() will be used. Default None.
        return_numpy(bool): whether convert the tensor to numpy.ndarray.
            Default True.

X
xuwei06 已提交
195 196 197 198 199 200
    Returns:
       LodTensor|numpy.ndarray
    """
    assert isinstance(name, str)
    if scope is None:
        scope = global_scope()
S
sneaxiy 已提交
201
    assert isinstance(scope, core._Scope)
X
xuwei06 已提交
202

Y
Yibing Liu 已提交
203
    var = scope.find_var(name)
204 205 206 207
    assert var is not None, (
        "Cannot find " + name + " in scope. Perhaps you need to make the"
        " variable persistable by using var.persistable = True in your"
        " program.")
X
xuwei06 已提交
208 209 210 211 212 213
    tensor = var.get_tensor()
    if return_numpy:
        tensor = as_numpy(tensor)
    return tensor


X
polish  
Xin Pan 已提交
214 215 216 217 218 219 220 221 222
def _to_name_str(var):
    if isinstance(var, Variable):
        return var.desc.name()
    elif isinstance(var, str):
        return var
    elif isinstance(var, six.string_types):
        return str(var)
    else:
        raise TypeError(str(var) + " should be Variable or str")
Q
qiaolongfei 已提交
223 224


X
polish  
Xin Pan 已提交
225 226 227
def _get_program_cache_key(feed, fetch_list):
    feed_var_names = list(feed.keys())
    fetch_var_names = list(map(_to_name_str, fetch_list))
Q
qiaolongfei 已提交
228 229 230 231

    return str(feed_var_names + fetch_var_names)


W
Wu Yi 已提交
232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262
def _as_lodtensor(data, place):
    """
        Convert numpy.ndarray to Tensor, its only support Tensor without LoD information.
        For higher dimensional sequence data, please use LoDTensor directly.

        Examples:
            >>> import paddle.fluid as fluid
            >>> place = fluid.CPUPlace()
            >>> exe = fluid.executor(place)
            >>> data = np.array(size=(100, 200, 300))
            >>> np_outs = map(lambda x: fluid.executor._as_lodtensor(x, place), data)
            >>>     ...

        Args:
            data(numpy.ndarray): a instance of array

        Returns:
            LoDTensor
        """
    if isinstance(data, list):
        raise RuntimeError("Some of your feed data hold LoD information. \
                They can not be completely cast from a list of Python \
                ndarray to LoDTensor. Please convert data to LoDTensor \
                directly before feeding the data.\
                ")
    # single tensor case
    tensor = core.LoDTensor()
    tensor.set(data, place)
    return tensor


Y
Yu Yang 已提交
263
class Executor(object):
264
    """
S
Fix doc  
sneaxiy 已提交
265 266
    An Executor in Python, supports single/multiple-GPU running, and single/multiple-CPU running.
    Python executor takes a program, adds feed operators and fetch operators to this program according
267
    to feed map and fetch_list. Feed map provides input data for the program. fetch_list provides
S
Fix doc  
sneaxiy 已提交
268
    the variables(or names) that user wants to get after program runs. Note: the executor will run all
269
    operators in the program but not only the operators dependent by the fetch_list.
S
Fix doc  
sneaxiy 已提交
270 271 272
    It stores the global variables into the global scope, and creates a local scope for the temporary
    variables. The contents in local scope may be discarded after every minibatch forward/backward
    finished. But the global scope variables will be persistent through different runs.
273

X
add doc  
Xin Pan 已提交
274 275

    Example:
S
Fix doc  
sneaxiy 已提交
276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297

        .. code-block:: python

            # First create the Executor.
            place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
            exe = fluid.Executor(place)

            # Run the startup program once and only once.
            # Not need to optimize/compile the startup program.
            exe.run(fluid.default_startup_program())

            # Run the main program directly without compile.
            loss, = exe.run(fluid.default_main_program(),
                            feed=feed_dict,
                            fetch_list=[loss.name])
            # Or, compiled the program and run. See `CompiledProgram` for more detail.
            compiled_prog = compiler.CompiledProgram(
                fluid.default_main_program()).with_data_parallel(
                loss_name=loss.name)
            loss, = exe.run(compiled_prog,
                            feed=feed_dict,
                            fetch_list=[loss.name])
X
add doc  
Xin Pan 已提交
298

299 300 301 302
    Args:
        place(core.CPUPlace|core.CUDAPlace(n)): indicate the executor run on which device
    """

D
dzhwinter 已提交
303 304
    def __init__(self, place):
        self.place = place
Q
qiaolongfei 已提交
305
        self.program_caches = dict()
306 307 308
        p = core.Place()
        p.set_place(self.place)
        self._default_executor = core.Executor(p)
Y
Yancey1989 已提交
309
        self._closed = False
D
dzhwinter 已提交
310

Q
Qiao Longfei 已提交
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
    def _get_program_cache(self, program_cache_key):
        return self.program_caches.get(program_cache_key, None)

    def _add_program_cache(self, program_cache_key, program):
        self.program_caches[program_cache_key] = program

    def _add_feed_fetch_ops(self, program, feed, fetch_list, feed_var_name,
                            fetch_var_name):
        tmp_program = program.clone()

        global_block = tmp_program.global_block()

        if feed_var_name in global_block.vars:
            feed_var = global_block.var(feed_var_name)
        else:
            feed_var = global_block.create_var(
                name=feed_var_name,
                type=core.VarDesc.VarType.FEED_MINIBATCH,
                persistable=True)

        if fetch_var_name in global_block.vars:
            fetch_var = global_block.var(fetch_var_name)
        else:
            fetch_var = global_block.create_var(
                name=fetch_var_name,
                type=core.VarDesc.VarType.FETCH_LIST,
                persistable=True)

        # prepend feed operators
        if not has_feed_operators(global_block, feed, feed_var_name):
            for i, name in enumerate(feed):
                out = global_block.var(name)
W
Wu Yi 已提交
343
                global_block._prepend_op(
Q
Qiao Longfei 已提交
344 345 346 347 348 349 350 351
                    type='feed',
                    inputs={'X': [feed_var]},
                    outputs={'Out': [out]},
                    attrs={'col': i})

        # append fetch_operators
        if not has_fetch_operators(global_block, fetch_list, fetch_var_name):
            for i, var in enumerate(fetch_list):
M
minqiyang 已提交
352 353 354
                assert isinstance(var, Variable) or isinstance(
                    var, six.string_types), (
                        "Wrong type for fetch_list[%s]: %s" % (i, type(var)))
Q
Qiao Longfei 已提交
355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
                global_block.append_op(
                    type='fetch',
                    inputs={'X': [var]},
                    outputs={'Out': [fetch_var]},
                    attrs={'col': i})

        return tmp_program

    def _feed_data(self, program, feed, feed_var_name, scope):
        # feed var to framework
        for op in program.global_block().ops:
            if op.desc.type() == 'feed':
                feed_target_name = op.desc.output('Out')[0]
                cur_feed = feed[feed_target_name]
                if not isinstance(cur_feed, core.LoDTensor):
W
Wu Yi 已提交
370
                    cur_feed = _as_lodtensor(cur_feed, self.place)
Q
Qiao Longfei 已提交
371 372 373 374 375 376 377 378
                idx = op.desc.attr('col')
                core.set_feed_variable(scope, cur_feed, feed_var_name, idx)
            else:
                break

    def _fetch_data(self, fetch_list, fetch_var_name, scope):
        outs = [
            core.get_fetch_variable(scope, fetch_var_name, i)
M
minqiyang 已提交
379
            for i in six.moves.range(len(fetch_list))
Q
Qiao Longfei 已提交
380 381 382
        ]
        return outs

S
Fix doc  
sneaxiy 已提交
383 384 385 386 387 388
    '''
    TODO(typhoonzero): Define "no longer use" meaning? Can user create
    a new Executor for the same program and run?
    TODO(panyx0718): Why ParallelExecutor doesn't have close?
    '''

Y
Yancey1989 已提交
389 390 391 392
    def close(self):
        """
        Close this executor.

X
fix  
Xin Pan 已提交
393
        You can no longer use this executor after calling this method.
Y
Yancey1989 已提交
394 395 396 397 398 399 400 401 402
        For the distributed training, this method would free the resource on PServers related to
        the current Trainer.

        Example:
            >>> cpu = core.CPUPlace()
            >>> exe = Executor(cpu)
            >>> ...
            >>> exe.close()
        """
403 404
        if not self._closed:
            self._default_executor.close()
Y
Yancey1989 已提交
405
            self._closed = True
Y
Yancey1989 已提交
406

X
fix  
Xin Pan 已提交
407
    def _run_parallel(self, program, scope, feed, fetch_list, fetch_var_name,
X
polish  
Xin Pan 已提交
408
                      return_numpy):
409
        exe = program._executor
410 411 412 413 414 415 416 417 418 419 420
        if isinstance(feed, dict):
            feed_tensor_dict = dict()
            for feed_name in feed:
                feed_tensor = feed[feed_name]
                if not isinstance(feed_tensor, core.LoDTensor):
                    feed_tensor = core.LoDTensor()
                    # always set to CPU place, since the tensor need to be splitted
                    # it is fast in CPU
                    feed_tensor.set(feed[feed_name], core.CPUPlace())
                feed_tensor_dict[feed_name] = feed_tensor

421
            exe.feed_and_split_tensor_into_local_scopes(feed_tensor_dict)
422
        elif isinstance(feed, list) or isinstance(feed, tuple):
X
fix  
Xin Pan 已提交
423
            if len(feed) != len(program._places):
424 425 426 427 428 429 430 431 432 433 434 435 436 437
                raise ValueError(
                    "Feed a list of tensor, the list should be the same size as places"
                )

            res = list()
            for i, each in enumerate(feed):
                if not isinstance(each, dict):
                    raise TypeError(
                        "Each element of feed list should be a dict")
                res_dict = dict()
                for feed_name in each:
                    tensor = each[feed_name]
                    if not isinstance(tensor, core.LoDTensor):
                        tmp = core.LoDTensor()
X
fix  
Xin Pan 已提交
438
                        tmp.set(tensor, program._places[i])
439 440 441
                        tensor = tmp
                    res_dict[feed_name] = tensor
                res.append(res_dict)
442
            exe.feed_tensors_into_local_scopes(res)
443

X
polish  
Xin Pan 已提交
444
        fetch_var_names = list(map(_to_name_str, fetch_list))
445
        exe.run(fetch_var_names, fetch_var_name)
446 447 448 449 450 451
        arr = scope.find_var(fetch_var_name).get_lod_tensor_array()

        if return_numpy:
            return as_numpy(arr)
        return [arr[i] for i in range(len(arr))]

Y
Yu Yang 已提交
452
    def run(self,
Y
Yu Yang 已提交
453
            program=None,
454 455
            feed=None,
            fetch_list=None,
Y
Yu Yang 已提交
456
            feed_var_name='feed',
Y
Yu Yang 已提交
457
            fetch_var_name='fetch',
D
dzhwinter 已提交
458
            scope=None,
459 460
            return_numpy=True,
            use_program_cache=False):
461 462
        """
        Run program by this Executor. Feed data by feed map, fetch result by fetch_list.
Q
qiaolongfei 已提交
463 464
        Python executor takes a program, add feed operators and fetch operators to this program according
        to feed map and fetch_list. Feed map provides input data for the program. fetch_list provides
465 466 467
        the variables(or names) that user want to get after program run.

        Note: the executor will run all
Q
qiaolongfei 已提交
468 469
        operators in the program but not only the operators dependent by the fetch_list

470
        Args:
X
add doc  
Xin Pan 已提交
471
            program(Program|CompiledProgram): the program that need to run,
X
fix  
Xin Pan 已提交
472
                if not provided, then default_main_program (not compiled) will be used.
X
add doc  
Xin Pan 已提交
473
            feed(dict): feed variable map, e.g. {"image": ImageData, "label": LabelData}
Z
Zeng Jinle 已提交
474 475 476 477 478 479 480 481
            fetch_list(list): a list of variable or variable names that user 
                wants to get, this method will return them according to this list.
            feed_var_name(str): the name for the input variable of 
                feed Operator.
            fetch_var_name(str): the name for the output variable of 
                fetch Operator.
            scope(Scope): the scope used to run this program, you can switch 
                it to different scope. default is global_scope
482
            return_numpy(bool): if convert the fetched tensor to numpy
Z
Zeng Jinle 已提交
483 484 485 486 487 488
            use_program_cache(bool): whether to use the cached program 
                settings across batches. Setting it be true would be faster 
                only when (1) the program is not compiled with data parallel, 
                and (2) program, feed variable names and fetch_list variable 
                names do not changed compared to the last step. 
                
489 490 491 492 493 494 495
        Returns:

            list(numpy.array): fetch result according to fetch_list.


        Examples:

T
Tink_Y 已提交
496 497 498 499 500
            >>> data = fluid.layers.data(name='X', shape=[1], dtype='float32')
            >>> out = fluid.layers.create_tensor(dtype='float32')
            >>> hidden = fluid.layers.fc(input=data, size=10)
            >>> fluid.layers.assign(hidden,out)
            >>> loss = fluid.layers.mean(out)
501
            >>> adam = fluid.optimizer.Adam()
T
Tink_Y 已提交
502
						>>> adam.minimize(loss)
503 504

            >>> cpu = core.CPUPlace()
T
Tink_Y 已提交
505 506
            >>> exe = fluid.Executor(cpu)
            >>> exe.run(fluid.default_startup_program())
507 508 509 510 511

            >>> x = numpy.random.random(size=(10, 1)).astype('float32')
            >>> outs = exe.run(
            >>>     feed={'X': x},
            >>>     fetch_list=[loss.name])
512
        """
Y
Yancey1989 已提交
513 514 515 516

        if self._closed:
            raise RuntimeError("Attempted to use a closed Executor")

517 518
        if scope is None:
            scope = global_scope()
X
polish  
Xin Pan 已提交
519 520
        if fetch_list is None:
            fetch_list = []
521

X
polish  
Xin Pan 已提交
522 523
        compiled = isinstance(program, compiler.CompiledProgram)
        # For backward compatibility, run directly.
524 525 526
        if not compiled:
            return self._run(
                program,
527
                self._default_executor,
528 529 530 531 532 533 534 535 536 537 538
                feed=feed,
                fetch_list=fetch_list,
                feed_var_name=feed_var_name,
                fetch_var_name=fetch_var_name,
                scope=scope,
                return_numpy=return_numpy,
                use_program_cache=use_program_cache)

        program._compile(scope, self.place)
        if program._is_data_parallel:
            return self._run_parallel(
X
fix  
Xin Pan 已提交
539
                program,
540 541 542
                scope=scope,
                feed=feed,
                fetch_list=fetch_list,
X
polish  
Xin Pan 已提交
543
                fetch_var_name=fetch_var_name,
544
                return_numpy=return_numpy)
F
flame 已提交
545
        elif program._is_inference:
X
Xin Pan 已提交
546
            return self._run_inference(program._executor, feed)
547
        else:
X
Xin Pan 已提交
548 549
            # TODO(panyx0718): Can compile program to optimize executor
            # performance.
X
Xin Pan 已提交
550
            # TODO(panyx0718): executor should be able to run graph.
X
Xin Pan 已提交
551
            assert program._program, "CompiledProgram is compiled from graph, can only run with_data_parallel."
552 553
            return self._run(
                program._program,
554
                self._default_executor,
555 556 557 558 559 560 561 562
                feed=feed,
                fetch_list=fetch_list,
                feed_var_name=feed_var_name,
                fetch_var_name=fetch_var_name,
                scope=scope,
                return_numpy=return_numpy,
                use_program_cache=use_program_cache)

563 564
    def _run(self, program, exe, feed, fetch_list, feed_var_name,
             fetch_var_name, scope, return_numpy, use_program_cache):
565

566 567
        if feed is None:
            feed = {}
S
sneaxiy 已提交
568 569 570 571
        elif isinstance(feed, (list, tuple)):
            assert len(feed) == 1, "Not compiled with data parallel"
            feed = feed[0]

Q
qiaolongfei 已提交
572
        if not isinstance(feed, dict):
D
dzhwinter 已提交
573 574 575
            raise TypeError(
                "feed requires dict as its Parameter. But you passed in %s" %
                (type(feed)))
Y
Yu Yang 已提交
576
        if program is None:
Y
Yu Yang 已提交
577
            program = default_main_program()
Y
Yu Yang 已提交
578

Y
Yu Yang 已提交
579
        if not isinstance(program, Program):
D
dzhwinter 已提交
580 581 582
            raise TypeError(
                "Executor requires Program as its Parameter. But you passed in %s"
                % (type(program)))
Y
Yu Yang 已提交
583

584
        cache_key = _get_program_cache_key(feed, fetch_list)
585
        if use_program_cache:
Q
Qiao Longfei 已提交
586 587 588 589 590 591 592 593 594 595
            cached_program = self._get_program_cache(cache_key)
            if cached_program is None:
                cached_program = self._add_feed_fetch_ops(
                    program=program,
                    feed=feed,
                    fetch_list=fetch_list,
                    feed_var_name=feed_var_name,
                    fetch_var_name=fetch_var_name)
                self._add_program_cache(cache_key, cached_program)
            program = cached_program
596
        else:
Q
Qiao Longfei 已提交
597 598 599 600 601 602 603 604 605
            self.program_caches.pop(cache_key, None)
            program = self._add_feed_fetch_ops(
                program=program,
                feed=feed,
                fetch_list=fetch_list,
                feed_var_name=feed_var_name,
                fetch_var_name=fetch_var_name)

        self._feed_data(program, feed, feed_var_name, scope)
S
sneaxiy 已提交
606
        exe.run(program.desc, scope, 0, True, True, fetch_var_name)
Q
Qiao Longfei 已提交
607
        outs = self._fetch_data(fetch_list, fetch_var_name, scope)
D
dzhwinter 已提交
608 609 610
        if return_numpy:
            outs = as_numpy(outs)
        return outs
F
flame 已提交
611

X
Xin Pan 已提交
612 613
    def _run_inference(self, exe, feed):
        return exe.run(feed)
D
dongdaxiang 已提交
614

615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630
    def _dump_debug_info(self, program=None, trainer=None):
        with open(str(id(program)) + "_train_desc.prototxt", "w") as fout:
            fout.write(trainer._desc())
        if program._fleet_opt:
            with open("fleet_desc.prototxt", "w") as fout:
                fout.write(str(program._fleet_opt["fleet_desc"]))

    def _prepare_trainer(self,
                         program=None,
                         dataset=None,
                         scope=None,
                         thread=0,
                         debug=False,
                         fetch_list=None,
                         fetch_info=None,
                         print_period=100):
D
dongdaxiang 已提交
631 632 633 634
        if scope is None:
            scope = global_scope()
        if fetch_list is None:
            fetch_list = []
D
dongdaxiang 已提交
635 636 637
        if fetch_info is None:
            fetch_info = []
        assert len(fetch_list) == len(fetch_info)
D
dongdaxiang 已提交
638 639
        compiled = isinstance(program, compiler.CompiledProgram)
        if not compiled:
640 641
            trainer = TrainerFactory()._create_trainer(program._fleet_opt)
            trainer._set_program(program)
642
        else:
643
            trainer = TrainerFactory()._create_trainer(
644
                program.program._fleet_opt)
645
            trainer._set_program(program.program)
646
        if thread <= 0:
D
dongdaxiang 已提交
647 648
            if dataset.thread_num <= 0:
                raise RuntimeError(
649 650
                    "You should set thread num first, either in Dataset"
                    "or in Executor.train_from_dataset")
D
dongdaxiang 已提交
651
            else:
652
                trainer._set_thread(dataset.thread_num)
653
        else:
654 655 656
            trainer._set_thread(thread)
        trainer._set_debug(debug)
        trainer._set_fetch_var_and_info(fetch_list, fetch_info, print_period)
657
        return scope, trainer
658 659 660 661 662 663

    def infer_from_dataset(self,
                           program=None,
                           dataset=None,
                           scope=None,
                           thread=0,
664 665 666 667
                           debug=False,
                           fetch_list=None,
                           fetch_info=None,
                           print_period=100):
668 669 670 671 672 673
        """
        The document of infer_from_dataset is almost the same as
        train_from_dataset, except that in distributed training,
        push gradients will be disabled in infer_from_dataset.
        infer_from_dataset() can be used for evaluation in multi-thread
        very easily.
674

675 676 677 678 679
        Args:
            program(Program|CompiledProgram): the program that needs to be run,
               if not provided, then default_main_program (not compiled) will be used.
            dataset(paddle.fluid.Dataset): dataset created outside this function,
               a user should provide a well-defined dataset before calling this function.
680
               Please check the document of Dataset if needed. default is None
681 682 683
            scope(Scope): the scope used to run this program, you can switch it to different scope
               for each run. default is global_scope
            thread(int): number of thread a user wants to run in this function. The actual number
684 685
               of thread will be min(Dataset.thread_num, thread) if thread > 0, default is 0
            debug(bool): whether a user wants to run infer_from_dataset, default is False
686
            fetch_list(Variable List): fetch variable list, each variable
687 688 689
                                       will be printed during training, default is None
            fetch_info(String List): print information for each variable, default is None
            print_period(int): the number of mini-batches for each print, default is 100
690

691 692 693 694
        Returns:
            None

        Examples:
695 696

            .. code-block:: python
697

698 699 700 701 702 703 704 705 706 707 708 709
                import paddle.fluid as fluid
                place = fluid.CPUPlace()
                exe = fluid.Executor(place)
                x = fluid.layers.data(name="x", type="int64")
                y = fluid.layers.data(name="y", type="int64")
                dataset = fluid.DatasetFactory().create_dataset()
                dataset.set_use_var([x, y])
                filelist = ["dataA.txt", "dataB.txt"]
                dataset.set_filelist(filelist)
                exe.run(fluid.default_startup_program())
                exe.infer_from_dataset(program=fluid.default_main_program(),
                                       dataset=dataset)        
710

711
        """
712 713 714
        if dataset == None:
            raise RuntimeError("dataset is needed and should be initialized")

715
        if not isinstance(self.place, core.CPUPlace):
D
dongdaxiang 已提交
716 717
            raise RuntimeError("infer_from_dataset is verified on CPUPlace"
                               "We will open CUDAPlace in the future")
718

719
        scope, trainer = self._prepare_trainer(
720 721 722 723 724 725 726 727
            program=program,
            dataset=dataset,
            scope=scope,
            thread=thread,
            debug=debug,
            fetch_list=fetch_list,
            fetch_info=fetch_info,
            print_period=print_period)
728
        trainer._set_infer(True)
729
        trainer._gen_trainer_desc()
730 731 732 733 734 735
        dataset._prepare_to_run()
        if debug:
            self._dump_debug_info(program=program, trainer=trainer)
        self._default_executor.run_from_dataset(program.desc, scope,
                                                dataset.dataset,
                                                trainer._desc())
736
        return None
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 765 766 767 768 769 770 771 772

    def train_from_dataset(self,
                           program=None,
                           dataset=None,
                           scope=None,
                           thread=0,
                           debug=False,
                           fetch_list=None,
                           fetch_info=None,
                           print_period=100):
        """
        Train from a pre-defined Dataset. Dataset is defined in paddle.fluid.dataset.
        Given a program, either a program or compiled program, train_from_dataset will
        consume all data samples in dataset. Input scope can be given by users. By default,
        scope is global_scope(). The total number of thread run in training is `thread`.
        Thread number used in training will be minimum value of threadnum in Dataset and
        the value of thread in this interface. Debug can be set so that executor will display
        Run-Time for all operators and the throughputs of current training task.
        
        Note: train_from_dataset will destroy all resources created within executor for each run.

        Args:
            program(Program|CompiledProgram): the program that needs to be run,
               if not provided, then default_main_program (not compiled) will be used.
            dataset(paddle.fluid.Dataset): dataset created outside this function,
               a user should provide a well-defined dataset before calling this function.
               Please check the document of Dataset if needed.
            scope(Scope): the scope used to run this program, you can switch it to different scope
               for each run. default is global_scope
            thread(int): number of thread a user wants to run in this function. The actual number
               of thread will be min(Dataset.thread_num, thread)
            debug(bool): whether a user wants to run train_from_dataset 
            fetch_list(Variable List): fetch variable list, each variable
                                       will be printed during training
            fetch_info(String List): print information for each variable
            print_period(int): the number of mini-batches for each print
773 774 775

        Returns:
            None
776
        
777
        Examples:
778
        
779 780 781 782 783 784 785 786 787 788 789 790 791 792 793
            .. code-block:: python

              import paddle.fluid as fluid
              place = fluid.CPUPlace()
              exe = fluid.Executor(place)
              x = fluid.layers.data(name="x", type="int64")
              y = fluid.layers.data(name="y", type="int64")
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_use_var([x, y])
              dataset.set_thread(2)
              filelist = ["dataA.txt", "dataB.txt"]
              dataset.set_filelist(filelist)
              exe.run(fluid.default_startup_program())
              exe.train_from_dataset(program=fluid.default_main_program(),
                                     dataset=dataset)
794 795

        """
796 797 798
        if dataset == None:
            raise RuntimeError("dataset is need and should be initialized")

799
        if not isinstance(self.place, core.CPUPlace):
D
dongdaxiang 已提交
800 801
            raise RuntimeError("train_from_dataset is verified on CPUPlace"
                               "We will open CUDAPlace in the future")
802

803
        scope, trainer = self._prepare_trainer(
804 805 806 807 808 809 810 811
            program=program,
            dataset=dataset,
            scope=scope,
            thread=thread,
            debug=debug,
            fetch_list=fetch_list,
            fetch_info=fetch_info,
            print_period=print_period)
812
        trainer._gen_trainer_desc()
D
dongdaxiang 已提交
813
        dataset._prepare_to_run()
D
dongdaxiang 已提交
814
        if debug:
815
            self._dump_debug_info(program=program, trainer=trainer)
D
dongdaxiang 已提交
816 817 818
        self._default_executor.run_from_dataset(program.desc, scope,
                                                dataset.dataset,
                                                trainer._desc())
819
        return None