executor.py 44.4 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

Z
Zeng Jinle 已提交
17
import logging
18 19
import os
import multiprocessing
C
chengduo 已提交
20
import sys
21
import warnings
D
dzhwinter 已提交
22
import numpy as np
S
rename  
sneaxiy 已提交
23
from .wrapped_decorator import signature_safe_contextmanager
24
import six
H
Huihuang Zheng 已提交
25
from .framework import Program, default_main_program, Variable, convert_np_dtype_to_dtype_
26
from . import core
27 28
from . import compiler
from .. import compat as cpt
29
from .trainer_factory import TrainerFactory
30
from .trainer_factory import FetchHandlerMonitor
31

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

Y
Yu Yang 已提交
34
g_scope = core.Scope()
F
flame 已提交
35 36
InferNativeConfig = core.NativeConfig
InferAnalysisConfig = core.AnalysisConfig
Y
Yu Yang 已提交
37

Y
Yu Yang 已提交
38

Y
Yang Yu 已提交
39
def global_scope():
Y
yuyang18 已提交
40 41 42 43
    """
    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`

44 45 46 47 48 49 50 51 52
    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          import numpy

          fluid.global_scope().var("data").get_tensor().set(numpy.ones((2, 2)), fluid.CPUPlace())
          numpy.array(fluid.global_scope().find_var("data").get_tensor())

Y
yuyang18 已提交
53 54 55
    Returns:
        Scope: The global/default scope instance.
    """
Y
Yang Yu 已提交
56 57 58
    return g_scope


59
def _switch_scope(scope):
Y
Yang Yu 已提交
60 61 62 63 64 65
    global g_scope
    ex = g_scope
    g_scope = scope
    return ex


S
rename  
sneaxiy 已提交
66
@signature_safe_contextmanager
Y
Yang Yu 已提交
67
def scope_guard(scope):
Y
yuyang18 已提交
68 69 70 71
    """
    Change the global/default scope instance by Python `with` statement. All
    variable in runtime will assigned to the new scope.

L
lujun 已提交
72 73 74
    Args:
        scope: The new global/default scope.

Y
yuyang18 已提交
75
    Examples:
76 77
        .. code-block:: python

78
            import paddle.fluid as fluid
L
lujun 已提交
79
            import numpy
Y
yuyang18 已提交
80

L
lujun 已提交
81 82 83 84
            new_scope = fluid.Scope()
            with fluid.scope_guard(new_scope):
                 fluid.global_scope().var("data").get_tensor().set(numpy.ones((2, 2)), fluid.CPUPlace())
            numpy.array(new_scope.find_var("data").get_tensor())
Y
yuyang18 已提交
85
    """
L
lujun 已提交
86

87
    ex = _switch_scope(scope)
Y
Yang Yu 已提交
88
    yield
89
    _switch_scope(ex)
Y
Yang Yu 已提交
90 91


D
dzhwinter 已提交
92
def as_numpy(tensor):
93 94 95
    """
    Convert a Tensor to a numpy.ndarray, its only support Tensor without LoD information.
    For higher dimensional sequence data, please use LoDTensor directly.
96

97
    Examples:
98 99 100 101 102 103 104 105 106 107
        .. 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())
108 109 110 111 112 113 114

    Args:
       tensor(Variable): a instance of Tensor

    Returns:
        numpy.ndarray
    """
C
chengduo 已提交
115 116
    if isinstance(tensor, core.LoDTensorArray):
        return [as_numpy(t) for t in tensor]
D
dzhwinter 已提交
117 118 119 120
    if isinstance(tensor, list):
        return [as_numpy(t) for t in tensor]
    assert isinstance(tensor, core.LoDTensor)
    lod = tensor.lod()
121
    if len(lod) > 0:
D
dzhwinter 已提交
122
        raise RuntimeError("Some of your fetched tensors hold LoD information. \
123 124 125
            They can not be completely cast to Python ndarray. \
            Please set the parameter 'return_numpy' as 'False' to \
            return LoDTensor itself directly.")
Q
qingqing01 已提交
126 127 128 129
    if tensor._is_initialized():
        return np.array(tensor)
    else:
        return None
D
dzhwinter 已提交
130 131


H
Huihuang Zheng 已提交
132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216
def dtype_is_compatible_with(first, second):
    """
    Returns True if the first dtype can be compatible the second one.
    Currently, we require the two dtype's have to be same.
      
    Args:
        dtype (np.dtype|VarType|str): The type of data: float32, int64, etc.
    
    Returns:
        True if the two types are same.
    """
    if not isinstance(first, core.VarDesc.VarType):
        first = convert_np_dtype_to_dtype_(first)
    if not isinstance(second, core.VarDesc.VarType):
        second = convert_np_dtype_to_dtype_(second)
    return first == second


def dimension_is_compatible_with(first, second):
    """
    Returns True if the two dimensions are compatible.

    A dimension is compatible with the other if:
    1. The length of the dimensions are same.
    2. Each non-negative number of the two dimentions are same.
    3. For negative number or 'None' in a dimention, it means unknown so it
       is compatible with any number.

    Args:
        first (list/tuple): integers representing shape. "None" or negative
            number means unknown.
        second (list/tuple): integers representing shape. "None" or negative
            number means unknown.

    Returns:
        True if the two dimensions are compatible.
    """

    dim_len = len(first)
    if dim_len != len(second):
        return False

    for i in range(dim_len):
        if first[i] is None or first[i] < 0:
            continue
        if second[i] is None or second[i] < 0:
            continue
        if first[i] != second[i]:
            return False

    return True


def check_feed_shape_type(var, feed):
    """
    Returns True if the variable doesn't require feed check or it is compatible
    with the shape and have same dtype as the feeded value.

    A dimension is compatible with the other if:
    1. The length of the dimensions are same.
    2. Each non-negative number of the two dimentions are same.
    3. For negative number or 'None' in a dimention, it means unknown so it
       is compatible with any number.
    
    Args:
        var (Variable): the Variable object
        feed (LoDTensor): the feeded value, which must be a LoDTensor
    Returns:
        True if the shape and dtype of variable is compatible with the feed value
    Raises:
        ValueError: if the shape or dtype of the variable is not compatible with
            the feed value
    """
    if var.desc.need_check_feed():
        if not dimension_is_compatible_with(feed.shape(), var.shape):
            raise ValueError('Cannot feed value of shape %r for Variable %r, '
                             'which has shape %r' %
                             (feed.shape, var.name, var.shape))
        if not dtype_is_compatible_with(feed._dtype(), var.dtype):
            raise ValueError('Cannot feed value of type %r for Variable %r, '
                             'which has type %r' %
                             (feed._dtype(), var.name, var.dtype))
    return True


217 218 219 220 221 222 223 224 225 226 227 228
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 已提交
229 230
        feed_holder_name: the name of the variable that holds the data of
            all feed targets. The type of this feed_holder variable is
231 232 233
            FEED_MINIBATCH, which is essentially vector<LoDTensor>.

    Returns:
X
xuwei06 已提交
234
        A boolean value that indicates whether a block has feed operators
235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256
        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 已提交
257

258 259 260 261 262 263 264 265 266
    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 已提交
267 268 269
        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>.
270

X
xuwei06 已提交
271 272 273
    Return:
        A boolean value that indicates whether a block has fetch operators
        that match the info contained in fetch_targets and fetch_holder_name.
274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294
    """

    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 已提交
295
def _fetch_var(name, scope=None, return_numpy=True):
X
xuwei06 已提交
296
    """
C
chengduoZH 已提交
297 298 299
    Fetch the value of the variable with the given name from the
    given scope.

X
xuwei06 已提交
300
    Args:
301 302 303 304
        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 已提交
305 306 307 308
            If None, global_scope() will be used. Default None.
        return_numpy(bool): whether convert the tensor to numpy.ndarray.
            Default True.

X
xuwei06 已提交
309 310 311 312 313 314
    Returns:
       LodTensor|numpy.ndarray
    """
    assert isinstance(name, str)
    if scope is None:
        scope = global_scope()
S
sneaxiy 已提交
315
    assert isinstance(scope, core._Scope)
X
xuwei06 已提交
316

Y
Yibing Liu 已提交
317
    var = scope.find_var(name)
318 319 320 321
    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 已提交
322 323 324 325 326 327
    tensor = var.get_tensor()
    if return_numpy:
        tensor = as_numpy(tensor)
    return tensor


X
polish  
Xin Pan 已提交
328 329 330 331 332 333 334 335 336
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 已提交
337 338


339 340 341 342
def _get_strong_program_cache_key(program, feed, fetch_list):
    return str(id(program)) + _get_program_cache_key(feed, fetch_list)


X
polish  
Xin Pan 已提交
343 344 345
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 已提交
346 347 348 349

    return str(feed_var_names + fetch_var_names)


W
Wu Yi 已提交
350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380
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


381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401
class FetchHandler(object):
    def __init__(self, fetch_target_names, period_secs=60, return_np=True):
        self.fetch_target_names = fetch_target_names
        self.period_secs = period_secs
        self.return_np = return_np

    def handler(self, fetch_target_vars):
        return

    @staticmethod
    def help():
        print("""
class FetchHandlerExamlpe(FetchHandler):
    def handler(self, fetch_target_vars):
        b_auc = fetch_target_vars[0]
        g_auc = fetch_target_vars[1]
                        
        print("b_auc: {}, g_auc: {} at time: {}".format(b_auc, g_auc, time.ctime()))
""")


Y
Yu Yang 已提交
402
class Executor(object):
403
    """
404 405 406 407 408 409 410 411 412 413 414
    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
    to feed map and fetch_list. Feed map provides input data for the
    program. fetch_list provides the variables(or names) that user wants
    to get after program runs. Note: the executor will run all operators
    in the program but not only the operators dependent by the fetch_list.
    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.
S
Fix doc  
sneaxiy 已提交
415

416
    Examples:
S
Fix doc  
sneaxiy 已提交
417 418
        .. code-block:: python

419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463
          import paddle.fluid as fluid
          import paddle.fluid.compiler as compiler
          import numpy
          import os

          use_cuda = True
          place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
          exe = fluid.Executor(place)

          train_program = fluid.Program()
          startup_program = fluid.Program()
          with fluid.program_guard(train_program, startup_program):
              data = fluid.layers.data(name='X', shape=[1], dtype='float32')
              hidden = fluid.layers.fc(input=data, size=10)
              loss = fluid.layers.mean(hidden)
              fluid.optimizer.SGD(learning_rate=0.01).minimize(loss)

          # Run the startup program once and only once.
          # Not need to optimize/compile the startup program.
          startup_program.random_seed=1
          exe.run(startup_program)

          # Run the main program directly without compile.
          x = numpy.random.random(size=(10, 1)).astype('float32')
          loss_data, = exe.run(train_program,
                               feed={"X": x},
                               fetch_list=[loss.name])

          # Or, compiled the program and run. See `CompiledProgram`
          # for more detail.
          # NOTE: If you use CPU to run the program, you need
          # to specify the CPU_NUM, otherwise, fluid will use
          # all the number of the logic core as the CPU_NUM,
          # in that case, the batch size of the input should be
          # greater than CPU_NUM, if not, the process will be
          # failed by an exception.
          if not use_cuda:
              os.environ['CPU_NUM'] = str(2)

          compiled_prog = compiler.CompiledProgram(
              train_program).with_data_parallel(
              loss_name=loss.name)
          loss_data, = exe.run(compiled_prog,
                               feed={"X": x},
                               fetch_list=[loss.name])
X
add doc  
Xin Pan 已提交
464

465
    Args:
466 467
        place(fluid.CPUPlace|fluid.CUDAPlace(n)): indicate the executor run on which device.

468 469
    """

D
dzhwinter 已提交
470 471
    def __init__(self, place):
        self.place = place
Q
qiaolongfei 已提交
472
        self.program_caches = dict()
473
        self.ctx_caches = dict()
474 475
        self.scope_caches = dict()
        self.var_caches = dict()
476 477 478
        p = core.Place()
        p.set_place(self.place)
        self._default_executor = core.Executor(p)
Y
Yancey1989 已提交
479
        self._closed = False
D
dzhwinter 已提交
480

481 482 483 484 485 486
    def _get_var_cache(self, program_cache_key):
        return self.var_caches.get(program_cache_key, None)

    def _get_scope_cache(self, program_cache_key):
        return self.scope_caches.get(program_cache_key, None)

487 488 489
    def _get_ctx_cache(self, program_cache_key):
        return self.ctx_caches.get(program_cache_key, None)

Q
Qiao Longfei 已提交
490 491 492 493 494 495
    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

496 497 498
    def _add_ctx_cache(self, ctx_cache_key, ctx):
        self.ctx_caches[ctx_cache_key] = ctx

499 500 501 502 503 504
    def _add_scope_cache(self, scope_cache_key, scope):
        self.scope_caches[scope_cache_key] = scope

    def _add_var_cache(self, var_cache_key, var):
        self.var_caches[var_cache_key] = var

Q
Qiao Longfei 已提交
505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530
    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 已提交
531
                global_block._prepend_op(
Q
Qiao Longfei 已提交
532 533 534 535 536 537 538 539
                    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 已提交
540 541 542
                assert isinstance(var, Variable) or isinstance(
                    var, six.string_types), (
                        "Wrong type for fetch_list[%s]: %s" % (i, type(var)))
Q
Qiao Longfei 已提交
543 544 545 546 547 548 549 550 551 552
                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
H
Huihuang Zheng 已提交
553 554
        global_block = program.global_block()
        for op in global_block.ops:
Q
Qiao Longfei 已提交
555 556 557 558
            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 已提交
559
                    cur_feed = _as_lodtensor(cur_feed, self.place)
H
Huihuang Zheng 已提交
560 561
                var = global_block.var(feed_target_name)
                check_feed_shape_type(var, cur_feed)
Q
Qiao Longfei 已提交
562 563 564 565 566 567 568 569
                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 已提交
570
            for i in six.moves.range(len(fetch_list))
Q
Qiao Longfei 已提交
571 572 573
        ]
        return outs

S
Fix doc  
sneaxiy 已提交
574 575 576 577 578 579
    '''
    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 已提交
580 581 582 583
    def close(self):
        """
        Close this executor.

X
fix  
Xin Pan 已提交
584
        You can no longer use this executor after calling this method.
585 586 587 588 589 590 591 592 593 594 595 596
        For the distributed training, this method would free the resource
        on PServers related to the current Trainer.

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid

              cpu = fluid.CPUPlace()
              exe = fluid.Executor(cpu)
              # execute training or testing
              exe.close()
Y
Yancey1989 已提交
597
        """
598 599
        if not self._closed:
            self._default_executor.close()
Y
Yancey1989 已提交
600
            self._closed = True
Y
Yancey1989 已提交
601

X
fix  
Xin Pan 已提交
602
    def _run_parallel(self, program, scope, feed, fetch_list, fetch_var_name,
X
polish  
Xin Pan 已提交
603
                      return_numpy):
604
        exe = program._executor
H
Huihuang Zheng 已提交
605 606 607 608 609
        # TODO(zhenghuihuang): quantization uses Graph in CompiledProgram
        # instead of program. We will add support for checking Vars in Graph
        need_check_feed = program._program is not None
        if need_check_feed:
            global_block = program._program.global_block()
610 611 612 613 614 615
        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()
616
                    # always set to CPU place, since the tensor need to be split
617
                    # it is fast in CPU
618 619 620
                    assert isinstance( feed[feed_name], np.ndarray ), \
                        "The input({}) should be numpy.array, but not {}.".format(
                        feed_name, type(feed[feed_name]))
621
                    feed_tensor.set(feed[feed_name], core.CPUPlace())
H
Huihuang Zheng 已提交
622 623 624
                if need_check_feed:
                    var = global_block.var(feed_name)
                    check_feed_shape_type(var, feed_tensor)
625 626
                feed_tensor_dict[feed_name] = feed_tensor

627
            exe.feed_and_split_tensor_into_local_scopes(feed_tensor_dict)
628
        elif isinstance(feed, list) or isinstance(feed, tuple):
X
fix  
Xin Pan 已提交
629
            if len(feed) != len(program._places):
630 631 632 633 634 635 636 637 638 639 640 641 642 643
                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()
644 645 646
                        assert isinstance(each[feed_name], np.ndarray), \
                            "The input({}) should be numpy.array, but not {}.".format(
                            feed_name, type(each[feed_name]))
X
fix  
Xin Pan 已提交
647
                        tmp.set(tensor, program._places[i])
648
                        tensor = tmp
H
Huihuang Zheng 已提交
649 650 651
                    if need_check_feed:
                        var = global_block.var(feed_name)
                        check_feed_shape_type(var, tensor)
652 653
                    res_dict[feed_name] = tensor
                res.append(res_dict)
654
            exe.feed_tensors_into_local_scopes(res)
655

X
polish  
Xin Pan 已提交
656
        fetch_var_names = list(map(_to_name_str, fetch_list))
657
        tensors = exe.run(fetch_var_names)._move_to_list()
658
        return as_numpy(tensors) if return_numpy else tensors
659

Y
Yu Yang 已提交
660
    def run(self,
Y
Yu Yang 已提交
661
            program=None,
662 663
            feed=None,
            fetch_list=None,
Y
Yu Yang 已提交
664
            feed_var_name='feed',
Y
Yu Yang 已提交
665
            fetch_var_name='fetch',
D
dzhwinter 已提交
666
            scope=None,
667 668
            return_numpy=True,
            use_program_cache=False):
669
        """
670 671 672 673
        Run program by this Executor. Feed data by feed map, fetch result by
        fetch_list. 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
674 675
        the variables(or names) that user want to get after program run.

676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700
        Note: the executor will run all operators in the program but not
        only the operators dependent by the fetch_list.

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              import numpy

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

              data = fluid.layers.data(name='X', shape=[1], dtype='float32')
              hidden = fluid.layers.fc(input=data, size=10)
              loss = fluid.layers.mean(hidden)
              adam = fluid.optimizer.Adam()
              adam.minimize(loss)

              # Run the startup program once and only once.
              exe.run(fluid.default_startup_program())

              x = numpy.random.random(size=(10, 1)).astype('float32')
              outs = exe.run(feed={'X': x},
                             fetch_list=[loss.name])
Q
qiaolongfei 已提交
701

702
        Args:
X
add doc  
Xin Pan 已提交
703
            program(Program|CompiledProgram): the program that need to run,
X
fix  
Xin Pan 已提交
704
                if not provided, then default_main_program (not compiled) will be used.
X
add doc  
Xin Pan 已提交
705
            feed(dict): feed variable map, e.g. {"image": ImageData, "label": LabelData}
Z
Zeng Jinle 已提交
706 707 708 709 710 711 712 713
            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
714
            return_numpy(bool): if convert the fetched tensor to numpy
Z
Zeng Jinle 已提交
715 716 717 718 719 720
            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. 
                
721 722 723
        Returns:

            list(numpy.array): fetch result according to fetch_list.
724
        """
C
chengduo 已提交
725 726 727 728 729 730 731 732 733 734 735 736
        try:
            return self._run_impl(
                program=program,
                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)
        except Exception as e:
            if not isinstance(e, core.EOFException):
737 738
                warnings.warn(
                    "The following exception is not an EOF exception.")
739
            six.reraise(*sys.exc_info())
C
chengduo 已提交
740 741 742

    def _run_impl(self, program, feed, fetch_list, feed_var_name,
                  fetch_var_name, scope, return_numpy, use_program_cache):
Y
Yancey1989 已提交
743 744 745
        if self._closed:
            raise RuntimeError("Attempted to use a closed Executor")

C
chengduo 已提交
746
        use_default_main_program = program is None
747 748
        if program is None:
            program = default_main_program()
C
chengduo 已提交
749
        if isinstance(program, Program) and \
750
                        len(program.global_block().ops) == 0:
C
chengduo 已提交
751 752 753 754
            error_info = "The current program is empty."
            if use_default_main_program:
                error_info += " Maybe you should pass the Program or the CompiledProgram manually."
            warnings.warn(error_info)
755

756 757
        if scope is None:
            scope = global_scope()
758 759 760 761 762 763 764 765 766

        if fetch_list is not None:
            if isinstance(fetch_list, Variable) or isinstance(fetch_list, str):
                fetch_list = [fetch_list]
            assert isinstance(fetch_list, tuple) or isinstance(fetch_list, list), \
                "Currently , The fetch_list type only should be list or tuple, \n"\
                "but the input type is {}. For more information please refer to \n"\
                "the executor.run(...).".format(type(fetch_list))
        else:
X
polish  
Xin Pan 已提交
767
            fetch_list = []
768

X
polish  
Xin Pan 已提交
769
        compiled = isinstance(program, compiler.CompiledProgram)
H
Huihuang Zheng 已提交
770

X
polish  
Xin Pan 已提交
771
        # For backward compatibility, run directly.
772
        if not compiled:
C
chengduo 已提交
773
            return self._run_program(
774 775 776 777 778 779 780 781 782 783
                program,
                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)
C
chengduo 已提交
784 785 786
        if program._is_inference:
            return self._run_inference(program._executor, feed)
        else:
787
            return self._run_parallel(
X
fix  
Xin Pan 已提交
788
                program,
789 790 791
                scope=scope,
                feed=feed,
                fetch_list=fetch_list,
X
polish  
Xin Pan 已提交
792
                fetch_var_name=fetch_var_name,
793 794
                return_numpy=return_numpy)

C
chengduo 已提交
795
    def _run_program(self, program, feed, fetch_list, feed_var_name,
C
chengduo 已提交
796
                     fetch_var_name, scope, return_numpy, use_program_cache):
797

798 799
        if feed is None:
            feed = {}
S
sneaxiy 已提交
800 801 802 803
        elif isinstance(feed, (list, tuple)):
            assert len(feed) == 1, "Not compiled with data parallel"
            feed = feed[0]

Q
qiaolongfei 已提交
804
        if not isinstance(feed, dict):
D
dzhwinter 已提交
805 806 807
            raise TypeError(
                "feed requires dict as its Parameter. But you passed in %s" %
                (type(feed)))
Y
Yu Yang 已提交
808

809
        assert program is not None, "The program should not be Empty"
Y
Yu Yang 已提交
810
        if not isinstance(program, Program):
D
dzhwinter 已提交
811 812 813
            raise TypeError(
                "Executor requires Program as its Parameter. But you passed in %s"
                % (type(program)))
Y
Yu Yang 已提交
814

815
        if use_program_cache:
816
            cache_key = _get_strong_program_cache_key(program, feed, fetch_list)
Q
Qiao Longfei 已提交
817
            cached_program = self._get_program_cache(cache_key)
818
            cached_ctx = self._get_ctx_cache(cache_key)
819 820
            cached_scope = self._get_scope_cache(cache_key)
            cached_var = self._get_var_cache(cache_key)
Q
Qiao Longfei 已提交
821 822 823 824 825 826 827 828
            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)
829
                fetch_list_str = list(map(_to_name_str, fetch_list))
830
                cached_ctx = self._default_executor.prepare_ctx_cache(
831 832 833 834 835 836 837 838 839
                    cached_program.desc, 0, fetch_list_str, False)
                cached_var = self._default_executor.create_variables(
                    cached_program.desc, scope, 0)
                # currently, we cache program, vars, sub_scope here
                # we suppose that in a life cycle of training, a user
                # will not create many programs. So, here the basic
                # rule of caching is to cache all unseen (program, var, scope)
                # when a user use use_program_cache.
                cached_scope = scope.new_scope()
840
                self._add_ctx_cache(cache_key, cached_ctx)
841 842
                self._add_var_cache(cache_key, cached_var)
                self._add_scope_cache(cache_key, cached_scope)
Q
Qiao Longfei 已提交
843
            program = cached_program
844
            ctx = cached_ctx
845 846
            scope = cached_scope
            var = cached_var
847
        else:
Q
Qiao Longfei 已提交
848 849 850 851 852 853 854 855
            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)
856
        if not use_program_cache:
C
chengduo 已提交
857 858
            self._default_executor.run(program.desc, scope, 0, True, True,
                                       fetch_var_name)
859
        else:
C
chengduo 已提交
860 861
            self._default_executor.run_cached_prepared_ctx(ctx, scope, False,
                                                           False, False)
862 863
        arr = scope.find_var(fetch_var_name).get_lod_tensor_array()
        tensors = arr._move_to_list()
D
dzhwinter 已提交
864
        if return_numpy:
865 866 867
            return as_numpy(tensors)
        else:
            return tensors
F
flame 已提交
868

X
Xin Pan 已提交
869 870
    def _run_inference(self, exe, feed):
        return exe.run(feed)
D
dongdaxiang 已提交
871

872 873
    def _dump_debug_info(self, program=None, trainer=None):
        with open(str(id(program)) + "_train_desc.prototxt", "w") as fout:
H
hutuxian 已提交
874
            fout.write(str(trainer))
875 876 877 878
        if program._fleet_opt:
            with open("fleet_desc.prototxt", "w") as fout:
                fout.write(str(program._fleet_opt["fleet_desc"]))

879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894
    def _adjust_pipeline_resource(self, pipeline_opt, dataset, pipeline_num):
        filelist_length = len(dataset.dataset.get_filelist())
        if filelist_length < pipeline_num:
            pipeline_num = filelist_length
            print(
                "Pipeline training: setting the pipeline num to %d is enough because there are only %d files"
                % (filelist_length, filelist_length))
        if filelist_length < pipeline_num * pipeline_opt["concurrency_list"][0]:
            print(
                "Pipeline training: setting the 1st element in concurrency_list to %d is enough because there are only %d files"
                % (filelist_length // pipeline_num, filelist_length))
            pipeline_opt["concurrency_list"][
                0] = filelist_length // pipeline_num
        dataset.set_thread(pipeline_opt["concurrency_list"][0] * pipeline_num)
        return pipeline_num

895 896 897 898 899 900 901 902 903
    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 已提交
904 905 906 907
        if scope is None:
            scope = global_scope()
        if fetch_list is None:
            fetch_list = []
D
dongdaxiang 已提交
908 909 910
        if fetch_info is None:
            fetch_info = []
        assert len(fetch_list) == len(fetch_info)
D
dongdaxiang 已提交
911 912
        compiled = isinstance(program, compiler.CompiledProgram)
        if not compiled:
H
hutuxian 已提交
913 914 915 916 917 918
            # TODO: Need a better way to distinguish and specify different execution mode
            if program._pipeline_opt:
                trainer = TrainerFactory()._create_trainer(
                    program._pipeline_opt)
            else:
                trainer = TrainerFactory()._create_trainer(program._fleet_opt)
919
            trainer._set_program(program)
920
        else:
H
hutuxian 已提交
921 922 923 924 925 926
            if program._pipeline_opt:
                trainer = TrainerFactory()._create_trainer(
                    program.program._pipeline_opt)
            else:
                trainer = TrainerFactory()._create_trainer(
                    program.program._fleet_opt)
927
            trainer._set_program(program.program)
H
hutuxian 已提交
928

929
        if thread <= 0:
D
dongdaxiang 已提交
930 931
            if dataset.thread_num <= 0:
                raise RuntimeError(
932 933
                    "You should set thread num first, either in Dataset"
                    "or in Executor.train_from_dataset")
D
dongdaxiang 已提交
934
            else:
935
                trainer._set_thread(dataset.thread_num)
936
        else:
937
            trainer._set_thread(thread)
H
hutuxian 已提交
938

939 940
        trainer._set_debug(debug)
        trainer._set_fetch_var_and_info(fetch_list, fetch_info, print_period)
941
        return scope, trainer
942

943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003
    def _run_from_dataset(self,
                          program=None,
                          dataset=None,
                          scope=None,
                          thread=0,
                          is_infer=False,
                          debug=False,
                          fetch_list=None,
                          fetch_info=None,
                          print_period=100,
                          fetch_handler=None):
        if dataset is None:
            raise RuntimeError("dataset is need and should be initialized")

        if program._pipeline_opt:
            thread = self._adjust_pipeline_resource(program._pipeline_opt,
                                                    dataset, thread)

        dataset._prepare_to_run()

        if fetch_handler is not None:
            fetch_instance = fetch_handler
        elif fetch_handler is None and fetch_list is not None:

            class FH(FetchHandler):
                def handler(self, fetch_target_vars):
                    for i in range(len(fetch_target_vars)):
                        print("{}: \n {}\n".format(fetch_info[i],
                                                   fetch_target_vars[i]))

            fetch_target_names = [var.name for var in fetch_list]
            fetch_instance = FH(fetch_target_names,
                                period_secs=print_period,
                                return_np=False)
        else:
            fetch_instance = FetchHandler([])

        scope, trainer = self._prepare_trainer(
            program=program,
            dataset=dataset,
            scope=scope,
            thread=thread,
            debug=debug)

        trainer._set_infer(is_infer)
        trainer._gen_trainer_desc()

        self._dump_debug_info(program=program, trainer=trainer)

        trainer_instance = self._default_executor.init_for_dataset(
            program.desc, trainer._desc(), scope, dataset.dataset)

        scope0 = trainer_instance.get_worker_scope(0)

        fetch_monitor = FetchHandlerMonitor(scope0, fetch_instance)
        fetch_monitor.start()
        self._default_executor.run_from_dataset(trainer_instance)
        fetch_monitor.stop()
        dataset._finish_to_run()
        return None

1004 1005 1006 1007 1008
    def infer_from_dataset(self,
                           program=None,
                           dataset=None,
                           scope=None,
                           thread=0,
1009 1010 1011
                           debug=False,
                           fetch_list=None,
                           fetch_info=None,
1012 1013
                           print_period=100,
                           fetch_handler=None):
1014 1015 1016 1017 1018 1019
        """
        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.
1020

1021 1022 1023 1024 1025
        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.
1026
               Please check the document of Dataset if needed. default is None
1027 1028 1029
            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
1030 1031
               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
1032
            fetch_list(Variable List): fetch variable list, each variable
1033 1034 1035
                                       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
1036
            fetch_handler(FetchHandler): a user define class for fetch output.
1037

1038 1039 1040 1041
        Returns:
            None

        Examples:
1042 1043

            .. code-block:: python
1044

1045
                import paddle.fluid as fluid
1046 1047

                place = fluid.CPUPlace() # you can set place = fluid.CUDAPlace(0) to use gpu
1048
                exe = fluid.Executor(place)
1049 1050
                x = fluid.layers.data(name="x", shape=[10, 10], dtype="int64")
                y = fluid.layers.data(name="y", shape=[1], dtype="int64", lod_level=1)
1051 1052
                dataset = fluid.DatasetFactory().create_dataset()
                dataset.set_use_var([x, y])
1053 1054
                dataset.set_thread(1)
                filelist = [] # you should set your own filelist, e.g. filelist = ["dataA.txt"]
1055 1056 1057 1058
                dataset.set_filelist(filelist)
                exe.run(fluid.default_startup_program())
                exe.infer_from_dataset(program=fluid.default_main_program(),
                                       dataset=dataset)        
1059

1060
        """
1061 1062 1063
        return self._run_from_dataset(program, dataset, scope, thread, True,
                                      debug, fetch_list, fetch_info,
                                      print_period, fetch_handler)
1064 1065 1066 1067 1068 1069 1070 1071 1072

    def train_from_dataset(self,
                           program=None,
                           dataset=None,
                           scope=None,
                           thread=0,
                           debug=False,
                           fetch_list=None,
                           fetch_info=None,
1073 1074
                           print_period=100,
                           fetch_handler=None):
1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100
        """
        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
1101
            fetch_handler(FetchHandler): a user define class for fetch output.
1102 1103 1104

        Returns:
            None
1105
        
1106
        Examples:
1107
        
1108 1109 1110
            .. code-block:: python

              import paddle.fluid as fluid
1111 1112

              place = fluid.CPUPlace() # you can set place = fluid.CUDAPlace(0) to use gpu
1113
              exe = fluid.Executor(place)
1114 1115
              x = fluid.layers.data(name="x", shape=[10, 10], dtype="int64")
              y = fluid.layers.data(name="y", shape=[1], dtype="int64", lod_level=1)
1116 1117
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_use_var([x, y])
1118 1119
              dataset.set_thread(1)
              filelist = [] # you should set your own filelist, e.g. filelist = ["dataA.txt"]
1120 1121 1122 1123
              dataset.set_filelist(filelist)
              exe.run(fluid.default_startup_program())
              exe.train_from_dataset(program=fluid.default_main_program(),
                                     dataset=dataset)
1124 1125

        """
1126 1127 1128
        return self._run_from_dataset(program, dataset, scope, thread, False,
                                      debug, fetch_list, fetch_info,
                                      print_period, fetch_handler)