# 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 paddle.fluid import core
import numpy as np
import math
import os
from paddle.fluid.executor import global_scope
from paddle.fluid import io
__all__ = ['Calibrator']
class Calibrator(object):
'''
The calibrator class transforms the program and updates the calculated scale into it.
This is INT8 v1 calibration tool, mainly for the support of ResNet-50 and MobileNet.
'''
# TODO(guomingz): Below op list will be updated once more INT8 op kernels are supported.
non_conv_int8_op_type = ("pool2d")
supported_int8_op_type = ("conv2d", "pool2d")
const_sign_op_type = ('pool2d', 'reshape', 'concat', 'transpose')
u8_max = 255
s8_max = 127
def __init__(self, *args, **kwargs):
self.program = kwargs['program']
self.pretrained_model = kwargs['pretrained_model']
self.debug = kwargs['debug'] if 'debug' in kwargs else False
self.algo = kwargs['algo']
self.output = kwargs['output']
self.feed_var_names = kwargs['feed_var_names']
self.fetch_list = kwargs['fetch_list']
self.exe = kwargs['exe']
self._conv_input_var_name = []
self._conv_output_var_name = []
self._pool2d_output_var_name = []
self._weights_var_name = []
self._residual_input_var_name = []
self._int8_output_var_op_index_dict = {}
self._conv_op_index = [
index for index, value in enumerate(self.program.global_block().ops)
if value.type == 'conv2d'
]
self._var_max_value_map = {}
self._var_max_range = {}
self._weights_scaling_factor = {}
self._u8_output_var = []
self._s8_output_var = []
self._persistable_vars = []
self._sampling_data = {}
self.__init_analysis()
self.__generate_output_program()
def save_int8_model(self):
self.__sampling(self._sampling_data)
self.__save_scale()
self.__update_program()
self.__update_output_program_attr()
self.__display_debug()
self.__save_offline_model()
def sample_data(self):
'''
Sampling the tensor data of variable.
'''
for i in self.sampling_program.list_vars():
if i.name in self.sampling_vars:
np_data = np.array(global_scope().find_var(i.name).get_tensor())
if i.name not in self._sampling_data:
self._sampling_data[i.name] = []
self._sampling_data[i.name].append(np_data)
def __save_offline_model(self):
'''
Save the quantized model to the disk.
'''
io.save_inference_model(self.output, self.feed_var_names,
self.fetch_list, self.exe,
self.sampling_program)
def __display_debug(self):
if self.debug:
self.__dot(self._output_program)
print(self._output_program)
def __get_max_range_by_var_name(self, program, var_name):
"""
Check the specified variable was generated from Relu layer or not.
If the variable was the output of one of the pool2d/reshape/concat
/transpose, we keep trace the ancestor of this variable;
If the variable was the output the conv op, we check it's has_relu
attr;
Otherwise, we return the Calibrator.s8 as default value.
Returns:
Return Calibrator.u8_max if the variable was generated by Relu,
otherwise it will returns Calibrator.s8
"""
search_end_index = -1
input_index_name = {}
output_index_name = {}
ops_type = []
for index, op in enumerate(program.current_block().ops):
ops_type.append(op.type)
input_index_name[index] = op.input_arg_names
output_index_name[index] = op.output_arg_names
if var_name in op.output_arg_names:
search_end_index = index
# analysis
while search_end_index >= 0:
if ops_type[search_end_index] == "relu":
return Calibrator.u8_max
input_name = input_index_name[search_end_index][0]
for i in output_index_name.keys():
if input_name in output_index_name[i]:
search_end_index = i
break
if ops_type[
search_end_index] not in Calibrator.const_sign_op_type and ops_type[
search_end_index] != 'conv2d':
return Calibrator.s8_max
if ops_type[search_end_index] != 'conv2d':
continue
if program.current_block().ops[search_end_index].has_attr(
'fuse_relu') and program.current_block().ops[
search_end_index].attr('fuse_relu'):
return Calibrator.u8_max
else:
return Calibrator.s8_max
return Calibrator.s8_max
def __check_op_type_with_specified_var_as_input(self,
program,
var_name,
start_index=0):
'''
Check whether all the type of ops that use the specified variable as the
input.If one of those op is not int8-enabled, return False.
'''
op_type_list = [
op.type for op in program.current_block().ops[start_index:]
if var_name in op.input_arg_names
]
for i in op_type_list:
if not i in Calibrator.supported_int8_op_type:
return False
return True
def __check_var_source_dt(self, var_name):
'''
Check whether the specified variable is the output of int8 conv op or not.
If true, return the original op index.
If false, return -1
'''
return self._int8_output_var_op_index_dict[
var_name] if var_name in self._int8_output_var_op_index_dict else -1
def __update_int8_output_var_op_index_dict(self, index, var_name=None):
'''
Update the int8_output_variable/op_index dictionary
'''
for k, v in self._int8_output_var_op_index_dict.items():
if v >= index:
self._int8_output_var_op_index_dict[k] = v + 1
if var_name:
self._int8_output_var_op_index_dict[var_name] = index
def __update_program(self):
'''
Update the program with the quantize/dequantize op insertion.
'''
quantize_index, dequantize_index = self.__get_quantize_dequantize_combination(
self._output_program)
inserted_op_length = 0
calc_max_func = self.__get_optimal_scaling_factor if self.algo == "KL" else np.max
insert_op_collection = sorted(quantize_index + dequantize_index)
for index in insert_op_collection:
if index in quantize_index:
quantize_tmp = self._output_program.current_block().create_var(
name="quantize_{}_tmp".format(index),
dtype=core.VarDesc.VarType.UINT8)
original_out_name = self._output_program.current_block().ops[
index + inserted_op_length - 1].output_names[0]
original_out = self._output_program.current_block().ops[
index + inserted_op_length - 1].output(original_out_name)[0]
op = self._output_program.current_block()._insert_op(
index=index + inserted_op_length,
type="quantize",
inputs={"Input": original_out},
outputs={"Output": quantize_tmp}, )
op._set_attr("data_format", "MKLDNNLAYOUT")
op._set_attr("use_mkldnn", 1)
op._set_attr(
"Scale", self._var_max_range[original_out] /
calc_max_func(self._var_max_value_map[original_out]))
if self.__get_max_range_by_var_name(
self._output_program,
original_out) == Calibrator.s8_max:
op._set_attr("is_negative_input", 1)
self.__update_int8_output_var_op_index_dict(
index + inserted_op_length, "quantize_{}_tmp".format(index))
inserted_op_length += 1
for op in self._output_program.current_block().ops[
index + inserted_op_length:]:
for j in op.input_names:
if op.input(j) and op.input(
j
)[0] == original_out and op.type in Calibrator.supported_int8_op_type:
op.desc.set_input(j,
["{}".format(quantize_tmp.name)])
else:
start_index = index + inserted_op_length
dequantize_tmp_var = self._output_program.current_block(
).create_var(
name="dequantize_{}_tmp".format(index + 1),
dtype="float32", )
original_out_var = None
for original_input in self._output_program.current_block().ops[
start_index].input_arg_names:
index_res = self.__get_op_index_by_output_var(
self._output_program, original_input)
if index_res != -1:
original_out_var = original_input
break
if original_out_var:
op = self._output_program.current_block()._insert_op(
index=start_index,
type="dequantize",
inputs={"Input": original_out_var},
outputs={"Output": dequantize_tmp_var})
op._set_attr("data_format", "MKLDNNLAYOUT")
op._set_attr("use_mkldnn", 1)
op._set_attr("Scale", self._var_max_range[original_out_var]
/ calc_max_func(self._var_max_value_map[
original_out_var]))
for op_index in range(
start_index + 1,
len(self._output_program.current_block().ops)):
if self._output_program.current_block(
).ops[op_index].type == "conv2d" and self._output_program.current_block(
).ops[op_index].attr("force_fp32_output"):
continue
else:
for j in self._output_program.current_block().ops[
op_index].input_names:
if len(self._output_program.current_block().ops[
op_index].input(j)
) and self._output_program.current_block(
).ops[op_index].input(j)[
0] == original_out_var:
self._output_program.current_block(
).ops[op_index].desc.set_input(
j,
["{}".format(dequantize_tmp_var.name)])
inserted_op_length += 1
op._set_attr("data_format", "MKLDNNLAYOUT")
op._set_attr("use_mkldnn", 1)
def __update_output_program_attr(self):
for i in self._output_program.list_vars():
if i.name in self._persistable_vars:
i.persistable = False
os.system("rm -rf {}/{}".format(self.pretrained_model, i.name))
for i in self._u8_output_var:
self._output_program.current_block().var(i).desc.set_dtype(
core.VarDesc.VarType.UINT8)
for i in self._s8_output_var:
self._output_program.current_block().var(i).desc.set_dtype(
core.VarDesc.VarType.INT8)
@property
def sampling_program(self):
return self._output_program
@property
def sampling_vars(self):
return self._weights_var_name + self._conv_input_var_name + self._conv_output_var_name + self._residual_input_var_name + self._pool2d_output_var_name
def _is_close(self, a, b, rel_tol=1e-09, abs_tol=0.0):
return abs(a - b) <= max(rel_tol * max(abs(a), abs(b)), abs_tol)
def __generate_output_program(self):
for i in self.program.list_vars():
if not i.persistable and i.name in self.sampling_vars:
i.persistable = True
self._persistable_vars.append(i.name)
self._output_program = self.program.clone()
def __save_scale(self):
'''
Update the convolution scale information.
'''
func = self.__get_optimal_scaling_factor if self.algo == 'KL' else np.max
for i in self._conv_op_index[1:]:
weights_var_name = self.program.current_block().ops[i].input(
'Filter')[0]
input_var_name = self.program.current_block().ops[i].input('Input')[
0]
output_var_name = self.program.current_block().ops[i].output(
'Output')[0]
self._output_program.current_block().ops[i]._set_attr(
"Scale_weights", self._weights_scaling_factor[weights_var_name])
self._output_program.current_block().ops[i]._set_attr(
"Scale_in", self._var_max_range[input_var_name] /
func(self._var_max_value_map[input_var_name]))
self._output_program.current_block().ops[i]._set_attr(
"Scale_out", self._var_max_range[output_var_name] /
func(self._var_max_value_map[output_var_name]))
if self._output_program.current_block().ops[i].desc.input(
"ResidualData"):
residual_var_name = self._output_program.current_block().ops[
i].desc.input("ResidualData")[0]
self._output_program.current_block().ops[i]._set_attr(
"Scale_in_eltwise", self._var_max_range[residual_var_name] /
func(self._var_max_value_map[residual_var_name]))
def __sampling(self, sampling_data):
'''
Sampling the variables data range.
'''
for i in self.program.list_vars():
if i.name not in self.sampling_vars:
continue
if i.name in self._weights_var_name:
scaling_factor_per_channel = []
data = sampling_data[i.name][0]
for j in range(data.shape[0]):
var_value = float(np.max(np.abs(data[j])))
if not self._is_close(var_value, 0.0):
scaling_factor_per_channel.append(Calibrator.s8_max /
var_value)
else:
scaling_factor_per_channel.append(0.0)
self._weights_scaling_factor[
i.name] = scaling_factor_per_channel
else:
if i.name in self._conv_output_var_name:
op_pos = self.__get_op_index_by_output_var(self.program,
i.name)
cur_op = self.program.current_block().ops[op_pos]
if cur_op.has_attr('fuse_relu') and cur_op.attr(
'fuse_relu'):
max_range = Calibrator.u8_max
self._u8_output_var.append(i.name)
else:
max_range = Calibrator.s8_max
self._s8_output_var.append(i.name)
else:
max_range = self.__get_max_range_by_var_name(self.program,
i.name)
max_value = [[np.abs(np_data)]
for np_data in sampling_data[i.name]]
self._var_max_range[i.name] = max_range
self._var_max_value_map[i.name] = max_value
def __check_force_fp32_attr_by_output_var(self, program, var_name):
for op in program.current_block().ops:
if op.type == "conv2d" and var_name in op.output_arg_names:
return op.attr("force_fp32_output")
return False
def __get_op_index_by_output_var(self, program, var_name, start_index=0):
'''
Check whether the specified input variable is the output of the
conv/pool2d op's output or not.
Returns:
The index if the variable is the output of any conv/pool2d op's
output.
-1 when the variable is not the output of any conv/pool2d op's
output.
'''
for index, op in enumerate(program.current_block().ops[start_index:]):
if var_name in op.output_arg_names and op.type in Calibrator.supported_int8_op_type:
return index
return -1
def __get_op_index_by_input_var(self, program, var_name, start_index=0):
'''
Get the op index by specified input variable.
Returns:
The op index if the variable is the input of this op or -1 if the
variable is not the input of any op.
'''
for index, op in enumerate(program.current_block().ops[start_index:]):
if var_name in op.input_arg_names:
return index
return -1
def __get_quantize_dequantize_combination(self, program):
"""
Get the quantize/dequantize op index for further inserting.
Args:
The program desc.
Returns:
Two lists contains the quantize op and dequantize op index information.
"""
quantize_op_index = []
dequantize_op_index = []
minimal_conv_count = 2 # there must be two conv ops if not enable the first conv int8.
if len(self._conv_op_index) < minimal_conv_count:
return [], []
for index, value in enumerate(self._conv_op_index):
if index == 0:
quantize_op_index.append(self._conv_op_index[index + 1])
elif index == len(self._conv_op_index) - 1:
output_var = program.current_block().ops[value].output(
"Output")[0]
if self.__check_op_type_with_specified_var_as_input(
program, output_var, index):
dequantize_op_index.append(self._conv_op_index[index] + 2)
else:
program.current_block().ops[value]._set_attr(
"force_fp32_output", True)
elif self._conv_op_index[index] + 1 < self._conv_op_index[index +
1]:
program.current_block().ops[self._conv_op_index[
index]]._set_attr("force_fp32_output", True)
for op_index in range(self._conv_op_index[index + 1],
self._conv_op_index[index], -1):
op_type = program.current_block().ops[op_index].type
op_has_int8_input = False
input_var_name = None
input_length = len(program.current_block().ops[op_index]
.input_arg_names)
for var_name in program.current_block().ops[
op_index].input_arg_names:
if self.__check_var_source_dt(var_name) != -1:
op_has_int8_input = True
input_var_name = var_name
break
if op_has_int8_input:
if op_type == "conv2d":
if program.current_block().ops[op_index +
1].type == "conv2d":
continue
elif program.current_block(
).ops[op_index +
1].type in Calibrator.non_conv_int8_op_type:
dequantize_op_index.append(op_index + 2)
break
else:
program.current_block().ops[op_index]._set_attr(
"force_fp32_output", True)
continue
elif not self.__check_force_fp32_attr_by_output_var(
program, input_var_name
) and op_index not in dequantize_op_index:
share_input_flag = True
for input_attr_name in program.current_block().ops[
op_index].input_names:
input_var_name = program.current_block().ops[
op_index].input(input_attr_name)[0]
cousin_op_index = self.__get_op_index_by_input_var(
program, input_var_name)
if cousin_op_index != -1 and cousin_op_index in dequantize_op_index:
share_input_flag = False
break
if share_input_flag:
dequantize_op_index.append(op_index)
elif input_length:
output_is_to_int8_op = False
share_input_flag = True
for var_name in program.current_block().ops[
op_index].input_arg_names:
if not self.__check_op_type_with_specified_var_as_input(
program, var_name):
share_input_flag = False
break
for var_name in program.current_block().ops[
op_index].output_arg_names:
if self.__get_op_index_by_output_var(
program, var_name, op_index) != -1:
output_is_to_int8_op = True
break
if share_input_flag or output_is_to_int8_op:
quantize_op_index.append(op_index)
return quantize_op_index, dequantize_op_index
def __init_analysis(self):
'''
Collect the variable names for sampling.
'''
start_index = 1 #analysis the conv op detail from second conv op.
for i in self._conv_op_index[start_index:]:
self._weights_var_name.append(self.program.current_block().ops[i]
.input('Filter')[0])
self._conv_input_var_name.append(self.program.current_block().ops[i]
.input('Input')[0])
self._conv_output_var_name.append(self.program.current_block().ops[
i].output('Output')[0])
self._int8_output_var_op_index_dict[self.program.current_block()
.ops[i].output('Output')[0]] = i
if self.program.current_block().ops[i].desc.input("ResidualData"):
self._residual_input_var_name.append(self.program.current_block(
).ops[i].desc.input("ResidualData")[0])
if self.program.current_block().ops[i + 1].type == "pool2d":
self._pool2d_output_var_name.append(self.program.current_block(
).ops[i + 1].output('Out')[0])
def __expand_quantized_bins(self, quantized_bins, reference_bins):
expanded_quantized_bins = [0] * len(reference_bins)
num_merged_bins = len(reference_bins) / len(quantized_bins)
j_start = 0
j_end = num_merged_bins
for idx in xrange(len(quantized_bins)):
zero_count = reference_bins[j_start:j_end].count(0)
num_merged_bins = j_end - j_start
if zero_count == num_merged_bins:
avg_bin_ele = 0
else:
avg_bin_ele = quantized_bins[idx] / (
num_merged_bins - zero_count + 0.0)
for idx1 in xrange(j_start, j_end):
expanded_quantized_bins[idx1] = (0 if reference_bins[idx1] == 0
else avg_bin_ele)
j_start += num_merged_bins
j_end += num_merged_bins
if (idx + 1) == len(quantized_bins) - 1:
j_end = len(reference_bins)
return expanded_quantized_bins
def __safe_entropy(self, reference_distr_P, P_sum, candidate_distr_Q,
Q_sum):
'''
Calculate the entropy.
'''
assert len(reference_distr_P) == len(candidate_distr_Q)
tmp_sum1 = 0
tmp_sum2 = 0
for idx in range(len(reference_distr_P)):
p_idx = reference_distr_P[idx]
q_idx = candidate_distr_Q[idx]
if p_idx == 0:
tmp_sum1 += 0
tmp_sum2 += 0
else:
if q_idx == 0:
print("Fatal error!, idx = " + str(idx) +
" qindex = 0! p_idx = " + str(p_idx))
tmp_sum1 += p_idx * (math.log(Q_sum * p_idx))
tmp_sum2 += p_idx * (math.log(P_sum * q_idx))
return (tmp_sum1 - tmp_sum2) / P_sum
# Reference: http://on-demand.gputechconf.com/gtc/2017/presentation/s7310-8-bit-inference-with-tensorrt.pdf
def __get_optimal_scaling_factor(self,
activation_blob,
num_quantized_bins=255):
'''
Using the KL-divergenc method to get the more precise scaling factor.
'''
max_val = np.max(activation_blob)
min_val = np.min(activation_blob)
if min_val >= 0:
hist, hist_edeges = np.histogram(
activation_blob, bins=2048, range=(min_val, max_val))
ending_iter = 2047
starting_iter = int(ending_iter * 0.7)
else:
th = max(abs(max_val), abs(min_val))
hist, hist_edeges = np.histogram(
activation_blob, bins=2048, range=(-th, th))
starting_iter = 0
ending_iter = 2047
if abs(max_val) > abs(min_val):
while starting_iter < ending_iter:
if hist[starting_iter] == 0:
starting_iter += 1
continue
else:
break
starting_iter += int((ending_iter - starting_iter) * 0.6)
else:
while ending_iter > 0:
if hist[ending_iter] == 0:
ending_iter -= 1
continue
else:
break
starting_iter = int(0.6 * ending_iter)
bin_width = hist_edeges[1] - hist_edeges[0]
P_sum = len(activation_blob)
min_kl_divergence = 0
min_kl_index = 0
kl_inited = False
for i in range(starting_iter, ending_iter + 1):
reference_distr_P = hist[0:i].tolist()
outliers_count = sum(hist[i:2048])
if reference_distr_P[i - 1] == 0:
continue
reference_distr_P[i - 1] += outliers_count
reference_distr_bins = reference_distr_P[:]
candidate_distr_Q = hist[0:i].tolist()
num_merged_bins = i / num_quantized_bins
candidate_distr_Q_quantized = [0] * num_quantized_bins
j_start = 0
j_end = num_merged_bins
for idx in xrange(num_quantized_bins):
candidate_distr_Q_quantized[idx] = sum(candidate_distr_Q[
j_start:j_end])
j_start += num_merged_bins
j_end += num_merged_bins
if (idx + 1) == num_quantized_bins - 1:
j_end = i
candidate_distr_Q = self.__expand_quantized_bins(
candidate_distr_Q_quantized, reference_distr_bins)
Q_sum = sum(candidate_distr_Q)
kl_divergence = self.__safe_entropy(reference_distr_P, P_sum,
candidate_distr_Q, Q_sum)
if not kl_inited:
min_kl_divergence = kl_divergence
min_kl_index = i
kl_inited = True
elif kl_divergence < min_kl_divergence:
min_kl_divergence = kl_divergence
min_kl_index = i
else:
pass
if min_kl_index == 0:
while starting_iter > 0:
if hist[starting_iter] == 0:
starting_iter -= 1
continue
else:
break
min_kl_index = starting_iter
return (min_kl_index + 0.5) * bin_width
@staticmethod
def __dot(program, output_name="model.dot"):
'''
Generate the graphiz dot file for debugging.
'''
dot_graph = ""
dot_nodes = []
dot_edges = []
dot_graph += "digraph pm {\n"
for block in program.blocks:
ops = list(block.ops)
for index, op in enumerate(ops):
op_type = op.type
op_name = op_type + "_" + op.output_arg_names[0].replace(
".", "_") + "___" + str(index)
for name in op.input_arg_names:
name = name.replace(".", "_")
dot_edge = name + " -> " + op_name
if dot_edge not in dot_edges:
dot_edges.append(dot_edge)
dot_node = name + " [shape=oval, style=filled, fillcolor=yellow]"
if dot_node not in dot_nodes:
dot_nodes.append(dot_node)
for name in op.output_arg_names:
name = name.replace(".", "_")
dot_edge = op_name + " -> " + name
if dot_edge not in dot_edges:
dot_edges.append(dot_edge)
if op_type in Calibrator.supported_int8_op_type:
if op_type == "conv2d" and op.has_attr(
'force_fp32_output') and op.attr(
"force_fp32_output"):
dot_node = op_name + " [shape=box, style=filled, color=deeppink]"
else:
dot_node = op_name + " [shape=box, style=filled, color=greenyellow]"
elif op_type in ["quantize", "dequantize"]:
dot_node = op_name + " [shape=box, style=filled, color=gold]"
else:
dot_node = op_name + " [shape=box, style=filled, fillcolor=red]"
if dot_node not in dot_nodes:
dot_nodes.append(dot_node)
for dot_edge in dot_edges:
dot_graph += dot_edge + "\n"
for dot_node in dot_nodes:
dot_graph += dot_node + "\n"
dot_graph += "}"
with open(output_name, 'w') as f:
f.write(dot_graph)