test_boxps.py

#   Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
#
# 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
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# 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.

import paddle
import paddle.fluid as fluid
import paddle.fluid.core as core
import unittest
from paddle.fluid.layers.nn import _pull_box_sparse
from paddle.fluid.transpiler import collective


class TestTranspile(unittest.TestCase):
    """TestCases for BoxPS Preload"""

    def get_transpile(self, mode, trainers="127.0.0.1:6174"):
        config = fluid.DistributeTranspilerConfig()
        config.mode = 'collective'
        config.collective_mode = mode
        t = fluid.DistributeTranspiler(config=config)
        return t

    def test_transpile(self):
        main_program = fluid.Program()
        startup_program = fluid.Program()
        t = self.get_transpile("single_process_multi_thread")
        t.transpile(
            trainer_id=0,
            startup_program=startup_program,
            trainers="127.0.0.1:6174",
            program=main_program,
        )
        t = self.get_transpile("grad_allreduce")
        try:
            t.transpile(
                trainer_id=0,
                startup_program=startup_program,
                trainers="127.0.0.1:6174",
                program=main_program,
            )
        except ValueError as e:
            print(e)

    def test_single_trainers(self):
        transpiler = collective.GradAllReduce(0)
        try:
            transpiler.transpile(
                startup_program=fluid.Program(),
                main_program=fluid.Program(),
                rank=1,
                endpoints="127.0.0.1:6174",
                current_endpoint="127.0.0.1:6174",
                wait_port="6174",
            )
        except ValueError as e:
            print(e)
        transpiler = collective.LocalSGD(0)
        try:
            transpiler.transpile(
                startup_program=fluid.Program(),
                main_program=fluid.Program(),
                rank=1,
                endpoints="127.0.0.1:6174",
                current_endpoint="127.0.0.1:6174",
                wait_port="6174",
            )
        except ValueError as e:
            print(e)


class TestRunCmd(unittest.TestCase):
    """TestCases for run_cmd"""

    def test_run_cmd(self):
        ret1 = int(core.run_cmd("ls; echo $?").strip().split('\n')[-1])
        ret2 = int(core.run_cmd("ls; echo $?", -1, -1).strip().split('\n')[-1])
        self.assertTrue(ret1 == 0)
        self.assertTrue(ret2 == 0)


class TestPullBoxSparseOP(unittest.TestCase):
    """TestCases for _pull_box_sparse op"""

    def test_pull_box_sparse_op(self):
        paddle.enable_static()
        program = fluid.Program()
        with fluid.program_guard(program):
            x = fluid.layers.data(
                name='x', shape=[1], dtype='int64', lod_level=0
            )
            y = fluid.layers.data(
                name='y', shape=[1], dtype='int64', lod_level=0
            )
            emb_x, emb_y = _pull_box_sparse([x, y], size=1)


if __name__ == '__main__':
    unittest.main()