test_parallel_op.py 5.8 KB
Newer Older
D
dzhwinter 已提交
1
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserve.
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.

Y
Yang Yang 已提交
15
import unittest
16

Y
Yang Yang 已提交
17
import paddle.v2.fluid as fluid
Y
Yang Yu 已提交
18 19 20 21
import numpy


class BaseParallelForTest(unittest.TestCase):
Y
Yang Yu 已提交
22 23 24 25 26 27 28 29 30
    def run_test(self, callback, feed, fetch):
        """
        Run the unittest for parallel.for
        Args:
            callback(callable): A callable function returns a generator. There 
                are two yields in the generator function. The first yield 
                returns the data layers, and the second yield returns the loss. 
                The modified data variables will be sent back during the first 
                yield.
31

Y
Yang Yu 已提交
32 33 34 35 36
            feed(dict): The executor feeding dictionary.
            fetch(list|basestr): The fetch name lists. 

        Returns:
            None
37

Y
Yang Yu 已提交
38 39 40 41 42
        Raises:
            AssertionError when the computation of cpu, parallel.for in cpu, 
                gpu, parallel.for in gpu are different.

        """
Y
Yang Yu 已提交
43
        cpu = fluid.CPUPlace()
Y
Yang Yu 已提交
44
        result_cpu = self._run_test_impl_(
Y
Yang Yu 已提交
45 46 47 48 49
            callback=callback,
            feed=feed,
            fetch=fetch,
            place=cpu,
            use_parallel=False)
Y
Yang Yu 已提交
50
        result_cpu_parallel = self._run_test_impl_(
51 52 53 54 55 56 57
            callback=callback,
            feed=feed,
            fetch=fetch,
            place=cpu,
            use_parallel=True)
        if fluid.core.is_compile_gpu():
            gpu = fluid.CUDAPlace(0)
Y
Yang Yu 已提交
58
            result_gpu = self._run_test_impl_(
59 60 61 62 63
                callback=callback,
                feed=feed,
                fetch=fetch,
                place=gpu,
                use_parallel=False)
Y
Yang Yu 已提交
64
            result_gpu_parallel = self._run_test_impl_(
65 66 67 68 69 70 71 72 73
                callback=callback,
                feed=feed,
                fetch=fetch,
                place=gpu,
                use_parallel=True)
            self._assert_same_(fetch, result_cpu, result_cpu_parallel,
                               result_gpu, result_gpu_parallel)
        else:
            self._assert_same_(fetch, result_cpu, result_cpu_parallel)
Y
Yang Yu 已提交
74

Y
Yang Yu 已提交
75 76 77 78 79 80 81 82 83 84 85
    def _run_test_impl_(self, callback, feed, fetch, place, use_parallel=False):
        """
        Run a single test, returns the fetch values
        Args:
            place(Place): the computation place. 
            use_parallel(bool): Whether use parallel.for or not. 

        Returns:
            Fetched numpy arrays.

        """
86 87
        if isinstance(fetch, basestring):
            fetch = [fetch]
Y
Yang Yu 已提交
88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
        main = fluid.Program()
        startup = fluid.Program()
        # Fix seed
        main.random_seed = 10
        startup.random_seed = 10

        with fluid.program_guard(main, startup):
            generator = callback()
            # Automatically insert parallel do if use_parallel = True
            if use_parallel:
                places = fluid.layers.get_places()
                pd = fluid.layers.ParallelDo(places)
                data = next(generator)

                if isinstance(data, fluid.Variable):
                    data = [data]
104

Y
Yang Yu 已提交
105 106 107 108
                with pd.do():
                    ins = map(pd.read_input, data)
                    if len(ins) == 1:
                        ins = ins[0]
109
                    loss = generator.send(ins)  # patch input
Y
Yang Yu 已提交
110 111 112 113 114
                    pd.write_output(loss)

                loss = pd()
            else:
                data = next(generator)
115 116
                loss = generator.send(data)
            self.assertIsNotNone(loss)
Y
Yang Yu 已提交
117 118 119 120 121 122 123
            avg_loss = fluid.layers.mean(x=loss)
            fluid.backward.append_backward(loss=avg_loss)

        exe = fluid.Executor(place)
        exe.run(startup)
        return exe.run(main, feed=feed, fetch_list=fetch)

124
    def _assert_same_(self, fetch, *args):
Y
Yang Yu 已提交
125 126 127 128 129 130 131 132 133 134 135 136 137 138
        """
        Assert the return values of `run_test` are same.
        Args:
            fetch: Fetch list. Used for print error message
            *args: The fetch result lists of each situations.

        Returns:
            None
            
        Raises:
            AssertionError

        """

139 140 141 142 143 144 145 146 147 148 149
        def _impl_(a, b, fetch_id, item_id):
            item_str = ['CPU', 'ParallelCPU', 'GPU', 'ParallelGPU']
            flag = numpy.allclose(a, b, rtol=0.1)
            self.assertTrue(flag, "The {0} are different in {1}".format(
                fetch[fetch_id], item_str[item_id]))

        for i, items in enumerate(zip(*args)):
            self.assertGreater(len(items), 0)
            for j in range(1, len(items)):
                _impl_(items[0], items[j], fetch_id=i, item_id=j)

Y
Yang Yu 已提交
150 151

class ParallelOpTest(BaseParallelForTest):
152 153 154 155 156 157 158
    @staticmethod
    def __network__():
        x = fluid.layers.data(shape=[784], dtype='float32', name='img')
        x = yield x
        hidden = fluid.layers.fc(input=x, size=200, param_attr='fc1.w')
        loss = fluid.layers.mean(x=hidden)
        yield loss
Y
Yang Yu 已提交
159

160
    def test_simple_fc(self):
Y
Yang Yu 已提交
161
        self.run_test(
162
            callback=ParallelOpTest.__network__,
Y
Yang Yu 已提交
163
            feed={
164
                'img': numpy.random.random(size=(51, 784)).astype('float32')
Y
Yang Yu 已提交
165
            },
Y
Yang Yang 已提交
166
            fetch=['fc1.w@GRAD'])
Y
Yang Yang 已提交
167

168 169 170 171
    def test_fc_with_tiny_data(self):
        self.run_test(
            callback=ParallelOpTest.__network__,
            feed={'img': numpy.random.random(size=(1, 784)).astype('float32')},
Y
Yang Yang 已提交
172
            fetch=['fc1.w@GRAD'])
173

Y
Yang Yang 已提交
174 175 176

if __name__ == '__main__':
    unittest.main()
新手
引导
客服 返回
顶部