# 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 time import os role = os.getenv("TRAINING_ROLE", "TRAINER") import numpy as np from pgl.utils.logger import log import paddle.fluid as F import paddle.fluid.layers as L from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import StrategyFactory from paddle.fluid.incubate.fleet.collective import DistributedStrategy from paddle.fluid.transpiler.distribute_transpiler import DistributeTranspilerConfig from paddle.fluid.incubate.fleet.collective import fleet as cfleet from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import fleet as tfleet import paddle.fluid.incubate.fleet.base.role_maker as role_maker from tensorboardX import SummaryWriter class Learner(object): @classmethod def factory(cls, name): if name == "cpu": return TranspilerLearner() elif name == "gpu": return CollectiveLearner() else: raise ValueError def build(self, model, data_gen, config): raise NotImplementedError def warmstart(self, program, path='./checkpoints'): def _existed_persitables(var): #if not isinstance(var, fluid.framework.Parameter): # return False if not F.io.is_persistable(var): return False param_path = os.path.join(path, var.name) log.info("Loading parameter: {} persistable: {} exists: {}".format( param_path, F.io.is_persistable(var), os.path.exists(param_path), )) return os.path.exists(param_path) F.io.load_vars( self.exe, path, main_program=program, predicate=_existed_persitables ) def start(self): batch = 0 start = time.time() trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0")) if trainer_id == 0: writer = SummaryWriter(os.path.join(self.config.output_path, "train_history")) for epoch_idx in range(self.config.epoch): for idx, batch_feed_dict in enumerate(self.model.data_loader()): try: cpu_time = time.time() batch += 1 batch_loss = self.exe.run( self.program, feed=batch_feed_dict, fetch_list=[self.model.loss]) end = time.time() if trainer_id == 0: writer.add_scalar("loss", np.mean(batch_loss), batch) if batch % self.config.log_per_step == 0: log.info( "Epoch %s Batch %s %s-Loss %s \t Speed(per batch) %.5lf/%.5lf sec" % (epoch_idx, batch, "train", np.mean(batch_loss), (end - start) /batch, (end - cpu_time))) writer.flush() if batch % self.config.save_per_step == 0: self.fleet.save_persistables(self.exe, os.path.join(self.config.output_path, str(batch))) except Exception as e: log.info("Pyreader train error") log.exception(e) log.info("epcoh %s done." % epoch_idx) def stop(self): self.fleet.save_persistables(self.exe, os.path.join(self.config.output_path, "last")) class TranspilerLearner(Learner): def __init__(self): training_role = os.getenv("TRAINING_ROLE", "TRAINER") paddle_role = role_maker.Role.WORKER place = F.CPUPlace() if training_role == "PSERVER": paddle_role = role_maker.Role.SERVER # set the fleet runtime environment according to configure port = os.getenv("PADDLE_PORT", "6174") pserver_ips = os.getenv("PADDLE_PSERVERS") # ip,ip... eplist = [] for ip in pserver_ips.split(","): eplist.append(':'.join([ip, port])) pserver_endpoints = eplist # ip:port,ip:port... worker_num = int(os.getenv("PADDLE_TRAINERS_NUM", "0")) trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0")) role = role_maker.UserDefinedRoleMaker( current_id=trainer_id, role=paddle_role, worker_num=worker_num, server_endpoints=pserver_endpoints) tfleet.init(role) tfleet.save_on_pserver = True def build(self, model, data_gen, config): self.optimize(model.loss, config.optimizer_type, config.lr) self.init_and_run_ps_worker(config.ckpt_path) self.program = self.complie_program(model.loss) self.fleet = tfleet model.data_loader.decorate_batch_generator( data_gen, places=F.cpu_places()) self.config = config self.model = model def optimize(self, loss, optimizer_type, lr): strategy = DistributeTranspilerConfig() strategy.sync_mode = False log.info('learning rate:%f' % lr) if optimizer_type == "sgd": optimizer = F.optimizer.SGD(learning_rate=lr) elif optimizer_type == "adam": # Don't slice tensor ensure convergence optimizer = F.optimizer.Adam(learning_rate=lr, lazy_mode=True) else: raise ValueError("Unknown Optimizer %s" % optimizer_type) #create the DistributeTranspiler configure optimizer = tfleet.distributed_optimizer(optimizer, strategy) optimizer.minimize(loss) def init_and_run_ps_worker(self, ckpt_path): # init and run server or worker self.exe = F.Executor(F.CPUPlace()) if tfleet.is_server(): tfleet.init_server() self.warmstart(tfleet.startup_program, path=ckpt_path) tfleet.run_server() exit() if tfleet.is_worker(): log.info("start init worker done") tfleet.init_worker() self.exe.run(tfleet.startup_program) def complie_program(self, loss): num_threads = int(os.getenv("CPU_NUM", 1)) exec_strategy = F.ExecutionStrategy() exec_strategy.num_threads = num_threads exec_strategy.use_thread_barrier = False build_strategy = F.BuildStrategy() build_strategy.enable_inplace = True build_strategy.memory_optimize = True build_strategy.remove_unnecessary_lock = False build_strategy.memory_optimize = False build_strategy.async_mode = False if num_threads > 1: build_strategy.reduce_strategy = F.BuildStrategy.ReduceStrategy.Reduce log.info("start build compile program...") compiled_prog = F.compiler.CompiledProgram(tfleet.main_program ).with_data_parallel( loss_name=loss.name, build_strategy=build_strategy, exec_strategy=exec_strategy) return compiled_prog class CollectiveLearner(Learner): def __init__(self): role = role_maker.PaddleCloudRoleMaker(is_collective=True) cfleet.init(role) def optimize(self, loss, optimizer_type, lr): optimizer = F.optimizer.Adam(learning_rate=lr) dist_strategy = DistributedStrategy() dist_strategy.enable_sequential_execution = True optimizer = cfleet.distributed_optimizer(optimizer, strategy=dist_strategy) _, param_grads = optimizer.minimize(loss, F.default_startup_program()) def build(self, model, data_gen, config): self.optimize(model.loss, config.optimizer_type, config.lr) self.program = cfleet.main_program gpu_id = int(os.getenv("FLAGS_selected_gpus", "0")) place = F.CUDAPlace(gpu_id) self.exe = F.Executor(place) self.exe.run(F.default_startup_program()) self.warmstart(F.default_startup_program(), config.ckpt_path) self.fleet = cfleet model.data_loader.decorate_batch_generator( data_gen, places=place) self.config = config self.model = model