program.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449
  1. # copyright (c) 2020 PaddlePaddle Authors. All Rights Reserve.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from __future__ import absolute_import
  15. from __future__ import division
  16. from __future__ import print_function
  17. import os
  18. import time
  19. import numpy as np
  20. from collections import OrderedDict
  21. import paddle
  22. import paddle.nn.functional as F
  23. from paddle.distributed import fleet
  24. from paddle.distributed.fleet import DistributedStrategy
  25. # from paddlex.ppcls.optimizer import OptimizerBuilder
  26. # from paddlex.ppcls.optimizer.learning_rate import LearningRateBuilder
  27. from paddlex.ppcls.arch import build_model
  28. from paddlex.ppcls.loss import build_loss
  29. from paddlex.ppcls.metric import build_metrics
  30. from paddlex.ppcls.optimizer import build_optimizer
  31. from paddlex.ppcls.optimizer import build_lr_scheduler
  32. from paddlex.ppcls.utils.misc import AverageMeter
  33. from paddlex.ppcls.utils import logger, profiler
  34. def create_feeds(image_shape, use_mix=False, class_num=None, dtype="float32"):
  35. """
  36. Create feeds as model input
  37. Args:
  38. image_shape(list[int]): model input shape, such as [3, 224, 224]
  39. use_mix(bool): whether to use mix(include mixup, cutmix, fmix)
  40. class_num(int): the class number of network, required if use_mix
  41. Returns:
  42. feeds(dict): dict of model input variables
  43. """
  44. feeds = OrderedDict()
  45. feeds['data'] = paddle.static.data(
  46. name="data", shape=[None] + image_shape, dtype=dtype)
  47. if use_mix:
  48. if class_num is None:
  49. msg = "When use MixUp, CutMix and so on, you must set class_num."
  50. logger.error(msg)
  51. raise Exception(msg)
  52. feeds['target'] = paddle.static.data(
  53. name="target", shape=[None, class_num], dtype="float32")
  54. else:
  55. feeds['label'] = paddle.static.data(
  56. name="label", shape=[None, 1], dtype="int64")
  57. return feeds
  58. def create_fetchs(out,
  59. feeds,
  60. architecture,
  61. topk=5,
  62. epsilon=None,
  63. class_num=None,
  64. use_mix=False,
  65. config=None,
  66. mode="Train"):
  67. """
  68. Create fetchs as model outputs(included loss and measures),
  69. will call create_loss and create_metric(if use_mix).
  70. Args:
  71. out(variable): model output variable
  72. feeds(dict): dict of model input variables.
  73. If use mix_up, it will not include label.
  74. architecture(dict): architecture information,
  75. name(such as ResNet50) is needed
  76. topk(int): usually top5
  77. epsilon(float): parameter for label smoothing, 0.0 <= epsilon <= 1.0
  78. class_num(int): the class number of network, required if use_mix
  79. use_mix(bool): whether to use mix(include mixup, cutmix, fmix)
  80. config(dict): model config
  81. Returns:
  82. fetchs(dict): dict of model outputs(included loss and measures)
  83. """
  84. fetchs = OrderedDict()
  85. # build loss
  86. if use_mix:
  87. if class_num is None:
  88. msg = "When use MixUp, CutMix and so on, you must set class_num."
  89. logger.error(msg)
  90. raise Exception(msg)
  91. target = paddle.reshape(feeds['target'], [-1, class_num])
  92. else:
  93. target = paddle.reshape(feeds['label'], [-1, 1])
  94. loss_func = build_loss(config["Loss"][mode])
  95. loss_dict = loss_func(out, target)
  96. loss_out = loss_dict["loss"]
  97. fetchs['loss'] = (loss_out, AverageMeter('loss', '7.4f', need_avg=True))
  98. # build metric
  99. if not use_mix:
  100. metric_func = build_metrics(config["Metric"][mode])
  101. metric_dict = metric_func(out, target)
  102. for key in metric_dict:
  103. if mode != "Train" and paddle.distributed.get_world_size() > 1:
  104. paddle.distributed.all_reduce(
  105. metric_dict[key], op=paddle.distributed.ReduceOp.SUM)
  106. metric_dict[key] = metric_dict[
  107. key] / paddle.distributed.get_world_size()
  108. fetchs[key] = (metric_dict[key], AverageMeter(
  109. key, '7.4f', need_avg=True))
  110. return fetchs
  111. def create_optimizer(config, step_each_epoch):
  112. # create learning_rate instance
  113. optimizer, lr_sch = build_optimizer(
  114. config["Optimizer"], config["Global"]["epochs"], step_each_epoch)
  115. return optimizer, lr_sch
  116. def create_strategy(config):
  117. """
  118. Create build strategy and exec strategy.
  119. Args:
  120. config(dict): config
  121. Returns:
  122. build_strategy: build strategy
  123. exec_strategy: exec strategy
  124. """
  125. build_strategy = paddle.static.BuildStrategy()
  126. exec_strategy = paddle.static.ExecutionStrategy()
  127. exec_strategy.num_threads = 1
  128. exec_strategy.num_iteration_per_drop_scope = (
  129. 10000
  130. if 'AMP' in config and config.AMP.get("use_pure_fp16", False) else 10)
  131. fuse_op = True if 'AMP' in config else False
  132. fuse_bn_act_ops = config.get('fuse_bn_act_ops', fuse_op)
  133. fuse_elewise_add_act_ops = config.get('fuse_elewise_add_act_ops', fuse_op)
  134. fuse_bn_add_act_ops = config.get('fuse_bn_add_act_ops', fuse_op)
  135. enable_addto = config.get('enable_addto', fuse_op)
  136. build_strategy.fuse_bn_act_ops = fuse_bn_act_ops
  137. build_strategy.fuse_elewise_add_act_ops = fuse_elewise_add_act_ops
  138. build_strategy.fuse_bn_add_act_ops = fuse_bn_add_act_ops
  139. build_strategy.enable_addto = enable_addto
  140. return build_strategy, exec_strategy
  141. def dist_optimizer(config, optimizer):
  142. """
  143. Create a distributed optimizer based on a normal optimizer
  144. Args:
  145. config(dict):
  146. optimizer(): a normal optimizer
  147. Returns:
  148. optimizer: a distributed optimizer
  149. """
  150. build_strategy, exec_strategy = create_strategy(config)
  151. dist_strategy = DistributedStrategy()
  152. dist_strategy.execution_strategy = exec_strategy
  153. dist_strategy.build_strategy = build_strategy
  154. dist_strategy.nccl_comm_num = 1
  155. dist_strategy.fuse_all_reduce_ops = True
  156. dist_strategy.fuse_grad_size_in_MB = 16
  157. optimizer = fleet.distributed_optimizer(optimizer, strategy=dist_strategy)
  158. return optimizer
  159. def mixed_precision_optimizer(config, optimizer):
  160. if 'AMP' in config:
  161. amp_cfg = config.AMP if config.AMP else dict()
  162. scale_loss = amp_cfg.get('scale_loss', 1.0)
  163. use_dynamic_loss_scaling = amp_cfg.get('use_dynamic_loss_scaling',
  164. False)
  165. use_pure_fp16 = amp_cfg.get('use_pure_fp16', False)
  166. optimizer = paddle.static.amp.decorate(
  167. optimizer,
  168. init_loss_scaling=scale_loss,
  169. use_dynamic_loss_scaling=use_dynamic_loss_scaling,
  170. use_pure_fp16=use_pure_fp16,
  171. use_fp16_guard=True)
  172. return optimizer
  173. def build(config,
  174. main_prog,
  175. startup_prog,
  176. class_num=None,
  177. step_each_epoch=100,
  178. is_train=True,
  179. is_distributed=True):
  180. """
  181. Build a program using a model and an optimizer
  182. 1. create feeds
  183. 2. create a dataloader
  184. 3. create a model
  185. 4. create fetchs
  186. 5. create an optimizer
  187. Args:
  188. config(dict): config
  189. main_prog(): main program
  190. startup_prog(): startup program
  191. class_num(int): the class number of network, required if use_mix
  192. is_train(bool): train or eval
  193. is_distributed(bool): whether to use distributed training method
  194. Returns:
  195. dataloader(): a bridge between the model and the data
  196. fetchs(dict): dict of model outputs(included loss and measures)
  197. """
  198. with paddle.static.program_guard(main_prog, startup_prog):
  199. with paddle.utils.unique_name.guard():
  200. mode = "Train" if is_train else "Eval"
  201. use_mix = "batch_transform_ops" in config["DataLoader"][mode][
  202. "dataset"]
  203. feeds = create_feeds(
  204. config["Global"]["image_shape"],
  205. use_mix,
  206. class_num=class_num,
  207. dtype="float32")
  208. # build model
  209. # data_format should be assigned in arch-dict
  210. input_image_channel = config["Global"]["image_shape"][
  211. 0] # default as [3, 224, 224]
  212. model = build_model(config["Arch"])
  213. out = model(feeds["data"])
  214. # end of build model
  215. fetchs = create_fetchs(
  216. out,
  217. feeds,
  218. config["Arch"],
  219. epsilon=config.get('ls_epsilon'),
  220. class_num=class_num,
  221. use_mix=use_mix,
  222. config=config,
  223. mode=mode)
  224. lr_scheduler = None
  225. optimizer = None
  226. if is_train:
  227. optimizer, lr_scheduler = build_optimizer(
  228. config["Optimizer"], config["Global"]["epochs"],
  229. step_each_epoch)
  230. optimizer = mixed_precision_optimizer(config, optimizer)
  231. if is_distributed:
  232. optimizer = dist_optimizer(config, optimizer)
  233. optimizer.minimize(fetchs['loss'][0])
  234. return fetchs, lr_scheduler, feeds, optimizer
  235. def compile(config, program, loss_name=None, share_prog=None):
  236. """
  237. Compile the program
  238. Args:
  239. config(dict): config
  240. program(): the program which is wrapped by
  241. loss_name(str): loss name
  242. share_prog(): the shared program, used for evaluation during training
  243. Returns:
  244. compiled_program(): a compiled program
  245. """
  246. build_strategy, exec_strategy = create_strategy(config)
  247. compiled_program = paddle.static.CompiledProgram(
  248. program).with_data_parallel(
  249. share_vars_from=share_prog,
  250. loss_name=loss_name,
  251. build_strategy=build_strategy,
  252. exec_strategy=exec_strategy)
  253. return compiled_program
  254. total_step = 0
  255. def run(dataloader,
  256. exe,
  257. program,
  258. feeds,
  259. fetchs,
  260. epoch=0,
  261. mode='train',
  262. config=None,
  263. vdl_writer=None,
  264. lr_scheduler=None,
  265. profiler_options=None):
  266. """
  267. Feed data to the model and fetch the measures and loss
  268. Args:
  269. dataloader(paddle io dataloader):
  270. exe():
  271. program():
  272. fetchs(dict): dict of measures and the loss
  273. epoch(int): epoch of training or evaluation
  274. model(str): log only
  275. Returns:
  276. """
  277. fetch_list = [f[0] for f in fetchs.values()]
  278. metric_dict = OrderedDict([("lr", AverageMeter(
  279. 'lr', 'f', postfix=",", need_avg=False))])
  280. for k in fetchs:
  281. metric_dict[k] = fetchs[k][1]
  282. metric_dict["batch_time"] = AverageMeter(
  283. 'batch_cost', '.5f', postfix=" s,")
  284. metric_dict["reader_time"] = AverageMeter(
  285. 'reader_cost', '.5f', postfix=" s,")
  286. for m in metric_dict.values():
  287. m.reset()
  288. use_dali = config["Global"].get('use_dali', False)
  289. tic = time.time()
  290. if not use_dali:
  291. dataloader = dataloader()
  292. idx = 0
  293. batch_size = None
  294. while True:
  295. # The DALI maybe raise RuntimeError for some particular images, such as ImageNet1k/n04418357_26036.JPEG
  296. try:
  297. batch = next(dataloader)
  298. except StopIteration:
  299. break
  300. except RuntimeError:
  301. logger.warning(
  302. "Except RuntimeError when reading data from dataloader, try to read once again..."
  303. )
  304. continue
  305. idx += 1
  306. # ignore the warmup iters
  307. if idx == 5:
  308. metric_dict["batch_time"].reset()
  309. metric_dict["reader_time"].reset()
  310. metric_dict['reader_time'].update(time.time() - tic)
  311. profiler.add_profiler_step(profiler_options)
  312. if use_dali:
  313. batch_size = batch[0]["data"].shape()[0]
  314. feed_dict = batch[0]
  315. else:
  316. batch_size = batch[0].shape()[0]
  317. feed_dict = {
  318. key.name: batch[idx]
  319. for idx, key in enumerate(feeds.values())
  320. }
  321. metrics = exe.run(program=program,
  322. feed=feed_dict,
  323. fetch_list=fetch_list)
  324. for name, m in zip(fetchs.keys(), metrics):
  325. metric_dict[name].update(np.mean(m), batch_size)
  326. metric_dict["batch_time"].update(time.time() - tic)
  327. if mode == "train":
  328. metric_dict['lr'].update(lr_scheduler.get_lr())
  329. fetchs_str = ' '.join([
  330. str(metric_dict[key].mean)
  331. if "time" in key else str(metric_dict[key].value)
  332. for key in metric_dict
  333. ])
  334. ips_info = " ips: {:.5f} images/sec.".format(
  335. batch_size / metric_dict["batch_time"].avg)
  336. fetchs_str += ips_info
  337. if lr_scheduler is not None:
  338. lr_scheduler.step()
  339. if vdl_writer:
  340. global total_step
  341. logger.scaler('loss', metrics[0][0], total_step, vdl_writer)
  342. total_step += 1
  343. if mode == 'eval':
  344. if idx % config.get('print_interval', 10) == 0:
  345. logger.info("{:s} step:{:<4d} {:s}".format(mode, idx,
  346. fetchs_str))
  347. else:
  348. epoch_str = "epoch:{:<3d}".format(epoch)
  349. step_str = "{:s} step:{:<4d}".format(mode, idx)
  350. if idx % config.get('print_interval', 10) == 0:
  351. logger.info("{:s} {:s} {:s}".format(epoch_str, step_str,
  352. fetchs_str))
  353. tic = time.time()
  354. end_str = ' '.join([str(m.mean) for m in metric_dict.values()] +
  355. [metric_dict["batch_time"].total])
  356. ips_info = "ips: {:.5f} images/sec.".format(batch_size /
  357. metric_dict["batch_time"].avg)
  358. if mode == 'eval':
  359. logger.info("END {:s} {:s} {:s}".format(mode, end_str, ips_info))
  360. else:
  361. end_epoch_str = "END epoch:{:<3d}".format(epoch)
  362. logger.info("{:s} {:s} {:s} {:s}".format(end_epoch_str, mode, end_str,
  363. ips_info))
  364. if use_dali:
  365. dataloader.reset()
  366. # return top1_acc in order to save the best model
  367. if mode == 'eval':
  368. return fetchs["top1"][1].avg