Resets the iterator after the full epoch. DALI iterators do not support resetting before the end of the epoch and will ignore such request.
(self)
| 494 | return [p._get_checkpoint(iterator_data=iterator_data) for p in self._pipes] |
| 495 | |
| 496 | def reset(self): |
| 497 | """ |
| 498 | Resets the iterator after the full epoch. |
| 499 | DALI iterators do not support resetting before the end of the epoch |
| 500 | and will ignore such request. |
| 501 | """ |
| 502 | # in the case of the DROP policy the user who runs DALI, based on the iterator length, |
| 503 | # can assume there is no more data in the pipeline where there still is the last, |
| 504 | # incomplete batch, we need to extract from the pipeline and drop before rising |
| 505 | # StopIteration indicating the pipeline is depleted. Here we first check if that |
| 506 | # is the case, and if so we run the pipeline and drop the last batch |
| 507 | if self._last_batch_policy == LastBatchPolicy.DROP: |
| 508 | should_end = self._advance_and_check_drop_last(dry_run=True, end_iteration=False) |
| 509 | already_ended = self._size > 0 and self._counter >= self._size |
| 510 | if should_end and not already_ended: |
| 511 | self._get_outputs() |
| 512 | self._schedule_runs() |
| 513 | self._advance_and_check_drop_last(end_iteration=False) |
| 514 | |
| 515 | if self._counter >= self._size or self._size < 0: |
| 516 | if self._last_batch_policy == LastBatchPolicy.FILL and not self._last_batch_padded: |
| 517 | if self._reader_name: |
| 518 | # accurate way |
| 519 | # get the number of samples read in this epoch by each GPU |
| 520 | # self._counter had initial value of min(self._counter_per_gpu) so subtract |
| 521 | # this to get the actual value |
| 522 | self._counter -= min(self._counter_per_gpu) |
| 523 | self._counter_per_gpu = self._counter_per_gpu + self._counter |
| 524 | # check how much each GPU read ahead from next shard, as shards have different |
| 525 | # size each epoch GPU may read ahead or not |
| 526 | self._counter_per_gpu = self._counter_per_gpu - self._shard_sizes_per_gpu |
| 527 | # to make sure that in the next epoch we read the whole shard we need |
| 528 | # to set start value to the smallest one |
| 529 | self._counter = min(self._counter_per_gpu) |
| 530 | else: |
| 531 | # legacy way |
| 532 | self._counter = self._counter % self._size |
| 533 | else: |
| 534 | self._counter = 0 |
| 535 | # advance to the next shard |
| 536 | if self._reader_name: |
| 537 | if not self._is_stick_to_shard: |
| 538 | # move shards id for wrapped pipelines |
| 539 | self._shards_id = (self._shards_id + 1) % self._shards_num |
| 540 | # revaluate _size |
| 541 | if self._last_batch_policy == LastBatchPolicy.FILL and not self._last_batch_padded: |
| 542 | # move all shards ids GPU ahead |
| 543 | if not self._is_stick_to_shard: |
| 544 | self._shard_sizes_per_gpu = np.roll(self._shard_sizes_per_gpu, 1) |
| 545 | # check how many samples we need to reach from each shard in next epoch |
| 546 | # per each GPU taking into account already read |
| 547 | read_in_next_epoch = self._shard_sizes_per_gpu - self._counter_per_gpu |
| 548 | # get the maximum number of samples and round it up to full batch sizes |
| 549 | self._size = ( |
| 550 | math.ceil(max(read_in_next_epoch) / self.batch_size) * self.batch_size |
| 551 | ) |
| 552 | # in case some epoch is skipped because we have read ahead in this epoch so |
| 553 | # much that in the next one we done already |
no test coverage detected