# Copyright (c) 2022 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 re import os import errno import pickle import warnings import logging import numpy as np import paddle from paddle import fluid from paddle.fluid import core from paddle.fluid.framework import static_only from .utils import get_dist_attr from .converter import Converter from .process_group import _g_process_group_map from ..utils import get_logger def check_filename(re_exp, filename): if re.search(re_exp, filename): return True else: return False def _process_path(path): filename = os.path.basename(path) if filename == "": raise ValueError( "path should be of 'dirname/filename' format, but received filename is empty string" ) try: dirname = os.path.dirname(path) os.makedirs(dirname) except OSError as e: if e.errno != errno.EEXIST: raise return dirname, filename class DistributedSaver: def __init__(self): self._logger = get_logger(logging.INFO) def save(self, path, serial_program, dist_main_program, dist_context): dirname, filename = _process_path(path) rank_id = paddle.distributed.get_rank() # save serial program when rank id is 0 if rank_id == 0: self._save_rank_mapping(dirname) serial_model_filename = filename + "_serial.pdmodel" serial_model_path = os.path.join(dirname, serial_model_filename) with open(serial_model_path, "wb") as f: f.write(serial_program.desc.serialize_to_string()) # save distributed main program dist_model_filename = filename + "_dist" + str(rank_id) + ".pdmodel" dist_model_path = os.path.join(dirname, dist_model_filename) with open(dist_model_path, "wb") as f: f.write(dist_main_program.desc.serialize_to_string()) # save distributed params dist_param_filename = filename + "_dist" + str(rank_id) + ".pdparams" dist_param_path = os.path.join(dirname, dist_param_filename) dist_param = { k: np.array(v) for k, v in dist_main_program.state_dict().items() } with open(dist_param_path, "wb") as f: pickle.dump(dist_param, f) # save distributed attribute dist_attr_filename = filename + "_dist" + str(rank_id) + ".pdattr" dist_attr_path = os.path.join(dirname, dist_attr_filename) dist_attrs = get_dist_attr(dist_main_program, dist_context) with open(dist_attr_path, "wb") as f: pickle.dump(dist_attrs, f) # TODO:save cluster.json def load(self, path, program, dist_context, strict=True, load_optimizer=True): # TODO: if `program` is None, load `path.pdmodel`. filename = os.path.basename(path) if filename == "": raise ValueError( "path should be of 'dirname/filename' format, but received filename is empty string" ) dirname = os.path.dirname(path) # load path.pdparam param_file_list = [] for param_file in os.listdir(dirname): if check_filename('{}(.*)_dist(.*).pdparams'.format(filename), param_file): param_file_list.append(os.path.join(dirname, param_file)) param_file_list.sort() self._logger.info("Load distributed attribute file: {}".format( param_file_list)) param_dict = {} for param_file in param_file_list: with open(param_file, 'rb') as f: state_dict_info = pickle.load(f, encoding='latin1') for name, value in state_dict_info.items(): if name in param_dict: param_dict[name].append(np.array(value)) else: param_dict[name] = [np.array(value)] # load path.pdattr dist_attr_file_list = [] for dist_attr_file in os.listdir(dirname): if check_filename('{}(.*)_dist(.*).pdattr'.format(filename), dist_attr_file): dist_attr_file_list.append( os.path.join(dirname, dist_attr_file)) dist_attr_file_list.sort() self._logger.info("Load distributed attribute file: {}".format( dist_attr_file_list)) pre_dist_attr = {} for dist_attr_file in dist_attr_file_list: with open(dist_attr_file, 'rb') as f: dist_attr = pickle.load(f, encoding='latin1') for name, attr in dist_attr.items(): if name not in pre_dist_attr: pre_dist_attr[name] = attr # get current dist_attr cur_dist_attr = get_dist_attr(program, dist_context) # param convert converter = Converter(param_dict, pre_dist_attr, cur_dist_attr) param_dict = converter.convert(strict=strict) program.set_state_dict(param_dict) def save_inference_model(self, path, feed_vars, fetch_vars, exe, **kwargs): dirname, filename = _process_path(path) # save distributed inference program rank_id = paddle.distributed.get_rank() if rank_id == 0: self._save_rank_mapping(dirname) op_role_key = core.op_proto_and_checker_maker.kOpRoleAttrName() op_role_forward = int(core.op_proto_and_checker_maker.OpRole.Forward) dist_main_prog = kwargs.get('program', None) if not dist_main_prog: dist_main_prog = fluid.default_main_program() global_block = dist_main_prog.global_block() ops = global_block.ops feed_vars_names = list(map(lambda x: x.name, feed_vars)) fetch_vars_names = list(map(lambda x: x.name, fetch_vars)) last_idx = -1 for idx, op in enumerate(ops): if op.attr(op_role_key) != op_role_forward: continue if op.type == "read" or op.type == "feed" or op.type == 'recv_v2': feed_vars_names += op.output("Out") if op.type == "send_v2": fetch_vars_names += op.input("X") last_idx = max(idx, last_idx) for out_name in op.output_arg_names: if out_name in fetch_vars_names: last_idx = max(idx, last_idx) used_inputs = [] used_outputs = [] for idx, op in enumerate(ops): if idx > last_idx: break used_inputs += op.input_arg_names used_outputs += op.output_arg_names dist_feed_vars_names = list(set(feed_vars_names) & set(used_inputs)) dist_fetch_vars_names = list(set(fetch_vars_names) & set(used_outputs)) dist_feed_vars = [ global_block.vars[name] for name in dist_feed_vars_names ] dist_fetch_vars = [ global_block.vars[name] for name in dist_fetch_vars_names ] # NOTE: `paddle.static.save_inference_model` does not support subblock. dist_filename = filename + "_dist" + str(rank_id) dist_path = os.path.join(dirname, dist_filename) paddle.static.save_inference_model( dist_path, dist_feed_vars, dist_fetch_vars, exe, program=dist_main_prog) def _save_rank_mapping(self, dirname): path = os.path.join(dirname, 'rank_mapping.csv') f = open(path, 'w') f.write('[ring_id -> ranks]\n') for process_group in _g_process_group_map.values(): ring_id = process_group._group_id ranks = [str(rank) for rank in process_group._ranks] id_to_rank = str(ring_id) + "," + ",".join(ranks) + '\n' f.write(id_to_rank) id_to_rank = "" f.write('[rank -> ring_ids]\n') rank_to_id_dict = {} for process_group in _g_process_group_map.values(): ring_id = process_group._group_id for rank in process_group._ranks: if rank in rank_to_id_dict: rank_to_id_dict[rank].append(str(ring_id)) else: rank_to_id_dict[rank] = [str(ring_id)] rank_to_id = "" for item, val in rank_to_id_dict.items(): rank_to_id += str(item) + "," rank_to_id += ",".join(val) + "\n" f.write(rank_to_id) rank_to_id = "" f.close()