dali.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. # Copyright (c) 2021 PaddlePaddle Authors. All Rights Reserved.
  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 division
  15. import copy
  16. import os
  17. import numpy as np
  18. import nvidia.dali.ops as ops
  19. import nvidia.dali.types as types
  20. import paddle
  21. from nvidia.dali import fn
  22. from nvidia.dali.pipeline import Pipeline
  23. from nvidia.dali.plugin.base_iterator import LastBatchPolicy
  24. from nvidia.dali.plugin.paddle import DALIGenericIterator
  25. class HybridTrainPipe(Pipeline):
  26. def __init__(self,
  27. file_root,
  28. file_list,
  29. batch_size,
  30. resize_shorter,
  31. crop,
  32. min_area,
  33. lower,
  34. upper,
  35. interp,
  36. mean,
  37. std,
  38. device_id,
  39. shard_id=0,
  40. num_shards=1,
  41. random_shuffle=True,
  42. num_threads=4,
  43. seed=42,
  44. pad_output=False,
  45. output_dtype=types.FLOAT,
  46. dataset='Train'):
  47. super(HybridTrainPipe, self).__init__(
  48. batch_size, num_threads, device_id, seed=seed)
  49. self.input = ops.readers.File(
  50. file_root=file_root,
  51. file_list=file_list,
  52. shard_id=shard_id,
  53. num_shards=num_shards,
  54. random_shuffle=random_shuffle)
  55. # set internal nvJPEG buffers size to handle full-sized ImageNet images
  56. # without additional reallocations
  57. device_memory_padding = 211025920
  58. host_memory_padding = 140544512
  59. self.decode = ops.decoders.ImageRandomCrop(
  60. device='mixed',
  61. output_type=types.DALIImageType.RGB,
  62. device_memory_padding=device_memory_padding,
  63. host_memory_padding=host_memory_padding,
  64. random_aspect_ratio=[lower, upper],
  65. random_area=[min_area, 1.0],
  66. num_attempts=100)
  67. self.res = ops.Resize(
  68. device='gpu', resize_x=crop, resize_y=crop, interp_type=interp)
  69. self.cmnp = ops.CropMirrorNormalize(
  70. device="gpu",
  71. dtype=output_dtype,
  72. output_layout='CHW',
  73. crop=(crop, crop),
  74. mean=mean,
  75. std=std,
  76. pad_output=pad_output)
  77. self.coin = ops.random.CoinFlip(probability=0.5)
  78. self.to_int64 = ops.Cast(dtype=types.DALIDataType.INT64, device="gpu")
  79. def define_graph(self):
  80. rng = self.coin()
  81. jpegs, labels = self.input(name="Reader")
  82. images = self.decode(jpegs)
  83. images = self.res(images)
  84. output = self.cmnp(images.gpu(), mirror=rng)
  85. return [output, self.to_int64(labels.gpu())]
  86. def __len__(self):
  87. return self.epoch_size("Reader")
  88. class HybridValPipe(Pipeline):
  89. def __init__(self,
  90. file_root,
  91. file_list,
  92. batch_size,
  93. resize_shorter,
  94. crop,
  95. interp,
  96. mean,
  97. std,
  98. device_id,
  99. shard_id=0,
  100. num_shards=1,
  101. random_shuffle=False,
  102. num_threads=4,
  103. seed=42,
  104. pad_output=False,
  105. output_dtype=types.FLOAT):
  106. super(HybridValPipe, self).__init__(
  107. batch_size, num_threads, device_id, seed=seed)
  108. self.input = ops.readers.File(
  109. file_root=file_root,
  110. file_list=file_list,
  111. shard_id=shard_id,
  112. num_shards=num_shards,
  113. random_shuffle=random_shuffle)
  114. self.decode = ops.decoders.Image(device="mixed")
  115. self.res = ops.Resize(
  116. device="gpu", resize_shorter=resize_shorter, interp_type=interp)
  117. self.cmnp = ops.CropMirrorNormalize(
  118. device="gpu",
  119. dtype=output_dtype,
  120. output_layout='CHW',
  121. crop=(crop, crop),
  122. mean=mean,
  123. std=std,
  124. pad_output=pad_output)
  125. self.to_int64 = ops.Cast(dtype=types.DALIDataType.INT64, device="gpu")
  126. def define_graph(self):
  127. jpegs, labels = self.input(name="Reader")
  128. images = self.decode(jpegs)
  129. images = self.res(images)
  130. output = self.cmnp(images)
  131. return [output, self.to_int64(labels.gpu())]
  132. def __len__(self):
  133. return self.epoch_size("Reader")
  134. def dali_dataloader(config, mode, device, seed=None):
  135. assert "gpu" in device, "gpu training is required for DALI"
  136. device_id = int(device.split(':')[1])
  137. config_dataloader = config[mode]
  138. seed = 42 if seed is None else seed
  139. ops = [
  140. list(x.keys())[0]
  141. for x in config_dataloader["dataset"]["transform_ops"]
  142. ]
  143. support_ops_train = [
  144. "DecodeImage", "NormalizeImage", "RandFlipImage", "RandCropImage"
  145. ]
  146. support_ops_eval = [
  147. "DecodeImage", "ResizeImage", "CropImage", "NormalizeImage"
  148. ]
  149. if mode.lower() == 'train':
  150. assert set(ops) == set(
  151. support_ops_train
  152. ), "The supported trasform_ops for train_dataset in dali is : {}".format(
  153. ",".join(support_ops_train))
  154. else:
  155. assert set(ops) == set(
  156. support_ops_eval
  157. ), "The supported trasform_ops for eval_dataset in dali is : {}".format(
  158. ",".join(support_ops_eval))
  159. normalize_ops = [
  160. op for op in config_dataloader["dataset"]["transform_ops"]
  161. if "NormalizeImage" in op
  162. ][0]["NormalizeImage"]
  163. channel_num = normalize_ops.get("channel_num", 3)
  164. output_dtype = types.FLOAT16 if normalize_ops.get("output_fp16",
  165. False) else types.FLOAT
  166. env = os.environ
  167. # assert float(env.get('FLAGS_fraction_of_gpu_memory_to_use', 0.92)) < 0.9, \
  168. # "Please leave enough GPU memory for DALI workspace, e.g., by setting" \
  169. # " `export FLAGS_fraction_of_gpu_memory_to_use=0.8`"
  170. gpu_num = paddle.distributed.get_world_size()
  171. batch_size = config_dataloader["sampler"]["batch_size"]
  172. file_root = config_dataloader["dataset"]["image_root"]
  173. file_list = config_dataloader["dataset"]["cls_label_path"]
  174. interp = 1 # settings.interpolation or 1 # default to linear
  175. interp_map = {
  176. 0: types.DALIInterpType.INTERP_NN, # cv2.INTER_NEAREST
  177. 1: types.DALIInterpType.INTERP_LINEAR, # cv2.INTER_LINEAR
  178. 2: types.DALIInterpType.INTERP_CUBIC, # cv2.INTER_CUBIC
  179. 3: types.DALIInterpType.
  180. INTERP_LANCZOS3, # XXX use LANCZOS3 for cv2.INTER_LANCZOS4
  181. }
  182. assert interp in interp_map, "interpolation method not supported by DALI"
  183. interp = interp_map[interp]
  184. pad_output = channel_num == 4
  185. transforms = {
  186. k: v
  187. for d in config_dataloader["dataset"]["transform_ops"]
  188. for k, v in d.items()
  189. }
  190. scale = transforms["NormalizeImage"].get("scale", 1.0 / 255)
  191. scale = eval(scale) if isinstance(scale, str) else scale
  192. mean = transforms["NormalizeImage"].get("mean", [0.485, 0.456, 0.406])
  193. std = transforms["NormalizeImage"].get("std", [0.229, 0.224, 0.225])
  194. mean = [v / scale for v in mean]
  195. std = [v / scale for v in std]
  196. sampler_name = config_dataloader["sampler"].get("name",
  197. "DistributedBatchSampler")
  198. assert sampler_name in ["DistributedBatchSampler", "BatchSampler"]
  199. if mode.lower() == "train":
  200. resize_shorter = 256
  201. crop = transforms["RandCropImage"]["size"]
  202. scale = transforms["RandCropImage"].get("scale", [0.08, 1.])
  203. ratio = transforms["RandCropImage"].get("ratio", [3.0 / 4, 4.0 / 3])
  204. min_area = scale[0]
  205. lower = ratio[0]
  206. upper = ratio[1]
  207. if 'PADDLE_TRAINER_ID' in env and 'PADDLE_TRAINERS_NUM' in env:
  208. shard_id = int(env['PADDLE_TRAINER_ID'])
  209. num_shards = int(env['PADDLE_TRAINERS_NUM'])
  210. device_id = int(env['FLAGS_selected_gpus'])
  211. pipe = HybridTrainPipe(
  212. file_root,
  213. file_list,
  214. batch_size,
  215. resize_shorter,
  216. crop,
  217. min_area,
  218. lower,
  219. upper,
  220. interp,
  221. mean,
  222. std,
  223. device_id,
  224. shard_id,
  225. num_shards,
  226. seed=seed + shard_id,
  227. pad_output=pad_output,
  228. output_dtype=output_dtype)
  229. pipe.build()
  230. pipelines = [pipe]
  231. # sample_per_shard = len(pipe) // num_shards
  232. else:
  233. pipe = HybridTrainPipe(
  234. file_root,
  235. file_list,
  236. batch_size,
  237. resize_shorter,
  238. crop,
  239. min_area,
  240. lower,
  241. upper,
  242. interp,
  243. mean,
  244. std,
  245. device_id=device_id,
  246. shard_id=0,
  247. num_shards=1,
  248. seed=seed,
  249. pad_output=pad_output,
  250. output_dtype=output_dtype)
  251. pipe.build()
  252. pipelines = [pipe]
  253. # sample_per_shard = len(pipelines[0])
  254. return DALIGenericIterator(
  255. pipelines, ['data', 'label'], reader_name='Reader')
  256. else:
  257. resize_shorter = transforms["ResizeImage"].get("resize_short", 256)
  258. crop = transforms["CropImage"]["size"]
  259. if 'PADDLE_TRAINER_ID' in env and 'PADDLE_TRAINERS_NUM' in env and sampler_name == "DistributedBatchSampler":
  260. shard_id = int(env['PADDLE_TRAINER_ID'])
  261. num_shards = int(env['PADDLE_TRAINERS_NUM'])
  262. device_id = int(env['FLAGS_selected_gpus'])
  263. pipe = HybridValPipe(
  264. file_root,
  265. file_list,
  266. batch_size,
  267. resize_shorter,
  268. crop,
  269. interp,
  270. mean,
  271. std,
  272. device_id=device_id,
  273. shard_id=shard_id,
  274. num_shards=num_shards,
  275. pad_output=pad_output,
  276. output_dtype=output_dtype)
  277. else:
  278. pipe = HybridValPipe(
  279. file_root,
  280. file_list,
  281. batch_size,
  282. resize_shorter,
  283. crop,
  284. interp,
  285. mean,
  286. std,
  287. device_id=device_id,
  288. pad_output=pad_output,
  289. output_dtype=output_dtype)
  290. pipe.build()
  291. return DALIGenericIterator(
  292. [pipe], ['data', 'label'], reader_name="Reader")