sparse_lookup.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557
  1. ## @package sparse_lookup
  2. # Module caffe2.python.layers.sparse_lookup
  3. from caffe2.python.optimizer import FP16_ENGINES, Optimizer
  4. from caffe2.python.helpers.arg_scope import get_current_scope
  5. from caffe2.python import schema
  6. from caffe2.python.layers.layers import (
  7. get_categorical_limit,
  8. get_key,
  9. IdList,
  10. IdScoreList,
  11. IdListWithEvicted,
  12. IdScoreListWithEvicted,
  13. LayerPsParam,
  14. ModelLayer,
  15. almost_equal_schemas,
  16. )
  17. import collections
  18. import functools
  19. import logging
  20. import math
  21. import numpy as np
  22. import operator
  23. logger = logging.getLogger(__name__)
  24. def get_trainer_version_based_on_optim(optim_def):
  25. if isinstance(optim_def, Optimizer) and hasattr(optim_def, "engine"):
  26. logger.info(
  27. "Attempting to set trainer version for engine {}".format(optim_def.engine)
  28. )
  29. if optim_def.engine in FP16_ENGINES:
  30. logger.info("Setting FP16 trainer for engine {}".format(optim_def.engine))
  31. return "fp16"
  32. else:
  33. logger.info("Setting FP32 trainer for engine {}".format(optim_def.engine))
  34. return "fp32"
  35. else:
  36. return "fp32"
  37. def get_sparse_lookup_predictor_version(
  38. version,
  39. blob_size=None,
  40. min_blob_size_4bits=None,
  41. embedding_dim=None,
  42. sparse_feature_name=None,
  43. ):
  44. assert version in {
  45. 'fp32', 'fp16', 'uint8rowwise', 'fused_uint8rowwise', 'fused_uint4rowwise'
  46. }, "Unexpected version of sparse_lookup layer {0}".format(version)
  47. if version == 'fused_uint4rowwise':
  48. if (
  49. blob_size is not None
  50. and min_blob_size_4bits is not None
  51. and embedding_dim is not None
  52. ):
  53. if blob_size < min_blob_size_4bits:
  54. logger.info(
  55. "{} fall back to uint8 because lookup table size {} < min_blob_size_4bits {}".format(
  56. sparse_feature_name,
  57. blob_size,
  58. min_blob_size_4bits,
  59. )
  60. )
  61. version = 'fused_uint8rowwise'
  62. if embedding_dim % 2 == 1:
  63. logger.info(
  64. "{} fall back to uint8 because lookup table dimension {} is not divisible by 2".format(
  65. sparse_feature_name, embedding_dim
  66. )
  67. )
  68. version = 'fused_uint8rowwise'
  69. else:
  70. raise ValueError(
  71. (
  72. "When 4 bit quantization is enabled for {}, "
  73. "(i.e., Sparse lookup predictor version:{}), "
  74. "requires arguments blob_size:{}, "
  75. "min_blob_size_4bits:{}, embedding_dim:{}"
  76. ).format(
  77. sparse_feature_name,
  78. version,
  79. blob_size,
  80. min_blob_size_4bits,
  81. embedding_dim
  82. )
  83. )
  84. return version
  85. def get_sparse_lookup_trainer_version(version):
  86. assert version in {'fp32', 'fp16'},\
  87. "Unexpected version of sparse_lookup layer {0}".format(version)
  88. return version
  89. def _is_id_list(input_record):
  90. return almost_equal_schemas(input_record, IdList)
  91. def _is_id_score_list(input_record):
  92. return almost_equal_schemas(input_record,
  93. IdScoreList,
  94. check_field_types=False)
  95. class SparseLookup(ModelLayer):
  96. _id_list_supported_reducers = [
  97. 'LogMeanExp', 'LogSumExp', 'Max', 'Mean', 'Sum',
  98. 'WeightedSum', 'WeightedMean', 'Sqrt', 'None']
  99. _id_score_list_supported_reducers = [
  100. 'PositionWeighted', 'RecencyWeighted', 'Mean', 'Sum', 'WeightedSum',
  101. 'WeightedMean', 'None'
  102. ]
  103. _fp16_compatible_init_op_types = [
  104. 'Float16UniformFill'
  105. ]
  106. _fp16_compatible_reducers = [
  107. 'Sum', 'Mean', 'Sqrt', 'PositionWeighted', 'RecencyWeighted',
  108. ]
  109. def __init__(self, model, input_record, inner_shape, reducer,
  110. weight_init=None, weight_optim=None,
  111. name='sparse_lookup', regularizer=None, use_external_weights=False,
  112. uniform_weight_init_scale_numerator=1.0, **kwargs):
  113. super(SparseLookup, self).__init__(model, name, input_record, **kwargs)
  114. self.sparse_key = get_key(self.input_record)()
  115. logger.info("Setup the sparse lookup layer for " + self.sparse_key)
  116. # TODO Add some asserts about input type
  117. if isinstance(inner_shape, int):
  118. inner_shape = [inner_shape]
  119. assert isinstance(inner_shape, list) or isinstance(inner_shape, tuple),\
  120. "Unexpected type for inner_shape, expected list or tuple, got {0} for {1}".\
  121. format(type(inner_shape), self.sparse_key)
  122. if reducer == "PositionWeighted":
  123. assert _is_id_score_list(self.input_record), (
  124. "PositionWeighted only support IdScoreList, but got {} for {}"
  125. + "please use PositionWeighted layer to convert IdList "
  126. + "to IdScoreList"
  127. ).format(repr(self.input_record), self.sparse_key)
  128. self.external_weights = self.input_record.values()
  129. elif reducer == "RecencyWeighted":
  130. assert _is_id_score_list(self.input_record), (
  131. "RecencyWeighted only supports IdScoreList, "
  132. "while the sparse feature {} is not.".format(self.sparse_key)
  133. )
  134. self.external_weights = self.input_record.values()
  135. # TODO: create a new type of reducer with external weights to wrap
  136. # this and the above two cases since essentially their input formats
  137. # are the same.
  138. elif use_external_weights:
  139. assert _is_id_score_list(self.input_record), (
  140. "Use_external_weights only supports IdScoreList, "
  141. "while the sparse feature {} is not.".format(self.sparse_key)
  142. )
  143. assert reducer in ["Sum", "WeightedSum"], (
  144. "Use_external_weights only supports Sum reducer, "
  145. "while the reducer is {}.".format(reducer)
  146. )
  147. self.external_weights = self.input_record.values()
  148. self.reducer = reducer
  149. self.use_external_weights = use_external_weights
  150. input_dim = get_categorical_limit(self.input_record)
  151. assert input_dim > 0, "{} should have categorical limit > 0, but got {}".format(
  152. self.sparse_key, input_dim
  153. )
  154. self.input_dim = input_dim
  155. self.shape = [input_dim] + inner_shape
  156. self.trainer_version = get_trainer_version_based_on_optim(
  157. weight_optim
  158. )
  159. self.uniform_weight_init_scale_numerator = uniform_weight_init_scale_numerator
  160. default_init_op = self._get_default_init_op()
  161. self.weight_init = weight_init or default_init_op
  162. self.evicted_values = None
  163. if schema.equal_schemas(
  164. self.input_record, IdListWithEvicted
  165. ) or schema.equal_schemas(
  166. self.input_record, IdScoreListWithEvicted, check_field_types=False
  167. ):
  168. self.evicted_values = self.input_record._evicted_values
  169. # If fp16 is used, make sure fp16 init op is used
  170. if self.trainer_version == "fp16":
  171. assert self.reducer in self._fp16_compatible_reducers or use_external_weights, (
  172. "Fp16 training is enabled. The reducer specified is not supported. "
  173. "Got {}. Supported reducers: {}. Right now, in general, sum, mean, "
  174. "positional pooling are supported. Attention is not. Please check "
  175. "if there is fp16 trained sparse features using advanced pooling.".format(
  176. self.reducer, self._fp16_compatible_reducers)
  177. )
  178. # if init op is UniformFill, we replace it directly
  179. if self.weight_init[0] == "UniformFill":
  180. self.weight_init = ("Float16UniformFill", self.weight_init[1])
  181. assert self.weight_init[0] in self._fp16_compatible_init_op_types, (
  182. "Fp16 training is enabled. Init op for weight parameter must be fp16 "
  183. "compatibale. Got {}. Supported ops: {}".format(
  184. self.weight_init[0],
  185. self._fp16_compatible_init_op_types)
  186. )
  187. assert regularizer is None, "Regularizer is not compatible with fp16"
  188. if self.input_record.lengths.metadata:
  189. avg_length = self.input_record.lengths.metadata.expected_value
  190. else:
  191. avg_length = None
  192. self.w = self.create_param(
  193. param_name='w',
  194. shape=self.shape,
  195. initializer=self.weight_init,
  196. optimizer=weight_optim,
  197. ps_param=LayerPsParam(
  198. sparse_key=self.sparse_key,
  199. average_length=avg_length),
  200. regularizer=regularizer
  201. )
  202. if self.evicted_values:
  203. self.reinit_vec = self.create_param(
  204. param_name="reinit_vec",
  205. shape=inner_shape,
  206. initializer=self.weight_init,
  207. optimizer=model.NoOptim,
  208. regularizer=None,
  209. )
  210. self.scale_bias_init = ('ConstantFill', {'value': 0.0})
  211. self.scale_bias = self.create_param(
  212. param_name='scale_bias',
  213. shape=[],
  214. initializer=self.scale_bias_init,
  215. optimizer=model.NoOptim,
  216. )
  217. self.output_schema = schema.Scalar(
  218. (np.float32, inner_shape),
  219. self.get_next_blob_reference('output'),
  220. )
  221. def get_memory_usage(self):
  222. return functools.reduce(operator.mul, self.shape) * 4
  223. def get_fp16_compatible_parameters(self):
  224. return [self.w]
  225. def support_8bit(self):
  226. # Rowwise quantization makes sense only if shape it's 2D matrix with
  227. # second dimension >= 8
  228. if len(self.shape) != 2 or self.shape[1] < 8:
  229. return False
  230. return True
  231. def get_8bits_compatible_parameters(self, fused=True):
  232. if not self.support_8bit():
  233. return []
  234. if fused:
  235. RowwiseQuantized8BitsWeight = collections.namedtuple(
  236. 'RowwiseQuantized8BitsWeight', 'w'
  237. )
  238. return [RowwiseQuantized8BitsWeight(self.w)]
  239. else:
  240. RowwiseQuantized8BitsWeight = collections.namedtuple(
  241. 'RowwiseQuantized8BitsWeight', 'w, scale_bias'
  242. )
  243. return [RowwiseQuantized8BitsWeight(self.w, self.scale_bias)]
  244. def _get_default_init_op(self):
  245. scale = math.sqrt(self.uniform_weight_init_scale_numerator / self.input_dim)
  246. if self.trainer_version == 'fp32':
  247. default_weight_init = ('UniformFill', {'min': -scale, 'max': scale})
  248. elif self.trainer_version == 'fp16':
  249. default_weight_init = ("Float16UniformFill", {'min': -scale, 'max': scale})
  250. else:
  251. raise NotImplementedError(
  252. "Train version {} is not currently supported for sparse feature {}".format(
  253. trainer_version, self.sparse_key
  254. )
  255. )
  256. return default_weight_init
  257. def _gather_wrapper(self, net, version, in_indices, out):
  258. # Gather can work on all kinds of input data types, and output
  259. # data with the same type. Convert the output of Gather to float,
  260. # because the follow-up Ops expect fp32.
  261. if version == 'fp32':
  262. return net.Gather([self.w, in_indices], out)
  263. elif version == 'fp16':
  264. gathered_w = net.Gather([self.w, in_indices], 'gathered_w')
  265. return net.HalfToFloat(gathered_w, out)
  266. elif version == 'uint8rowwise':
  267. gathered_w = net.Gather([self.w, in_indices], 'gathered_w')
  268. gathered_scale_bias = net.Gather(
  269. [self.scale_bias, in_indices],
  270. 'gathered_scale_bias'
  271. )
  272. return net.Rowwise8BitQuantizedToFloat(
  273. [gathered_w, gathered_scale_bias], out)
  274. elif version == 'fused_uint8rowwise':
  275. gathered_w = net.Gather([self.w, in_indices], 'gathered_w')
  276. return net.Fused8BitRowwiseQuantizedToFloat(gathered_w, out)
  277. elif version == 'fused_uint4rowwise':
  278. gathered_w = net.Gather([self.w, in_indices], 'gathered_w')
  279. return net.Fused4BitRowwiseQuantizedToFloat(gathered_w, out)
  280. else:
  281. raise "Unsupported version of operators in SparseLookup " +\
  282. "layer: {0} for sparse feature {1}".format(
  283. version, self.sparse_key
  284. )
  285. def _sparse_lengths_weighted_reducer(
  286. self,
  287. in_indices,
  288. weights,
  289. reducer,
  290. net,
  291. version,
  292. grad_on_weights=0,
  293. ):
  294. op_input = [
  295. self.w,
  296. weights,
  297. in_indices,
  298. self.input_record.lengths(),
  299. ]
  300. layer_name = 'SparseLengths' + reducer
  301. if version in ['fp32', 'fp16']:
  302. # SparseLengths* Ops will accept either fp16 or fp32 embedding
  303. # matrix and output fp32 pooled embedding
  304. # A special case here is that we need FP16 engine for
  305. # SparseLengthsWeightedSum when FP16 embeedings are used for
  306. # correct backward updates
  307. if reducer == "WeightedSum" and version == "fp16":
  308. net.SparseLengthsWeightedSum(
  309. op_input,
  310. self.output_schema.field_blobs(),
  311. grad_on_weights=grad_on_weights,
  312. engine='FP16',
  313. )
  314. else:
  315. net.__getattr__(layer_name)(
  316. op_input,
  317. self.output_schema.field_blobs(),
  318. grad_on_weights=grad_on_weights,
  319. )
  320. elif version == 'uint8rowwise':
  321. op_input.insert(len(op_input), self.scale_bias)
  322. net.__getattr__(layer_name + '8BitsRowwise')(
  323. op_input, self.output_schema.field_blobs())
  324. elif version == 'fused_uint8rowwise':
  325. net.__getattr__(layer_name + 'Fused8BitRowwise')(
  326. op_input, self.output_schema.field_blobs())
  327. elif version == 'fused_uint4rowwise':
  328. net.__getattr__(layer_name + 'Fused4BitRowwise')(
  329. op_input, self.output_schema.field_blobs())
  330. else:
  331. raise "Unsupported version of operator in SparseLookUp " +\
  332. "layer: {0} for sparse feature {1}".format(
  333. version, self.sparse_key
  334. )
  335. # deal with sparse features of id_list type
  336. def _add_ops_id_list(self, net, version):
  337. assert self.reducer in self._id_list_supported_reducers, (
  338. "Unsupported reducer: {} for ID_LIST {}".format(
  339. self.reducer, self.sparse_key
  340. )
  341. )
  342. if self.reducer in ['Sum', 'Mean', 'WeightedSum', 'WeightedMean']:
  343. op_input = [self.w,
  344. self.input_record.items(),
  345. self.input_record.lengths()]
  346. # For id list features, the behaviors of 'Sum' and
  347. # 'WeightedSum' are identical, since we can regard the weight on each
  348. # id as 1. Similarly, for 'Mean' and 'WeightedMean'.
  349. if self.reducer == 'WeightedSum':
  350. self.reducer = 'Sum'
  351. elif self.reducer == 'WeightedMean':
  352. self.reducer = 'Mean'
  353. layer_name = 'SparseLengths' + self.reducer
  354. if version in ['fp32', 'fp16']:
  355. # SparseLengths* Ops will accept either fp16 or fp32 embedding
  356. # matrix and output fp32 pooled embedding
  357. net.__getattr__(layer_name)(
  358. op_input,
  359. self.output_schema.field_blobs(),
  360. )
  361. elif version == 'uint8rowwise':
  362. op_input.insert(len(op_input), self.scale_bias)
  363. net.__getattr__(layer_name + '8BitsRowwise')(
  364. op_input, self.output_schema.field_blobs())
  365. elif version == 'fused_uint8rowwise':
  366. net.__getattr__(layer_name + 'Fused8BitRowwise')(
  367. op_input, self.output_schema.field_blobs())
  368. elif version == 'fused_uint4rowwise':
  369. net.__getattr__(layer_name + 'Fused4BitRowwise')(
  370. op_input, self.output_schema.field_blobs())
  371. else:
  372. raise "Unsupported version of operator in SparseLookUp " +\
  373. "layer: {0} for sparse feature {1}".format(
  374. version, self.sparse_key
  375. )
  376. elif self.reducer == 'Sqrt':
  377. sqrt_weight = net.LengthsToWeights(
  378. [self.input_record.lengths()],
  379. [net.NextScopedBlob('lengths_sqrt')],
  380. power=0.5,
  381. )
  382. self._sparse_lengths_weighted_reducer(
  383. self.input_record.items(),
  384. sqrt_weight,
  385. 'WeightedSum', net, version)
  386. elif self.reducer == 'None':
  387. # Gather operator will gather the embedding for each id of
  388. # each IdList.
  389. self._gather_wrapper(net, version, self.input_record.items(),
  390. self.output_schema.field_blobs())
  391. else:
  392. table_rows = self._gather_wrapper(
  393. net, version, self.input_record.items(), 'table_rows')
  394. segment_ids = net.LengthsToSegmentIds(
  395. self.input_record.lengths(),
  396. net.NextScopedBlob(self.input_record.lengths() + '_sid'))
  397. net.__getattr__('SortedSegmentRange' + self.reducer)(
  398. [table_rows, segment_ids],
  399. self.output_schema.field_blobs(),
  400. )
  401. # deal with sparse features of id_score_list type
  402. def _add_ops_id_score_list(self, net, version):
  403. assert self.reducer in self._id_score_list_supported_reducers, (
  404. "Unsupported reducer: {} for ID_SCORE_LIST {}".format(
  405. self.reducer, self.sparse_key
  406. )
  407. )
  408. if self.reducer in ['WeightedSum', 'WeightedMean']:
  409. self._sparse_lengths_weighted_reducer(
  410. self.input_record.keys(),
  411. self.input_record.values(),
  412. self.reducer, net, version)
  413. elif self.reducer in ['PositionWeighted', 'RecencyWeighted'] or self.use_external_weights:
  414. self._sparse_lengths_weighted_reducer(
  415. self.input_record.keys(),
  416. self.external_weights,
  417. 'WeightedSum', net, version, grad_on_weights=1)
  418. elif self.reducer in ['Sum', 'Mean']:
  419. op_input = [self.w,
  420. self.input_record.keys(),
  421. self.input_record.lengths()]
  422. layer_name = 'SparseLengths' + self.reducer
  423. if version in ['fp32', 'fp16']:
  424. net.__getattr__(layer_name)(
  425. op_input,
  426. self.output_schema.field_blobs(),
  427. )
  428. elif version == 'uint8rowwise':
  429. net.__getattr__(layer_name + '8BitsRowwise')(
  430. op_input, self.output_schema.field_blobs())
  431. elif version == 'fused_uint8rowwise':
  432. net.__getattr__(layer_name + 'Fused8BitRowwise')(
  433. op_input, self.output_schema.field_blobs())
  434. elif version == 'fused_uint4rowwise':
  435. net.__getattr__(layer_name + 'Fused4BitRowwise')(
  436. op_input, self.output_schema.field_blobs())
  437. else:
  438. raise "Unsupported version of operator in SparseLookUp " +\
  439. "layer: {0} for sparse feature {1}".format(
  440. version, self.sparse_key
  441. )
  442. elif self.reducer == 'None':
  443. # Gather operator will gather the embedding for each id of
  444. # each IdList.
  445. self._gather_wrapper(net, version, self.input_record.keys(),
  446. self.output_schema.field_blobs())
  447. else:
  448. raise "Only Sum, Mean, None are supported for IdScoreList input." +\
  449. "Trying to create with {} for sparse feature {}".format(
  450. self.reducer, self.sparse_key
  451. )
  452. def _add_ops(self, net, version='fp32', is_train=True):
  453. if self.evicted_values and is_train:
  454. net.CopyRowsToTensor(
  455. [self.w, self.evicted_values.get(), self.reinit_vec], [self.w])
  456. if _is_id_list(self.input_record):
  457. self._add_ops_id_list(net, version=version)
  458. elif _is_id_score_list(self.input_record):
  459. self._add_ops_id_score_list(net, version=version)
  460. else:
  461. raise "Unsupported input type {0}".format(self.input_record)
  462. def add_train_ops(self, net):
  463. self._add_ops(net, self.trainer_version, is_train=True)
  464. def add_ops(self, net):
  465. version_info = get_current_scope().get(
  466. get_sparse_lookup_predictor_version.__name__, {'version': 'fp32'}
  467. )
  468. lookup_table_blob_size = self.shape[0] * self.shape[1]
  469. version = get_sparse_lookup_predictor_version(
  470. version_info['version'],
  471. blob_size=lookup_table_blob_size,
  472. min_blob_size_4bits=(
  473. version_info['min_blob_size_4bits']
  474. if 'min_blob_size_4bits' in version_info
  475. else None
  476. ),
  477. embedding_dim=self.shape[1],
  478. sparse_feature_name=self.sparse_key,
  479. )
  480. # TODO(amalevich): Layer should not be responsible for decision about
  481. # quantization.
  482. if not self.support_8bit() and version in {'uint8rowwise',
  483. 'fused_uint8rowwise',
  484. 'fused_uint4rowwise'}:
  485. version = 'fp16'
  486. self._add_ops(net, version, is_train=False)