executor.py 4.6 KB
Newer Older
D
dzhwinter 已提交
1
import numpy as np
2
from . import core
Y
Yu Yang 已提交
3
from framework import Program, default_main_program
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 49
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)

        self.executor = core.Executor(act_places)
D
dzhwinter 已提交
50 51 52 53 54 55 56 57 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
        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 已提交
91 92

    def run(self,
Y
Yu Yang 已提交
93
            program=None,
94 95
            feed=None,
            fetch_list=None,
Y
Yu Yang 已提交
96
            feed_var_name='feed',
Y
Yu Yang 已提交
97
            fetch_var_name='fetch',
D
dzhwinter 已提交
98 99
            scope=None,
            return_numpy=True):
100 101 102 103 104
        if feed is None:
            feed = {}
        if fetch_list is None:
            fetch_list = []

Y
Yu Yang 已提交
105
        if program is None:
Y
Yu Yang 已提交
106
            program = default_main_program()
Y
Yu Yang 已提交
107

Y
Yu Yang 已提交
108 109 110
        if not isinstance(program, Program):
            raise TypeError()

Y
Yu Yang 已提交
111 112 113
        if scope is None:
            scope = g_scope

Y
Yu Yang 已提交
114 115 116 117 118 119 120 121 122 123 124 125 126 127
        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 已提交
128 129 130 131
            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 已提交
132 133 134 135 136 137 138 139 140 141 142 143

        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})

Y
Yu Yang 已提交
144
        self.executor.run(program.desc, scope, 0, True)
D
dzhwinter 已提交
145
        outs = [
Y
Yu Yang 已提交
146
            core.get_fetch_variable(scope, fetch_var_name, i)
Y
Yu Yang 已提交
147 148
            for i in xrange(len(fetch_list))
        ]
D
dzhwinter 已提交
149 150 151 152

        if return_numpy:
            outs = as_numpy(outs)
        return outs