(self)
| 77 | self.assertEqual(coordinator._coordinators, []) |
| 78 | |
| 79 | def testRNNInput(self): |
| 80 | workspace.ResetWorkspace() |
| 81 | model = model_helper.ModelHelper(name="rnn_test") |
| 82 | old_seq_id = data_workers.global_coordinator._fetcher_id_seq |
| 83 | coordinator = data_workers.init_data_input_workers( |
| 84 | model, |
| 85 | ["data1", "label1", "seq_lengths1"], |
| 86 | dummy_fetcher_rnn, |
| 87 | 32, |
| 88 | 2, |
| 89 | dont_rebatch=False, |
| 90 | batch_columns=[1, 1, 0], |
| 91 | ) |
| 92 | new_seq_id = data_workers.global_coordinator._fetcher_id_seq |
| 93 | self.assertEqual(new_seq_id, old_seq_id + 2) |
| 94 | |
| 95 | coordinator.start() |
| 96 | |
| 97 | workspace.RunNetOnce(model.param_init_net) |
| 98 | workspace.CreateNet(model.net) |
| 99 | |
| 100 | while coordinator._coordinators[0]._state._inputs < 100: |
| 101 | time.sleep(0.01) |
| 102 | |
| 103 | # Run a couple of rounds |
| 104 | workspace.RunNet(model.net.Proto().name) |
| 105 | workspace.RunNet(model.net.Proto().name) |
| 106 | |
| 107 | # Wait for the enqueue thread to get blocked |
| 108 | time.sleep(0.2) |
| 109 | |
| 110 | # We don't dequeue on caffe2 side (as we don't run the net) |
| 111 | # so the enqueue thread should be blocked. |
| 112 | # Let's now shutdown and see it succeeds. |
| 113 | self.assertTrue(coordinator.stop()) |
| 114 | |
| 115 | @unittest.skip("Test is flaky: https://github.com/pytorch/pytorch/issues/9064") |
| 116 | def testInputOrder(self): |
nothing calls this directly
no test coverage detected