executor.py 5.0 KB
Newer Older
D
dzhwinter 已提交
1
import numpy as np
2
from . import core
T
typhoonzero 已提交
3
from framework import Program, default_main_program, Parameter, Variable
4 5

__all__ = ['Executor', 'g_scope']
Y
Yu Yang 已提交
6

Y
Yu Yang 已提交
7 8
g_scope = core.Scope()

Y
Yu Yang 已提交
9

D
dzhwinter 已提交
10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
def as_numpy(tensor):
    if isinstance(tensor, list):
        return [as_numpy(t) for t in tensor]
    assert isinstance(tensor, core.LoDTensor)
    lod = tensor.lod()
    tensor_data = np.array(tensor)
    if len(lod) == 0:
        ans = tensor_data
    else:
        raise RuntimeError("LoD Calculate lacks unit tests and buggy")
    # elif len(lod) == 1:
    #     ans = []
    #     idx = 0
    #     while idx < len(lod) - 1:
    #         ans.append(tensor_data[lod[idx]:lod[idx + 1]])
    #         idx += 1
    # else:
    #     for l in reversed(lod):
    #         ans = []
    #         idx = 0
    #         while idx < len(l) - 1:
    #             ans.append(tensor_data[l[idx]:l[idx + 1]])
    #             idx += 1
    #         tensor_data = ans
    #     ans = tensor_data
    return ans


Y
Yu Yang 已提交
38 39 40 41 42 43 44 45 46 47 48
class Executor(object):
    def __init__(self, places):
        if not isinstance(places, list) and not isinstance(places, tuple):
            places = [places]

        act_places = []
        for each in places:
            p = core.Place()
            p.set_place(each)
            act_places.append(p)

D
dzhwinter 已提交
49
        # TODO(dzhwinter) : consider that our fluid tests all written in 
D
dzhwinter 已提交
50
        # CUDAPlace(gpu_id), this will be changed in the future
D
dzhwinter 已提交
51 52 53 54 55
        if core.is_compile_gpu():
            core.init_devices(["CPU", "GPU:0"])
        else:
            core.init_devices(["CPU"])

D
dzhwinter 已提交
56 57
        # TODO(dzhwinter) : only use the first place
        self.executor = core.Executor(act_places[0])
D
dzhwinter 已提交
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
        self.places = places

    def aslodtensor(self, data):
        def accumulate(data):
            if not isinstance(data, list):
                return 1
            return sum([accumulate(sub) for sub in data])

        def parselod(data):
            seq_lens = [accumulate(seq) for seq in data]
            cur_len = 0
            lod = [cur_len]
            for l in seq_lens:
                cur_len += l
                lod.append(cur_len)
            return lod

        assert len(self.places) != 0
        if not isinstance(data, list):
            # pure tensor case
            tensor = core.LoDTensor()
            tensor.set(data, self.places[0])
            return tensor
        else:
            raise RuntimeError("Current implementation lacks unittests")
            # lodtensor case
            lod = []
            if not isinstance(data[0], list):
                lod.append(parselod(data))
                flattened_data = np.concatenate(data, axis=0).astype("int64")
            else:
                while isinstance(data[0], list):
                    lod.append(parselod(seq))
                    flattened_data = [item for seq in data for item in seq]
                    data = flattened_data
                flattened_data = np.concatenate(data, axis=0).astype("int64")
            flattened_data = flattened_data.reshape([len(flattened_data), 1])
            tensor = core.LoDTensor()
            tensor.set(flattened_data, self.places[0])
            tensor.set_lod(lod)
            return tensor
Y
Yu Yang 已提交
99 100

    def run(self,
Y
Yu Yang 已提交
101
            program=None,
102 103
            feed=None,
            fetch_list=None,
Y
Yu Yang 已提交
104
            feed_var_name='feed',
Y
Yu Yang 已提交
105
            fetch_var_name='fetch',
D
dzhwinter 已提交
106 107
            scope=None,
            return_numpy=True):
108 109 110 111 112
        if feed is None:
            feed = {}
        if fetch_list is None:
            fetch_list = []

Y
Yu Yang 已提交
113
        if program is None:
Y
Yu Yang 已提交
114
            program = default_main_program()
Y
Yu Yang 已提交
115

Y
Yu Yang 已提交
116 117 118
        if not isinstance(program, Program):
            raise TypeError()

Y
Yu Yang 已提交
119 120 121
        if scope is None:
            scope = g_scope

Y
Yu Yang 已提交
122 123 124 125 126 127 128 129 130 131 132 133 134 135
        program = program.clone()
        global_block = program.global_block()
        feed_var = global_block.create_var(
            name=feed_var_name,
            type=core.VarDesc.VarType.FEED_MINIBATCH,
            persistable=True)

        for i, name in enumerate(feed):
            out = global_block.var(name)
            global_block.prepend_op(
                'feed',
                inputs={'X': [feed_var]},
                outputs={'Out': [out]},
                attrs={'col': i})
D
dzhwinter 已提交
136 137 138 139
            cur_feed = feed[name]
            if not isinstance(cur_feed, core.LoDTensor):
                cur_feed = self.aslodtensor(cur_feed)
            core.set_feed_variable(scope, cur_feed, feed_var.name, i)
Y
Yu Yang 已提交
140 141 142 143 144 145 146 147 148 149 150 151

        fetch_var = global_block.create_var(
            name=fetch_var_name,
            type=core.VarDesc.VarType.FETCH_LIST,
            persistable=True)
        for i, var in enumerate(fetch_list):
            global_block.append_op(
                type='fetch',
                inputs={'X': [var]},
                outputs={'Out': [fetch_var]},
                attrs={'col': i})

T
typhoonzero 已提交
152
        self.executor.run(program.desc, scope, 0, True, True)
D
dzhwinter 已提交
153
        outs = [
Y
Yu Yang 已提交
154
            core.get_fetch_variable(scope, fetch_var_name, i)
Y
Yu Yang 已提交
155 156
            for i in xrange(len(fetch_list))
        ]
D
dzhwinter 已提交
157 158 159 160

        if return_numpy:
            outs = as_numpy(outs)
        return outs