diff --git a/paddle/fluid/operators/delete_var_op.cc b/paddle/fluid/operators/delete_var_op.cc index d7a9bfbc437dbf4c723b9c87ff62ec6b62c38638..89416f7ab5d07ddac5b540b9bb361f831c1ef360 100644 --- a/paddle/fluid/operators/delete_var_op.cc +++ b/paddle/fluid/operators/delete_var_op.cc @@ -32,6 +32,11 @@ class DeleteVarOp : public framework::OperatorBase { } }; +class DeleteVarOpShapeInference : public framework::InferShapeBase { + public: + void operator()(framework::InferShapeContext *ctx) const override {} +}; + class DeleteVarOpInfoMaker : public framework::OpProtoAndCheckerMaker { public: void Make() override { @@ -48,4 +53,5 @@ It should not be configured by users directly. REGISTER_OPERATOR(delete_var, paddle::operators::DeleteVarOp, paddle::framework::EmptyGradOpMaker, - paddle::operators::DeleteVarOpInfoMaker); + paddle::operators::DeleteVarOpInfoMaker, + paddle::operators::DeleteVarOpShapeInference); diff --git a/python/paddle/fluid/io.py b/python/paddle/fluid/io.py index 604f3eacd75beff306915b224b30c369dd3a486f..22c60c1cbe4faa8577fa655766e42694652e498d 100644 --- a/python/paddle/fluid/io.py +++ b/python/paddle/fluid/io.py @@ -884,12 +884,13 @@ def _load_slice_up_vars(executor, dirname, slice_vars_and_attrs): load_prog = Program() load_block = load_prog.global_block() + need_delete_vars = [] for var_tuple in slice_vars_and_attrs: orig_var = var_tuple[0] start = var_tuple[1] slice_var = var_tuple[2] - end = start + reduce(lambda x, y: x * y, slice_var.shape) + end = start + slice_var.shape[0] clone_orig_var = load_block.create_var( name=orig_var.name, @@ -917,5 +918,8 @@ def _load_slice_up_vars(executor, dirname, slice_vars_and_attrs): attrs={'axes': [0], 'starts': [start], 'ends': [end]}) - + need_delete_vars.append(clone_orig_var) + load_block.append_op( + type='delete_var', + inputs={'X': need_delete_vars}, ) executor.run(load_prog) diff --git a/python/paddle/fluid/tests/unittests/dist_save_load.py b/python/paddle/fluid/tests/unittests/dist_save_load.py new file mode 100644 index 0000000000000000000000000000000000000000..edc60550058f53da456c21de4b41142b907743df --- /dev/null +++ b/python/paddle/fluid/tests/unittests/dist_save_load.py @@ -0,0 +1,174 @@ +# Copyright (c) 2018 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. + +from __future__ import print_function + +import os +import sys +import signal +import subprocess +import argparse +import time +import math +import random +from multiprocessing import Process +from functools import reduce + +import numpy as np +import unittest +import six + +import paddle +import paddle.fluid as fluid +from paddle.fluid import core +from paddle.fluid import io + +from test_dist_base import TestDistRunnerBase, runtime_main, RUN_STEP +from dist_simnet_bow import TestDistSimnetBow2x2, DATA_URL, DATA_MD5 + + +class TestDistSaveLoad2x2(TestDistSimnetBow2x2): + def _load_persistable_vars(self, executor, dirname, program): + def _is_checkpoint_var(var): + """ + the checkpoint will not save or load all the variables. + var type is FEED_MINIBATCH/FETCH_LIST/RAW or var name ends with @GRAD are discarded. + + : param var(Variable) + """ + if var.desc.type() == core.VarDesc.VarType.FEED_MINIBATCH or \ + var.desc.type() == core.VarDesc.VarType.FETCH_LIST or \ + var.desc.type() == core.VarDesc.VarType.RAW: + return False + # @GRAD are named for gradient variables, checkpoint will not save it. + if "@GRAD" in var.name: + return False + # .trainer_ are named for distribute train variables, checkpoint will not save it. + if ".trainer_" in var.name: + return False + + # .block is named for distribute train variables, checkpoint will not save it. + if ".block" in var.name: + return False + + if "tmp_" in var.name: + return False + + return var.persistable + + io.load_vars( + executor, + dirname=dirname, + main_program=program, + predicate=_is_checkpoint_var, + filename=None) + + def run_pserver(self, args): + self.get_model(batch_size=2) + # NOTE: pserver should not call memory optimize + t = self.get_transpiler(args.trainer_id, + fluid.default_main_program(), args.endpoints, + args.trainers, args.sync_mode) + pserver_prog = t.get_pserver_program(args.current_endpoint) + startup_prog = t.get_startup_program(args.current_endpoint, + pserver_prog) + + need_load = bool(int(os.getenv("LOAD", "0"))) + model_dir = os.getenv("MODEL_DIR", "") + + place = fluid.CPUPlace() + exe = fluid.Executor(place) + exe.run(startup_prog) + + if need_load and model_dir: + self._load_persistable_vars(exe, model_dir, startup_prog) + exe.run(pserver_prog) + + def run_trainer(self, args): + test_program, avg_cost, train_reader, test_reader, batch_acc, predict = \ + self.get_model(batch_size=2) + + if args.mem_opt: + fluid.memory_optimize(fluid.default_main_program(), skip_grads=True) + if args.is_dist: + t = self.get_transpiler(args.trainer_id, + fluid.default_main_program(), + args.endpoints, args.trainers, + args.sync_mode) + + trainer_prog = t.get_trainer_program() + else: + trainer_prog = fluid.default_main_program() + + if args.use_cuda: + place = fluid.CUDAPlace(0) + else: + place = fluid.CPUPlace() + + startup_exe = fluid.Executor(place) + startup_exe.run(fluid.default_startup_program()) + + strategy = fluid.ExecutionStrategy() + strategy.num_threads = 1 + strategy.allow_op_delay = False + + build_stra = fluid.BuildStrategy() + + if args.use_reduce: + build_stra.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce + else: + build_stra.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce + + exe = fluid.ParallelExecutor( + args.use_cuda, + loss_name=avg_cost.name, + exec_strategy=strategy, + build_strategy=build_stra) + + feed_var_list = [ + var for var in trainer_prog.global_block().vars.values() + if var.is_data + ] + + feeder = fluid.DataFeeder(feed_var_list, place) + reader_generator = train_reader() + + def get_data(): + origin_batch = next(reader_generator) + if args.is_dist and args.use_reader_alloc: + new_batch = [] + for offset, item in enumerate(origin_batch): + if offset % 2 == args.trainer_id: + new_batch.append(item) + return new_batch + else: + return origin_batch + + need_save = bool(int(os.getenv("SAVE", "0"))) + model_dir = os.getenv("MODEL_DIR", "") + + if need_save: + for _ in six.moves.xrange(RUN_STEP): + loss, = exe.run(fetch_list=[avg_cost.name], + feed=feeder.feed(get_data())) + if need_save and model_dir: + io.save_persistables(startup_exe, model_dir, trainer_prog) + + var = np.array(fluid.global_scope().find_var('__fc_b__').get_tensor()) + print(np.ravel(var).tolist()) + + +if __name__ == "__main__": + paddle.dataset.common.download(DATA_URL, 'simnet', DATA_MD5, "train") + runtime_main(TestDistSaveLoad2x2) diff --git a/python/paddle/fluid/tests/unittests/test_dist_save_load.py b/python/paddle/fluid/tests/unittests/test_dist_save_load.py new file mode 100644 index 0000000000000000000000000000000000000000..8b50a312341ddb9bffcaf9f685805f80519f8427 --- /dev/null +++ b/python/paddle/fluid/tests/unittests/test_dist_save_load.py @@ -0,0 +1,89 @@ +# Copyright (c) 2018 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. +from __future__ import print_function + +import os +import shutil +import unittest +import tempfile + +import numpy as np + +from test_dist_base import TestDistBase, RUN_STEP + + +class TestDistSaveLoadDense2x2(TestDistBase): + def _setup_config(self): + self._sync_mode = True + self._enforce_place = "CPU" + + def check_with_place(self, + model_file, + delta=1e-3, + check_error_log=False, + need_envs={}): + + required_envs = { + "PATH": os.getenv("PATH", ""), + "PYTHONPATH": os.getenv("PYTHONPATH", ""), + "LD_LIBRARY_PATH": os.getenv("LD_LIBRARY_PATH", ""), + "http_proxy": "" + } + + required_envs.update(need_envs) + + if check_error_log: + required_envs["GLOG_v"] = "7" + required_envs["GLOG_logtostderr"] = "1" + + model_dir = tempfile.mkdtemp() + + local_env = {} + local_env["SAVE"] = "1" + local_env["MODEL_DIR"] = model_dir + local_env.update(required_envs) + + cluster_env = {} + cluster_env["LOAD"] = "1" + cluster_env["MODEL_DIR"] = model_dir + cluster_env.update(required_envs) + + local_var = self._run_local(model_file, local_env, check_error_log) + tr0_var, tr1_var = self._run_cluster(model_file, cluster_env, + check_error_log) + + shutil.rmtree(model_dir) + + local_np = np.array(eval(local_var[0])) + train0_np = np.array(eval(tr0_var[0])) + train1_np = np.array(eval(tr1_var[0])) + self.assertAlmostEqual(local_np.all(), train0_np.all(), delta=delta) + self.assertAlmostEqual(local_np.all(), train1_np.all(), delta=delta) + self.assertAlmostEqual(train0_np.all(), train1_np.all(), delta=delta) + + def test_dist(self): + need_envs = { + "IS_DISTRIBUTED": '0', + "IS_SPARSE": '0', + 'IS_SELF_CONTAINED_LR': '1' + } + self.check_with_place( + "dist_save_load.py", + delta=0, + check_error_log=False, + need_envs=need_envs) + + +if __name__ == "__main__": + unittest.main() diff --git a/python/paddle/fluid/transpiler/distribute_transpiler.py b/python/paddle/fluid/transpiler/distribute_transpiler.py index 4af13b605fa7054df097e3bf0ed0c71f468f02de..9066fc9d1bf13176862f6debf0ed0bedaaaf3eba 100644 --- a/python/paddle/fluid/transpiler/distribute_transpiler.py +++ b/python/paddle/fluid/transpiler/distribute_transpiler.py @@ -920,11 +920,11 @@ to transpile() call.") block_idx = int(block_name.split(block_suffix)[1]) orig_var = self.origin_program.global_block().vars[orig_var_name] - skip_numel = 0 + skip_dim0 = 0 slice_vars = self.param_var_mapping[orig_var_name] for slice_var in slice_vars[:block_idx]: - skip_numel += reduce(lambda x, y: x * y, slice_var.shape) - slice_vars_and_attrs.append([orig_var, skip_numel, param]) + skip_dim0 += slice_var.shape[0] + slice_vars_and_attrs.append([orig_var, skip_dim0, param]) return slice_vars_and_attrs