# Copyright (c) 2020 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 annotations import os import numpy as np import paddle from paddle.base import core, framework, unique_name from paddle.base.dygraph.base import switch_to_static_graph from paddle.framework import in_dynamic_mode from paddle.nn.layer import layers from paddle.pir.core import datatype_to_vartype __all__ = [] PIR_INFER_MODEL_SUFFIX = ".json" from .translated_layer import ( BUFFER_NAME_PREFIX, INFER_PARAMS_SUFFIX, PARAMETER_NAME_PREFIX, ) def _load_pir_program(model_file_path): program = paddle.static.Program() trainable = paddle.base.core.deserialize_pir_program( model_file_path, program ) return program, trainable @switch_to_static_graph def _generate_unique_var_name(prefix): return unique_name.generate_with_ignorable_key(prefix) @switch_to_static_graph def _generate_unique_var_name(prefix): return unique_name.generate(prefix) from paddle.static.pir_io import get_pir_parameters def _get_pir_parameters_var_names(program): persistable_vars = [] persistable_names = [] rename_new_old_dict = {} param, opt = get_pir_parameters(program) vars = param + opt for var in vars: persistable_vars.append(var) origin_name = var.name var.name = _generate_unique_var_name(var.name) rename_new_old_dict[var.name] = origin_name persistable_names.append(var.name) return ( persistable_vars, persistable_names, rename_new_old_dict, ) class _PirProgramHolder: def __init__(self, program, trainable): super().__init__() # input, output, persistable, self._input_vars = [] self._output_vars = [] self._parameter_vars = [] self._parameter_names = [] self.support_train = trainable # append suffix var name dict self._suffix_varname_dict = None self._infer_program = program self._preprocess() def _preprocess(self): ( self._parameter_vars, self._parameter_names, self._suffix_varname_dict, ) = _get_pir_parameters_var_names(self._infer_program) block = self._infer_program.global_block() for op in block.ops: if op.name() == 'pd_op.data': self._input_vars.append(op.result(0)) elif op.name() == 'pd_op.feed': var_name = op.attr()["name"] org_value = op.result(0) with block: value = paddle._pir_ops.data( name=var_name, shape=org_value.shape, dtype=org_value.dtype, place=paddle.base.core.Place(), ) org_value.replace_all_uses_with(value) value.get_defining_op().move_before(op) block.remove_op(op) if op.name() == 'pd_op.fetch': self._output_vars.append(op.operand_source(0)) with block: paddle._pir_ops.set_persistable_value( op.operand_source(0), "output_" + str(len(self._output_vars) - 1), ) block.remove_op(op) @property def infer_program(self): return self._infer_program @property def input_vars(self): return self._input_vars @property def output_vars(self): return self._output_vars @property def persistable_names(self): return self._parameter_names @property def persistable_vars(self): return self._parameter_vars # [ PirTranslatedLayer : Run program in dygraph mode ] # # DESIGN IDEA: using an special operator `PirRunProgram`, execute program inside operator. # # Op's Inputs: # - the input variable of the user feed # - the necessary parameters of the network # Op's Outputs: # - the output variable of fetch # # This op receives a complete program desc, internally creates scope # and executor, executes this program. Key points: # # 1. Data Sharing: # The variable/parameter of the dynamic graph is not in the scope, so before the op # executes the program internally, create persistent variables with the # same name as feed, parameters, and fetch in the scope, and share the # DenseTensor of the op input. # # 2. Forward and Backward Separation: # Because the dynamic graph op performs the forward and backward separately, # in the forward op RunProgram, we only execute the forward part of whole program, # and in the backward op RunProgramGrad, we execute the backward part of program. # We can not separate the program into forward and backward part, which will # make some control flow execution logic wrong. # NOTE: [compatible] deal with model saved by save_inference_model, # which need get var info from program desc def _load_pir_parameter_vars(model_path, program_holder, params_filename): # construct var dict load_var_dict = {} load_var_list = [] other_var_dict = {} load_densetensor_list = [] persistable_var = program_holder.persistable_vars persistable_var_name = program_holder.persistable_names origin_persistable_var_name = [ program_holder._suffix_varname_dict[var_name] for var_name in persistable_var_name ] for name, var in sorted(zip(origin_persistable_var_name, persistable_var)): if var.persistable: # use default shape and dtype new_var = framework.EagerParamBase( shape=var.shape, # only to pass check, this shape is not meaningful dtype=core.VarDesc.VarType.FP32, name=var.name, persistable=True, ) new_var.stop_gradient = var.stop_gradient load_var_dict[name] = new_var load_var_list.append(new_var) load_densetensor_list.append(new_var.get_tensor()) else: new_var = core.eager.Tensor( dtype=datatype_to_vartype[var.dtype], dims=var.shape, name=var.name, type=core.VarDesc.VarType.DENSE_TENSOR, place=framework._current_expected_place(), persistable=False, ) other_var_dict[name] = new_var # load all vars assert params_filename is not None, "params_filename should not be None." var_file_path = os.path.join(model_path, params_filename) if os.path.exists(var_file_path): core.load_combine_func( var_file_path, list(load_var_dict.keys()), load_densetensor_list, False, framework._current_expected_place(), ) else: raise ValueError( f"The file {var_file_path} does not exist. Please check the model path." ) load_var_dict.update(other_var_dict) return load_var_dict def _construct_program_holders(model_path, model_filename=None): # make sure the path has been checked program_holder_dict = {} if model_filename is not None: # [compatible] if assign model_filename, only can load one program as Layer.forward model_filename = os.path.basename(model_filename) model_file_path = os.path.join(model_path, model_filename) model_name = model_filename[: -len(PIR_INFER_MODEL_SUFFIX)] # Load every file that meets the requirements in the directory model_path. for filename in os.listdir(model_path): if model_filename == filename: func_name = 'forward' model_file_path = os.path.join(model_path, model_filename) elif filename.endswith( PIR_INFER_MODEL_SUFFIX ) and filename.startswith(model_name): parsing_names = filename[ len(model_name) : -len(PIR_INFER_MODEL_SUFFIX) + 1 ].split('.') if len(parsing_names) == 3 and len(parsing_names[1]) > 0: func_name = parsing_names[1] model_file_path = os.path.join(model_path, filename) else: continue else: continue program, trainable = _load_pir_program(model_file_path) program_holder_dict[func_name] = _PirProgramHolder( program, trainable ) else: for _, _, file_names in os.walk(model_path): for name in file_names: if 'model' in name: model_file_path = os.path.join(model_path, name) method_name = name.strip('_') if method_name == 'model': method_name = 'forward' else: method_name.replace('model', '') program, trainable = _load_pir_program(model_file_path) program_holder_dict[method_name] = _PirProgramHolder( program, trainable ) return program_holder_dict def _construct_params_and_buffers(model_path, programs, params_filename=None): params_path = os.path.join(model_path, str(params_filename)) if params_filename is not None and not os.path.exists(params_path): # When saving XX, there is only '*.pdmodel' return {} else: var_dict = _load_pir_parameter_vars( model_path, programs['forward'], params_filename ) model_name = params_filename[: -len(INFER_PARAMS_SUFFIX)] # Load every file that meets the requirements in the directory model_path. for file_name in os.listdir(model_path): if file_name.startswith(model_name) and file_name.endswith( INFER_PARAMS_SUFFIX ): parsing_names = file_name[ len(model_name) : -len(INFER_PARAMS_SUFFIX) + 1 ].split('.') if len(parsing_names) == 3 and len(parsing_names[1]) > 0: func_name = parsing_names[1] else: continue else: continue var_dict.update( _load_pir_parameter_vars( model_path, programs[func_name], file_name ) ) return var_dict def _run_dygraph(instance, input, program_holder, method_name): # 1. prepare inputs, outputs, attrs input_tensors = [] input_tensor_names = [] for i, value in enumerate(input): if not isinstance(value, (np.ndarray, core.eager.Tensor)): raise TypeError( f"The type of input in PirTranslatedLayer must be numpy array or Variable(Tensor), but received {type(value)}." ) # NOTE: In order to unify the API, firstly convert the input to Tensor if isinstance(value, np.ndarray): tensor = core.eager.Tensor( value=value, name=program_holder.input_vars[i].name, persistable=False, place=framework._current_expected_place(), zero_copy=True, ) else: tensor = value # NOTE: we changed var name here, # but it may be an important name set by user tensor.name = program_holder.input_vars[i].name input_tensor_names.append(tensor.name) input_tensors.append(tensor) if instance._get_partial_program_layer(method_name) is None: persistable_tensors = [] origin_persistable_var_name = [ program_holder._suffix_varname_dict[var_name] for var_name in program_holder.persistable_names ] for var_name in origin_persistable_var_name: dy_var_name = instance._persistable_var_name_dict[var_name] if dy_var_name in instance._parameters: persistable_tensors.append(instance._parameters[dy_var_name]) elif dy_var_name in instance._buffers: persistable_tensors.append(instance._buffers[dy_var_name]) else: raise ValueError( f"The persistable variable {var_name} does not exist in current PirTranslatedLayer." ) from paddle.jit.dy2static.pir_partial_program import PartialProgramLayer inputs = program_holder.input_vars outputs = program_holder.output_vars parameters = (persistable_tensors, program_holder.persistable_vars) layer = PartialProgramLayer( program_holder.infer_program, inputs, outputs, parameters, ) instance._set_partial_program_layer(method_name, layer) layer = instance._get_partial_program_layer(method_name) if instance._is_test: layer.training = False else: if not program_holder.support_train: raise ValueError( "The model is not trainable, please check model_file of jit.save." ) else: layer.training = True return layer(input_tensors) def _run_static_graph(inputs, program_holder, src_program): ''' This function is used when the pirTranslatedLayer is applied for dy_to_static conversion. ''' dst_program = paddle.static.default_main_program() value_map = paddle.pir.IrMapping() # Establish a mapping relationship between existing parameters # and corresponding parameters in the program to be copied len_dst_op = len(dst_program.global_block().ops) for dst_op in dst_program.global_block().ops: if dst_op.name() == "builtin.parameter": for src_op in src_program.global_block().ops[:len_dst_op]: if ( src_op.name() == dst_op.name() and src_op.result(0).name == dst_op.result(0).name ): for i in range(src_op.num_results()): value_map.add(src_op.result(i), dst_op.result(i)) # Establish a mapping relationship between truly inputs # and corresponding inputs in the program to be copied src_inputs = program_holder.input_vars if len(src_inputs) != len(inputs): raise ValueError( f"The number of input is invalid, expected {len(src_inputs)}, but received {len(inputs)}." ) for src_input, input_ in zip(src_inputs, inputs): value_map.add(src_input, input_) # find the insert point for copy current_insert_point = paddle.pir.get_current_insertion_point() current_block = current_insert_point.block() src_program.copy_to_block(value_map, current_block) output = [value_map.look_up(v) for v in program_holder.output_vars] return output[0] if len(output) == 1 else output def _collect_current_and_parent_var(program, block_idx): ''' Get variables in current block and its parent block. Args: program(Program): The program containing the current block. block_idx(int): index of current block. Returns: List: list of variables. ''' vars = [] if block_idx < 0: return vars for var in program.block(block_idx).vars: vars.append(var) parent_idx = program.block(block_idx).parent_idx if parent_idx > -1: vars += _collect_current_and_parent_var(program, parent_idx) return vars class PirTranslatedLayer(layers.Layer): """ PirTranslatedLayer is a ``paddle.nn.Layer`` for holding the model loaded by :ref:`api_paddle_jit_load` . It can be used like a general Layer object in eval or train mode. .. note: The PirTranslatedLayer objects should not be created by constructor, it only can be loaded and constructed by :ref:`api_paddle_jit_load` . Examples: .. code-block:: pycon >>> # doctest: +SKIP('`paddle.jit.to_static` can not run in xdoctest') >>> import numpy as np >>> import paddle >>> import paddle.nn as nn >>> import paddle.optimizer as opt >>> BATCH_SIZE = 16 >>> BATCH_NUM = 4 >>> EPOCH_NUM = 4 >>> IMAGE_SIZE = 784 >>> CLASS_NUM = 10 >>> # define a random dataset >>> class RandomDataset(paddle.io.Dataset): # type: ignore[type-arg] ... def __init__(self, num_samples): ... self.num_samples = num_samples ... ... def __getitem__(self, idx): ... image = np.random.random([IMAGE_SIZE]).astype('float32') ... label = np.random.randint(0, CLASS_NUM - 1, (1,)).astype('int64') ... return image, label ... ... def __len__(self): ... return self.num_samples >>> class LinearNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear = nn.Linear(IMAGE_SIZE, CLASS_NUM) ... ... @paddle.jit.to_static ... def forward(self, x): ... return self._linear(x) >>> def train(layer, loader, loss_fn, opt): ... for epoch_id in range(EPOCH_NUM): ... for batch_id, (image, label) in enumerate(loader()): ... out = layer(image) ... loss = loss_fn(out, label) ... loss.backward() ... opt.step() ... opt.clear_grad() ... print("Epoch {} batch {}: loss = {}".format(epoch_id, batch_id, np.mean(loss.numpy()))) >>> # 1. train & save model. >>> # create network >>> layer = LinearNet() >>> loss_fn = nn.CrossEntropyLoss() >>> adam = opt.Adam(learning_rate=0.001, parameters=layer.parameters()) >>> # create data loader >>> dataset = RandomDataset(BATCH_NUM * BATCH_SIZE) >>> loader = paddle.io.DataLoader( ... dataset, ... batch_size=BATCH_SIZE, ... shuffle=True, ... drop_last=True, ... num_workers=2, ... ) >>> # train >>> train(layer, loader, loss_fn, adam) >>> # save >>> model_path = "linear.example.model" >>> paddle.jit.save(layer, model_path) >>> # 2. load model as PirTranslatedLayer >>> # load >>> translated_layer = paddle.jit.load(model_path) >>> # inference >>> translated_layer.eval() >>> x = paddle.randn([1, IMAGE_SIZE], 'float32') >>> pred = translated_layer(x) >>> # fine-tune >>> translated_layer.train() >>> adam = opt.Adam(learning_rate=0.001, parameters=translated_layer.parameters()) >>> train(translated_layer, loader, loss_fn, adam) """ def __init__( self, programs: dict[str, paddle.static.Program], persistable_vars: dict[str, paddle.Tensor], ): super().__init__() if not isinstance(programs, dict): raise TypeError( "PirTranslatedLayer need to use _ProgramHolder's dict for initialization." ) if not isinstance(persistable_vars, dict): raise TypeError( "PirTranslatedLayer need to use persistable variable dict for initialization." ) self._program_holder_dict = programs # NOTE(chenweihang): [ why not use var name directly? ] # When add parameter or buffer to Layer by follow apis, # the variable name can't contain `.`, because which may cause # AttributeError when access the newly added parameter or buffer # in the form of `self.**.**``, but the EagerParamBase or BarBase # name contains `.` originally, such as `linear_0.w_0`, so here # need to generate new var name for each var self._persistable_var_name_dict = {} # the PirTranslatedLayer object held var names count started from 0 with unique_name.guard(): for name, var in persistable_vars.items(): if isinstance(var, framework.EagerParamBase): dy_name = _generate_unique_var_name(PARAMETER_NAME_PREFIX) self._persistable_var_name_dict[name] = dy_name self.add_parameter(dy_name, var) elif isinstance(var, core.eager.Tensor): dy_name = _generate_unique_var_name(BUFFER_NAME_PREFIX) self._persistable_var_name_dict[name] = dy_name self.register_buffer(dy_name, var) else: raise TypeError( "Adding persistent variable which to layer is not supported now" ) self._is_test = True self._input_args_names = None self._partial_program_layers = {} @staticmethod @framework.dygraph_only def _construct(model_path, configs=None): # 0. dir and filename check model_path = os.path.normpath(model_path) if not os.path.isdir(model_path): raise ValueError(f"There is no directory named '{model_path}'") model_filename = None params_filename = None if configs is not None: model_filename = configs.model_filename params_filename = configs.params_filename # 1. load program desc & construct _ProgramHolder programs = _construct_program_holders(model_path, model_filename) # 2. load layer parameters & buffers persistable_vars = _construct_params_and_buffers( model_path, programs, params_filename ) # 3. construct PirTranslatedLayer object translated_layer = PirTranslatedLayer(programs, persistable_vars) # 4. create PirTranslatedLayer's execution method for method_name, program_holder in programs.items(): if translated_layer._input_args_names is None: translated_layer._input_args_names = [ ins.name for ins in program_holder.input_vars ] setattr( PirTranslatedLayer, method_name, PirTranslatedLayer._execution_method_creator( method_name, program_holder ), ) # 5. set PirTranslatedLayer's default mode to eval translated_layer.eval() return translated_layer @staticmethod def _execution_method_creator(method_name, program_holder): def __i_m_p_l__(self, *input): program_holder = self._program_holder_dict[__i_m_p_l__.__name__] # When using jit.save, it runs in static graph mode. # Run in dynamic graph mode when the model is inferring. if in_dynamic_mode(): return _run_dygraph(self, input, program_holder, method_name) else: return _run_static_graph( input, program_holder, program_holder.infer_program ) __i_m_p_l__.__name__ = method_name return __i_m_p_l__ def train(self): self._is_test = False self.training = True def eval(self): self._is_test = True self.training = False def program(self, method_name='forward'): """ Gets translated program of specified method. Args: - method_name (string): method name corresponding to the program to be obtained. Default: 'forward'. Returns: Program Examples: .. code-block:: pycon >>> # doctest: +SKIP('`paddle.jit.to_static` can not run in xdoctest') >>> import numpy as np >>> import paddle >>> from paddle import nn >>> import paddle.optimizer as opt >>> BATCH_SIZE = 16 >>> BATCH_NUM = 4 >>> EPOCH_NUM = 4 >>> IMAGE_SIZE = 784 >>> CLASS_NUM = 10 >>> # define a random dataset >>> class RandomDataset(paddle.io.Dataset): # type: ignore[type-arg] ... def __init__(self, num_samples): ... self.num_samples = num_samples ... ... def __getitem__(self, idx): ... image = np.random.random([IMAGE_SIZE]).astype('float32') ... label = np.random.randint(0, CLASS_NUM - 1, (1,)).astype('int64') ... return image, label ... ... def __len__(self): ... return self.num_samples >>> class LinearNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self._linear = nn.Linear(IMAGE_SIZE, CLASS_NUM) ... ... @paddle.jit.to_static ... def forward(self, x): ... return self._linear(x) >>> def train(layer, loader, loss_fn, opt): ... for epoch_id in range(EPOCH_NUM): ... for batch_id, (image, label) in enumerate(loader()): ... out = layer(image) ... loss = loss_fn(out, label) ... loss.backward() ... opt.step() ... opt.clear_grad() ... print("Epoch {} batch {}: loss = {}".format(epoch_id, batch_id, np.mean(loss.numpy()))) >>> # create network >>> layer = LinearNet() >>> loss_fn = nn.CrossEntropyLoss() >>> adam = opt.Adam(learning_rate=0.001, parameters=layer.parameters()) >>> # create data loader >>> dataset = RandomDataset(BATCH_NUM * BATCH_SIZE) >>> loader = paddle.io.DataLoader( ... dataset, ... batch_size=BATCH_SIZE, ... shuffle=True, ... drop_last=True, ... num_workers=2, ... ) >>> # train >>> train(layer, loader, loss_fn, adam) >>> # save >>> model_path = "linear.example.model" >>> paddle.jit.save(layer, model_path) >>> # load >>> translated_layer = paddle.jit.load(model_path) >>> # get program >>> program = translated_layer.program() """ # 1. get program holder program_holder = self._get_program_holder(method_name) # 2. get inference program desc program = program_holder.infer_program return program def _get_program_holder(self, method_name='forward'): program_holder = self._program_holder_dict.get(method_name, None) if program_holder is None: raise ValueError( f"The method `{method_name}` does not exist in loaded PirTranslatedLayer." ) return program_holder def _input_spec(self, method_name='forward'): # 1. get program holder program_holder = self._get_program_holder(method_name) # 2. build input spec by input desc input_spec = [] for var in program_holder.input_vars: spec = paddle.static.InputSpec( shape=var.shape, dtype=var.dtype, name=var.name, ) input_spec.append(spec) return input_spec def _output_spec(self, method_name='forward'): # 1. get program holder program_holder = self._get_program_holder(method_name) # 2. build output spec by output desc output_spec = [] for var in program_holder.output_vars: # NOTE(chenweihang): InputSpec describes a tensor, not just input. # Maybe the name is not good enough. Here we use InputSpec to # construct the description of Output tensor spec = paddle.static.InputSpec( shape=var.shape, dtype=var.dtype, name=var.name, ) output_spec.append(spec) return output_spec def _get_partial_program_layer(self, method_name): return self._partial_program_layers.get(method_name, None) def _set_partial_program_layer(self, method_name, layer): self._partial_program_layers[method_name] = layer