(self)
| 72 | assert_allclose(torch.cat(res_1), torch.cat(res_2), type_test=False) |
| 73 | |
| 74 | def test_time(self): |
| 75 | dataset = Dataset(data=self.datalist * 2, transform=self.transform) # contains data for 2 batches |
| 76 | dataloader = DataLoader(dataset=dataset, batch_size=2, num_workers=0) |
| 77 | |
| 78 | tbuffer = ThreadBuffer(dataloader) |
| 79 | |
| 80 | with PerfContext() as pc: |
| 81 | for _ in dataloader: |
| 82 | time.sleep(0.5) # each batch takes 0.8 s to generate on top of this time |
| 83 | |
| 84 | unbuffered_time = pc.total_time |
| 85 | |
| 86 | with PerfContext() as pc: |
| 87 | for _ in tbuffer: |
| 88 | time.sleep(0.5) # while "computation" is happening the next batch is being generated, saving 0.4 s |
| 89 | |
| 90 | buffered_time = pc.total_time |
| 91 | if sys.platform == "darwin": # skip macOS measure |
| 92 | print(f"darwin: Buffered time {buffered_time} vs unbuffered time {unbuffered_time}") |
| 93 | else: |
| 94 | self.assertTrue( |
| 95 | buffered_time < unbuffered_time, |
| 96 | f"Buffered time {buffered_time} should be less than unbuffered time {unbuffered_time}", |
| 97 | ) |
| 98 | |
| 99 | def test_dataloader_repeats(self): |
| 100 | dataset = Dataset(data=self.datalist, transform=self.transform) |
nothing calls this directly
no test coverage detected